OIiveirra commented on code in PR #69526:
URL: https://github.com/apache/doris/pull/69526#discussion_r4367401981
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java:
##########
@@ -137,6 +146,324 @@ private void getFileListFromBackend() throws
AnalysisException {
}
}
+ /** Apply LOCAL admission to the effective sink before planning or
filesystem side effects. */
+ public static void checkSinkEnabled(String tvfName, Map<String, String>
properties) {
+ if ("local".equalsIgnoreCase(tvfName) ||
isLocalWriter(properties.getOrDefault("writer_class", ""))) {
+ checkEnabled();
+ }
+ if ("hdfs".equalsIgnoreCase(tvfName)) {
+ if (!properties.containsKey("writer_class")) {
+ checkHdfsSinkEnabled(properties);
+ }
+ if
(Boolean.parseBoolean(properties.getOrDefault("delete_existing_files",
"false"))) {
+ checkHadoopFileSystemEnabled(
+
FileSystemUtil.extractParentDirectory(properties.getOrDefault("file_path",
"")), properties);
+ }
+ }
+ }
+
+ private static boolean isLocalWriter(String writer) {
+ // BE splits the complete identifier first, then passes each half
through JNI C strings.
+ int separator = writer.indexOf(':');
+ return separator > 0 &&
"java-writer".equals(nativeStringPrefix(writer.substring(0, separator)))
+ &&
"local-file".equals(nativeStringPrefix(writer.substring(separator + 1)));
+ }
+
+ /** Check the NameNode and provider used by the native FILE_HDFS writer. */
+ public static void checkHdfsSinkEnabled(Map<String, String> properties) {
+ checkHdfsSinkEnabled(properties.getOrDefault("file_path", ""),
properties);
+ }
+
+ public static void checkHdfsSinkEnabled(String outputPrefix, Map<String,
String> properties) {
+ String nameNode =
nativeStringPrefix(properties.getOrDefault("fs.defaultFS", "")).trim();
+ if ("default".equals(nameNode)) {
+ // libhdfs delegates this selector to the BE's defaults, which may
select local storage.
+ checkEnabled();
+ }
+ // libhdfs prefixes bare NameNodes with hdfs:// rather than using the
output path's scheme.
+ checkHadoopSchemeEnabled(nameNode.contains("://") ?
uriScheme(nameNode) : "hdfs", properties);
+ if (!Config.enable_local_tvf) {
+ try {
+ URI filesystem = new URI(nameNode.contains("://") ? nameNode :
"hdfs://" + nameNode);
Review Comment:
Fixed in 893c9371db7e84f16451ab8de8a6e81b94c933c4. Native admission removes
canonical fs.defaultFS into the NameNode projection, retains the legacy alias
in the Hadoop configuration, and completes a missing authority using that
effective default before selecting a ViewFS table. Java defaults are projected
independently. Conflicting C-string/config identities and unknown dynamic
delegation fail closed.
Tests include the exact default/local-table case and an explicitly remote
control. Actual bundled Hadoop contract checks confirm authority completion.
All 54 selected FE tests and Checkstyle passed with fresh XML and unchanged
inputs.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java:
##########
@@ -137,6 +146,324 @@ private void getFileListFromBackend() throws
AnalysisException {
}
}
+ /** Apply LOCAL admission to the effective sink before planning or
filesystem side effects. */
+ public static void checkSinkEnabled(String tvfName, Map<String, String>
properties) {
+ if ("local".equalsIgnoreCase(tvfName) ||
isLocalWriter(properties.getOrDefault("writer_class", ""))) {
+ checkEnabled();
+ }
+ if ("hdfs".equalsIgnoreCase(tvfName)) {
+ if (!properties.containsKey("writer_class")) {
+ checkHdfsSinkEnabled(properties);
+ }
+ if
(Boolean.parseBoolean(properties.getOrDefault("delete_existing_files",
"false"))) {
+ checkHadoopFileSystemEnabled(
+
FileSystemUtil.extractParentDirectory(properties.getOrDefault("file_path",
"")), properties);
+ }
+ }
+ }
+
+ private static boolean isLocalWriter(String writer) {
+ // BE splits the complete identifier first, then passes each half
through JNI C strings.
+ int separator = writer.indexOf(':');
+ return separator > 0 &&
"java-writer".equals(nativeStringPrefix(writer.substring(0, separator)))
+ &&
"local-file".equals(nativeStringPrefix(writer.substring(separator + 1)));
+ }
+
+ /** Check the NameNode and provider used by the native FILE_HDFS writer. */
+ public static void checkHdfsSinkEnabled(Map<String, String> properties) {
+ checkHdfsSinkEnabled(properties.getOrDefault("file_path", ""),
properties);
+ }
+
+ public static void checkHdfsSinkEnabled(String outputPrefix, Map<String,
String> properties) {
+ String nameNode =
nativeStringPrefix(properties.getOrDefault("fs.defaultFS", "")).trim();
+ if ("default".equals(nameNode)) {
+ // libhdfs delegates this selector to the BE's defaults, which may
select local storage.
+ checkEnabled();
+ }
+ // libhdfs prefixes bare NameNodes with hdfs:// rather than using the
output path's scheme.
+ checkHadoopSchemeEnabled(nameNode.contains("://") ?
uriScheme(nameNode) : "hdfs", properties);
+ if (!Config.enable_local_tvf) {
+ try {
+ URI filesystem = new URI(nameNode.contains("://") ? nameNode :
"hdfs://" + nameNode);
+ // BE appends a generated filename directly to this prefix,
then convert_path
+ // removes the NameNode. Resolve its parent, not an exact
mount-point filename.
+ String generated = outputPrefix + "doris-tvf-file";
+ String rawNameNode = properties.getOrDefault("fs.defaultFS",
"");
+ if (!rawNameNode.isEmpty() && generated.contains(rawNameNode))
{
+ generated = generated.substring(rawNameNode.length());
+ }
+ if (!generated.startsWith("/")) {
+ generated = "/" + generated;
+ }
+ String path =
hadoopPathUri(nativeStringPrefix(generated)).getPath();
+ path = path.substring(0, path.lastIndexOf('/') + 1);
+ if (path.length() > 1) {
+ path = path.substring(0, path.length() - 1);
+ }
+ checkViewFsEnabled(filesystem, path, properties, new
HashSet<>());
+ } catch (URISyntaxException e) {
+ // Invalid NameNodes remain subject to the existing storage
validation.
+ }
+ }
+ }
+
+ /** Check Hadoop's path/default-filesystem selection before FE-side
cleanup. */
+ public static void checkHadoopFileSystemEnabled(String path, Map<String,
String> properties) {
+ String scheme = uriScheme(path);
+ if (!scheme.isEmpty()) {
+ checkHadoopSchemeEnabled(scheme, properties);
+ checkViewFsPathEnabled(path, path, properties);
+ return;
+ }
+ String defaultFs = hadoopSelectorProperty("fs.defaultFS", properties);
+ String legacyDefaultFs = hadoopSelectorProperty("fs.default.name",
properties);
+ // Configuration.set synchronizes aliases; either value can win when
the map is applied.
+ // Do not assume that the canonical key overrides a local legacy
selector.
+ if (defaultFs != null) {
+ checkHadoopDefaultEnabled(defaultFs, path, properties);
+ }
+ if (legacyDefaultFs != null) {
+ checkHadoopDefaultEnabled(legacyDefaultFs, path, properties);
+ }
+ if (defaultFs == null && legacyDefaultFs == null) {
+ checkHadoopSchemeEnabled("file", properties);
+ }
+ }
+
+ private static void checkHadoopDefaultEnabled(String value, String path,
Map<String, String> properties) {
+ value = hadoopSelectorValue(value).trim();
+ // Hadoop retains the old "local" default-FS spelling.
+ String scheme = "local".equals(value) ? "file" : uriScheme(value);
+ checkHadoopSchemeEnabled(scheme.isEmpty() ? "hdfs" : scheme,
properties);
+ checkViewFsPathEnabled(scheme.isEmpty() ? "hdfs://" + value : value,
path, properties);
+ }
+
+ private static void checkViewFsPathEnabled(String location, String path,
Map<String, String> properties) {
+ if (!Config.enable_local_tvf) {
+ try {
+ URI filesystem = uriScheme(path).isEmpty() ? new URI(location)
: hadoopPathUri(path);
+ checkViewFsEnabled(filesystem, hadoopPathUri(path).getPath(),
properties, new HashSet<>());
+ } catch (URISyntaxException e) {
+ checkEnabled();
+ }
+ }
+ }
+
+ private static void checkHadoopSchemeEnabled(String scheme, Map<String,
String> properties) {
+ if ("file".equalsIgnoreCase(scheme)) {
+ checkEnabled();
+ }
+ String implementationKey = "fs." + scheme + ".impl";
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ // The native Hadoop configuration boundary consumes C strings for
keys and values.
+ if
(implementationKey.equals(nativeStringPrefix(entry.getKey()).trim())) {
+ String implementation =
hadoopSelectorValue(nativeStringPrefix(entry.getValue())).trim();
+ if
("org.apache.hadoop.fs.LocalFileSystem".equals(implementation)
+ ||
"org.apache.hadoop.fs.RawLocalFileSystem".equals(implementation)) {
+ checkEnabled();
+ }
+ }
+ }
+ }
+
+ private static void checkViewFsEnabled(URI filesystem, String path,
Map<String, String> properties,
+ Set<String> activeTables) {
+ if (Config.enable_local_tvf) {
+ return;
+ }
+ String scheme = filesystem.getScheme();
+ checkHadoopSchemeEnabled(scheme, properties);
+ String implementation = hadoopSelectorProperty("fs." + scheme +
".impl", properties);
+ if (implementation != null
+ &&
"org.apache.hadoop.fs.viewfs.ViewFileSystemOverloadScheme".equals(implementation.trim()))
{
+ // OverloadScheme can reload runtime mount tables and overrides
provider selection.
+ checkEnabled();
+ }
+ boolean viewFs = isViewFs(scheme, implementation);
+ if (!viewFs) {
+ return;
+ }
+ String table = filesystem.getAuthority();
+ String ignorePort =
hadoopSelectorProperty("fs.viewfs.ignore.port.in.mount.table.name", properties);
+ if (ignorePort != null &&
Boolean.parseBoolean(hadoopSelectorValue(ignorePort).trim())
+ && table != null && filesystem.getPort() != -1) {
+ table = filesystem.getHost();
+ if (table == null) {
+ checkEnabled();
+ }
+ }
+ if (table == null || table.isEmpty()) {
+ table =
hadoopSelectorProperty("fs.viewfs.mounttable.default.name.key", properties);
+ table = table == null ? "default" : hadoopSelectorValue(table);
+ }
+ // Nested ViewFS cycles or arbitrarily deep delegation cannot
establish remote storage.
+ if (activeTables.size() >= 32 || !activeTables.add(table)) {
+ checkEnabled();
+ }
+ try {
+ String prefix = "fs.viewfs.mounttable." + table + ".";
+ String source = "";
+ List<String> selectedKeys = new ArrayList<>();
+ List<String> mountSources = new ArrayList<>();
+ String fallback = null;
+ String rootMerge = null;
+ for (String rawKey : properties.keySet()) {
+ String key = nativeStringPrefix(rawKey).trim();
+ if (!key.startsWith(prefix)) {
+ continue;
+ }
+ String entry = key.substring(prefix.length());
+ if (entry.equals("linkFallback")) {
+ fallback = key;
+ continue;
+ }
+ if (entry.equals("linkMergeSlash")) {
+ rootMerge = key;
+ continue;
+ }
+ if (entry.startsWith("linkRegex")) {
+ // Regex interception rewrites runtime paths before
ordinary mount resolution.
+ checkEnabled();
+ }
+ int slash = entry.indexOf('/');
+ if (slash < 0 || !(entry.startsWith("link.") ||
entry.startsWith("linkMerge.")
+ || entry.startsWith("linkNfly."))) {
+ continue;
+ }
+ String candidate = entry.substring(slash);
+ // InodeTree splits source components and discards trailing
empty components.
+ while (candidate.length() > 1 && candidate.endsWith("/")) {
+ candidate = candidate.substring(0, candidate.length() - 1);
+ }
+ mountSources.add(candidate);
+ if (!(path.equals(candidate) || path.startsWith(candidate +
"/"))) {
+ continue;
+ }
+ if (candidate.length() > source.length()) {
+ source = candidate;
+ selectedKeys.clear();
+ }
+ if (candidate.equals(source)) {
+ selectedKeys.add(key);
+ }
+ }
+ if (rootMerge != null) {
+ selectedKeys.add(rootMerge);
+ source = "";
+ } else if (selectedKeys.isEmpty() && fallback != null) {
+ selectedKeys.add(fallback);
+ }
+ if (selectedKeys.isEmpty()) {
+ // No explicit target can prove what a BE-inherited mount
table will select.
+ checkEnabled();
Review Comment:
Fixed in 893c9371db7e84f16451ab8de8a6e81b94c933c4. Sink binding now
normalizes explicitly selected Hadoop resources through the existing storage
implementation before route admission. The final TVF sink and cleanup still
validate the actual backend properties before filesystem creation; no Hadoop
filesystem is constructed by the admission helper.
A real temporary XML resource defines only the remote ViewFS mount while the
statement supplies fs.defaultFS. INSERT/EXPLAIN and direct sink controls remain
admitted with LOCAL disabled; local/unknown resource routes remain rejected.
The standard FE set passed 54 tests, and independent actual-Hadoop/resource
normalization checks passed. No real cleanup was performed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
