OIiveirra commented on code in PR #69526:
URL: https://github.com/apache/doris/pull/69526#discussion_r4367401558
##########
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:
Fixed in 893c9371db7e84f16451ab8de8a6e81b94c933c4. Hadoop TVF admission now
has separate Java and native property/path projections. FE cleanup preserves
Java property-key identity, including embedded NULs; native keys follow JNI
C-string semantics. Checks consume normalized storage properties before any
filesystem factory or cleanup call.
The complete consumer audit also corrected native convert_path ordering and
ambiguous non-prefix NameNode matches. Tests cover the exact NUL-key/fallback
case, INSERT/EXPLAIN/direct binding and rejection before factory/executor
interactions. Actual bundled Hadoop classes independently confirm the consumer
identities. All 54 selected FE tests passed with fresh XML, zero
failures/errors/skips and unchanged inputs; Checkstyle passed. No real Hadoop
deletion was executed.
##########
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:
Fixed in 893c9371db7e84f16451ab8de8a6e81b94c933c4. The shared admission
module recognizes ViewDistributedFileSystem and ViewFileSystemOverloadScheme as
runtime delegation, rather than direct remote providers. Disabled local access
rejects their unproven routes before native binding or FE cleanup; explicit
startup opt-in retains existing write/cleanup behavior. Hadoop TVF reads
require a proven remote route regardless of LOCAL opt-in.
Raw/resource-normalized bindings, cleanup and remote controls are covered by
the 54 passing selected FE tests. Actual bundled Hadoop recording providers
confirm VDFS delegation without connecting to a real NameNode or deleting data.
--
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]
