github-actions[bot] commented on code in PR #69526:
URL: https://github.com/apache/doris/pull/69526#discussion_r4366875898
##########
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:
[P2] Load explicit mount-table resources before rejecting unknown routes
A remote mount defined in an explicitly selected FE Hadoop resource is
rejected before it can be normalized. For example, let operator-controlled
`remote.xml` define
`fs.viewfs.mounttable.cluster.link./data=hdfs://namenode:8020/base`. With LOCAL
disabled, `EXPLAIN INSERT INTO hdfs('file_path'='/data/prefix_',
'format'='csv', 'fs.defaultFS'='viewfs://cluster',
'hadoop.config.resources'='remote.xml') SELECT 1` reaches this branch through
`BindSink`'s raw-property check: the XML mount is absent from that map, so
admission throws. `HdfsProperties.initBackendConfigProperties` would load the
named resource into the backend map, and `TVFTableSink` already validates that
normalized map. Consequently an explicitly configured remote route now requires
enabling local access or duplicating operator resource contents in every
statement. Normalize the explicitly named resource before deciding the route is
unknown, while retaining the rejection before filesystem creation; add a remote
resource c
ontrol with `fs.defaultFS` present in the statement. This is separate from
intentionally rejecting mount tables inherited only on BE.
##########
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();
Review Comment:
[P1] Preserve Java property-key identity when checking FE cleanup
The cleanup path consumes Java strings, but this shared resolver truncates
mount keys as native C strings. With `enable_local_tvf=false`, `INSERT INTO
s3('file_path'='/data/sub/prefix_', 'format'='csv',
'delete_existing_files'='true', 'fs.defaultFS'='viewfs://cluster',
'fs.viewfs.mounttable.cluster.link./data\0ignored'='hdfs://namenode:8020/base',
'fs.viewfs.mounttable.cluster.linkFallback'='file:///tmp') SELECT 1` passes
both cleanup checks: SQL `\0` becomes an embedded NUL, and the validator sees a
remote `/data` mount. `HdfsProperties` preserves the full key and
`HdfsConfigBuilder.build` calls Java `Configuration.set(k,v)`, so Hadoop
instead mounts the distinct `/data<NUL>ignored` component. The actual
`/data/sub/` cleanup misses it and uses the local fallback, recursively
deleting `/tmp/data/sub` on the FE before BE execution, with global LOAD
privilege. Preserve the consumer's key semantics or reject ambiguous
NUL-bearing inputs before filesystem creation. This differs from t
he earlier direct ViewFS/LOCAL-adapter findings: the selected HDFS adapter is
checked, but the validator checks a different key than the Java filesystem uses.
##########
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;
Review Comment:
[P1] Gate the shipped ViewDistributedFileSystem provider
With `enable_local_tvf=false`, `INSERT INTO
hdfs('file_path'='/local/prefix_', 'format'='csv',
'fs.defaultFS'='hdfs://cluster',
'fs.hdfs.impl'='org.apache.hadoop.hdfs.ViewDistributedFileSystem',
'fs.viewfs.mounttable.cluster.link./local'='file:///tmp') SELECT 1` passes raw
and normalized admission because this provider is neither a direct local class
nor either recognized ViewFS class. The bundled Hadoop 3.4.2
[ViewDistributedFileSystem](https://github.com/apache/hadoop/blob/rel/release-3.4.2/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ViewDistributedFileSystem.java#L133-L173)
initializes `ViewFileSystemOverloadScheme` internally and delegates ordinary
`create` to it. The native FILE_HDFS writer therefore creates
`/tmp/prefix_<query>_0.csv` on the BE with global LOAD privilege. Recognize
this provider before returning, and cover raw/resource-derived binding and
cleanup. This is a distinct instance from the earlier direct `viewfs` finding:
this `hdfs
` provider returns before the new mount validator runs.
##########
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:
[P1] Resolve Hadoop's default authority before selecting the mount table
With the setting disabled, `INSERT INTO hdfs('file_path'='/data/prefix_',
'format'='csv', 'fs.defaultFS'='viewfs:///',
'fs.default.name'='viewfs://local',
'fs.viewfs.mounttable.default.link./data'='hdfs://namenode:8020/base',
'fs.viewfs.mounttable.local.link./data'='file:///tmp') SELECT 1` passes both
native-sink checks: this URI has no authority, so the resolver validates the
remote `default` table. On BE, `parse_properties` removes canonical
`fs.defaultFS` into the NameNode field but retains `fs.default.name` in Hadoop
configuration. That legacy alias supplies the effective default URI;
[FileSystem.newInstance](https://github.com/apache/doris/blob/071d4b6ec7c8c9a561fe752f7e42390bf38d1aa7/fe/be-java-extensions/hadoop-deps/src/main/java/org/apache/hadoop/fs/FileSystem.java#L621-L637)
fills the missing authority from it and opens `viewfs://local`. The writer
then creates `/tmp/prefix_<query>_0.csv` with global LOAD privilege. Resolve or
conservatively gate this authority completion
before admission. Unlike the earlier direct local-mount finding, the table
inspected here is remote; Hadoop selects a different table afterward.
--
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]
