OIiveirra commented on code in PR #69528: URL: https://github.com/apache/doris/pull/69528#discussion_r4367889998
########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/HadoopTvfAdmission.java: ########## @@ -0,0 +1,502 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.tablefunction; + +import org.apache.doris.common.Config; +import org.apache.doris.common.util.LocationPath; +import org.apache.doris.nereids.exceptions.AnalysisException; + +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * TVF-only admission for Hadoop consumers. Input properties have already been normalized by storage. + * This module neither loads Hadoop resources nor constructs a filesystem. Java strings and native + * C strings are projected separately before filesystem selection and mount-table interpretation. + */ +public final class HadoopTvfAdmission { + public enum Access { + NATIVE_WRITE, + JAVA_DELETE, + HADOOP_READ + } + + private enum Provider { + REMOTE, + LOCAL, + MOUNT_TABLE, + DYNAMIC, + UNKNOWN + } + + private static final Set<String> REMOTE_SCHEMES = new HashSet<>(Arrays.asList( + "hdfs", "s3a", "s3n", "oss", "ofs", "jfs", "abfs", "abfss", "wasb", "wasbs", "gs", "cosn")); + private static final Set<String> REMOTE_IMPLEMENTATIONS = new HashSet<>(Arrays.asList( + "org.apache.hadoop.hdfs.DistributedFileSystem", + "org.apache.hadoop.fs.s3a.S3AFileSystem", + "org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem", + "org.apache.hadoop.fs.azurebfs.SecureAzureBlobFileSystem", + "org.apache.hadoop.fs.azure.NativeAzureFileSystem", + "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem", + "com.aliyun.jindodata.oss.JindoOssFileSystem", + "org.apache.hadoop.fs.obs.OBSFileSystem", + "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")); + + private final Access access; + private final Map<String, String> properties = new HashMap<>(); + + private HadoopTvfAdmission(Access access, Map<String, String> normalized, boolean nativeConsumer) { + this.access = access; + Set<String> nativeRoutingKeys = new HashSet<>(); + for (Map.Entry<String, String> entry : normalized.entrySet()) { + // parse_properties / generateHdfsParam consumes the canonical NameNode before JNI. + if (nativeConsumer && "fs.defaultFS".equalsIgnoreCase(entry.getKey())) { + continue; + } + String key = (nativeConsumer ? nativeStringPrefix(entry.getKey()) : entry.getKey()).trim(); + // Native writes first pass through StringCaseMap, before JNI truncates strings. + // Its comparator uses strncasecmp followed by the full UTF-8 byte length. + if (nativeConsumer && access == Access.NATIVE_WRITE && isRoutingKey(asciiCaseFold(key))) { + String identity = asciiCaseFold(nativeStringPrefix(entry.getKey())) + "\0" + + entry.getKey().getBytes(StandardCharsets.UTF_8).length; + if (!nativeRoutingKeys.add(identity)) { + reject(); + } + } + String value = nativeConsumer ? nativeStringPrefix(entry.getValue()) : entry.getValue(); + String previous = properties.put(key, value); + if (previous != null && !previous.equals(value) && isRoutingKey(key)) { + reject(); + } + } + } + + /** Reject local/unknown Hadoop TVF routes before listing, deletion, schema discovery or data IO. */ + public static void check(Access access, String path, Map<String, String> normalized) { + if (access != Access.HADOOP_READ && Config.enable_local_tvf) { + return; + } + try { + if (access == Access.NATIVE_WRITE) { + String nameNode = nativeNameNode(access, normalized); + URI output = nativeFileUri(access, path + "doris-tvf-file", nameNode); + String parent = output.getPath(); + parent = parent.substring(0, parent.lastIndexOf('/') + 1); + if (parent.length() > 1) { + parent = parent.substring(0, parent.length() - 1); + } + new HadoopTvfAdmission(access, normalized, true).checkNative(nameNode, parent); + } else { + HadoopTvfAdmission javaConsumer = new HadoopTvfAdmission(access, normalized, false); + // Project the Java consumer independently from native convert_path/JNI. + URI javaPath = hadoopPathUri(path.trim()); + javaConsumer.checkRoute(javaPath, javaPath.getPath(), new HashSet<>()); + if (access == Access.HADOOP_READ) { + // FileQueryScanNode sends this filesystem identifier in range.fs_name. BE + // falls back to hdfs_params.fs_name only when that range field is empty. + URI fileUri = LocationPath.of(path).getPath().toUri(); + String nameNode = fileUri.getScheme() + "://" + fileUri.getAuthority(); + if (nameNode.isEmpty()) { + nameNode = nativeNameNode(access, normalized); + } + HadoopTvfAdmission nativeConsumer = new HadoopTvfAdmission(access, normalized, true); + nativeConsumer.checkNativeRead(nameNode, path); + // Schema RPCs have no range.fs_name; they use THdfsParams.fs_name instead. + nativeConsumer.checkNativeRead(nativeNameNode(access, normalized), path); + } + } + } catch (URISyntaxException e) { + new HadoopTvfAdmission(access, normalized, false).reject(); + } + } + + private static String nativeNameNode(Access access, Map<String, String> normalized) { + String result = null; + for (Map.Entry<String, String> entry : normalized.entrySet()) { + if ("fs.defaultFS".equalsIgnoreCase(entry.getKey())) { + if (result != null && !result.equals(entry.getValue())) { + reject(access); + } + result = entry.getValue(); + } + } + return result == null ? "" : result; + } + + private void checkNative(String nameNode, String path) throws URISyntaxException { + nameNode = selector(nativeStringPrefix(nameNode)).trim(); + if (nameNode.isEmpty() && access == Access.NATIVE_WRITE) { + // BE passes fs_name.c_str(), never NULL. libhdfs prefixes this empty string + // with hdfs://, whose URI construction fails before any filesystem is selected. + // Preserve existing analysis; execution reports the invalid NameNode as before. + return; + } + if ("default".equals(nameNode) || nameNode.isEmpty()) { + reject(); + } + checkRoute(new URI(nameNode.contains("://") ? nameNode : "hdfs://" + nameNode), + path, new HashSet<>()); + } + + private void checkNativeRead(String nameNode, String path) throws URISyntaxException { + checkNative(nameNode, nativeFileUri(access, path, nameNode).getPath()); + } + + private List<URI> defaults() throws URISyntaxException { + List<URI> defaults = new ArrayList<>(); + for (String key : Arrays.asList("fs.defaultFS", "fs.default.name")) { + String value = property(key); + if (value != null) { + value = selector(value).trim(); + // FileSystem.fixName treats slash-free host[:port] strings as HDFS defaults. + defaults.add(new URI("local".equals(value) ? "file:///" + : (value.indexOf('/') < 0 ? "hdfs://" + value : value))); + } + } + if (defaults.isEmpty()) { + defaults.add(new URI("file:///")); + } + return defaults; + } + + private void checkRoute(URI filesystem, String path, Set<String> activeTables) throws URISyntaxException { + if (filesystem.getScheme() == null) { + for (URI defaultFs : defaults()) { + checkViewFsEnabled(defaultFs, path, activeTables); + } + return; + } + if (filesystem.getAuthority() == null) { + boolean completed = false; + for (URI defaultFs : defaults()) { + if (filesystem.getScheme().equals(defaultFs.getScheme()) && defaultFs.getAuthority() != null) { + checkViewFsEnabled(defaultFs, path, activeTables); + completed = true; + } + } + if (completed) { + return; + } + } + checkViewFsEnabled(filesystem, path, activeTables); + } + + private Provider provider(String scheme) { + if ("file".equalsIgnoreCase(scheme)) { + return Provider.LOCAL; + } + String implementation = property("fs." + scheme + ".impl"); + if (implementation != null) { + implementation = selector(implementation).trim(); + if ("org.apache.hadoop.fs.LocalFileSystem".equals(implementation) + || "org.apache.hadoop.fs.RawLocalFileSystem".equals(implementation)) { + return Provider.LOCAL; + } + if ("org.apache.hadoop.fs.viewfs.ViewFileSystem".equals(implementation)) { + return Provider.MOUNT_TABLE; + } + if ("org.apache.hadoop.fs.viewfs.ViewFileSystemOverloadScheme".equals(implementation) + || "org.apache.hadoop.hdfs.ViewDistributedFileSystem".equals(implementation)) { + return Provider.DYNAMIC; + } + return REMOTE_IMPLEMENTATIONS.contains(implementation) ? Provider.REMOTE : Provider.UNKNOWN; + } + if ("viewfs".equals(scheme)) { + return Provider.MOUNT_TABLE; + } + return REMOTE_SCHEMES.contains(scheme) ? Provider.REMOTE : Provider.UNKNOWN; + } + + private void checkGlobTargets(String key, Set<String> activeTables) throws URISyntaxException { + for (String target : selector(property(key)).split(",", -1)) { + if (target.isEmpty()) { + reject(); + } + URI uri = new URI(target); + // Listing can touch any descendant of a selected mount. Reuse the same route + // interpreter for nested tables instead of assuming their root is representative. + String path = uri.getPath() == null ? "/" : uri.getPath(); + checkRoute(uri, (path.endsWith("/") ? path : path + "/") + "*", activeTables); + } + } + + private void checkViewFsEnabled(URI filesystem, String path, Set<String> activeTables) { + String scheme = filesystem.getScheme(); + Provider provider = provider(scheme); + if (provider == Provider.LOCAL || provider == Provider.UNKNOWN || provider == Provider.DYNAMIC) { + reject(); + } + if (provider != Provider.MOUNT_TABLE) { + return; + } + if (!path.startsWith("/")) { + // Java resolves relative paths against the selected filesystem's working directory. + // Its user/home configuration is not a proven absolute mount-table path here. + reject(); + } + String table = filesystem.getAuthority(); + String ignorePort = property("fs.viewfs.ignore.port.in.mount.table.name"); + if (ignorePort != null && Boolean.parseBoolean(selector(ignorePort).trim()) + && table != null && filesystem.getPort() != -1) { + table = filesystem.getHost(); + if (table == null) { + reject(); + } + } + if (table == null || table.isEmpty()) { + table = property("fs.viewfs.mounttable.default.name.key"); + table = table == null ? "default" : selector(table); + } + // Nested ViewFS cycles or arbitrarily deep delegation cannot establish remote storage. + if (activeTables.size() >= 32 || !activeTables.add(table)) { + reject(); + } + 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 key : properties.keySet()) { + 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. + reject(); + } + 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 (access == Access.HADOOP_READ && hasGlob(path)) { + String literal = path.substring(0, firstGlob(path)); + int slash = literal.lastIndexOf('/'); + literal = slash < 0 ? "/" : literal.substring(0, slash + 1); + for (String key : properties.keySet()) { + if (!key.startsWith(prefix)) { + continue; + } + String entry = key.substring(prefix.length()); + int sourceStart = entry.indexOf('/'); + if (sourceStart >= 0 && (entry.startsWith("link.") || entry.startsWith("linkMerge.") + || entry.startsWith("linkNfly."))) { + String mount = entry.substring(sourceStart); + if (mount.startsWith(literal) || literal.startsWith(mount + "/")) { + checkGlobTargets(key, activeTables); + } + } + } + } + if (access == Access.HADOOP_READ && !hasGlob(path)) { + String subtree = path.endsWith("/") ? path : path + "/"; + for (String key : properties.keySet()) { + if (key.startsWith(prefix)) { + String entry = key.substring(prefix.length()); + int sourceStart = entry.indexOf('/'); + if (sourceStart >= 0 && entry.substring(sourceStart).startsWith(subtree)) { + checkGlobTargets(key, activeTables); + } + } + } + } + if (rootMerge != null) { + selectedKeys.add(rootMerge); + source = ""; + } else if (selectedKeys.isEmpty() && fallback != null) { + selectedKeys.add(fallback); + } + if (selectedKeys.isEmpty()) { + // Missing inline fallback does not prove absence of one in default resources. + // Keep rejecting unproven routes, including globs that cross mount names. + reject(); + } + for (String key : selectedKeys) { + String targets = selector(property(key)); + for (String target : targets.split(",", -1)) { Review Comment: Confirmed Hadoop SINGLE, SINGLE_FALLBACK and MERGE_SLASH each consume one URI. Admission now preserves their complete comma-containing paths in both selected and glob-target checks; only MERGE/NFLY use comma lists. Updated controls cover read/write/cleanup/glob and retain rejection of local/malformed multi-target routes and truly empty single targets. Follow-up commit: aad01e0da033db903a8cd088cd733ee0623c8709. 73 selected FE tests passed with fresh XML, no failures/errors/skips and unchanged inputs; targeted Checkstyle passed. ########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java: ########## @@ -135,6 +143,38 @@ private void getFileListFromBackend() throws AnalysisException { } } + /** Apply LOCAL admission to sinks that select the built-in local JNI writer as well. */ + public static void checkSinkEnabled(String tvfName, Map<String, String> properties) { + String writerClass = properties.getOrDefault("writer_class", ""); + // JNI consumes a C string, and ScannerLoader accepts slash-separated class names. + int nul = writerClass.indexOf('\0'); + if (nul >= 0) { + writerClass = writerClass.substring(0, nul); + } + if ("local".equalsIgnoreCase(tvfName) Review Comment: Reopening for the retained inherited-provider variant. The explicit local provider/NameNode examples are fixed, but the BE-only core-site LocalFileSystem case remains valid: the FE map does not represent that effective runtime Configuration. Earlier replies overstated coverage of inherited configuration. A forced provider substitution would change administrator configuration semantics and still would not close ViewFS overlays, so it is not included as an unproven minimal fix. The PR description now states this remaining P1; complete runtime enforcement is not claimed. Follow-up commit: aad01e0da033db903a8cd088cd733ee0623c8709. 73 selected FE tests passed with fresh XML, no failures/errors/skips and unchanged inputs; targeted Checkstyle passed. ########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java: ########## @@ -135,6 +150,332 @@ 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( + BrokerUtil.extractParentDirectory(properties.getOrDefault("file_path", "")), properties); + } + } + } + + private static boolean isLocalWriter(String writer) { + // ScannerLoader accepts slashes, and JNI passes the class name as a C string. + return "org.apache.doris.writer.LocalFileJniWriter".equals(nativeStringPrefix(writer).replace('/', '.')); + } + + /** 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) { Review Comment: Reopening for the retained implicit overloaded-provider variant. Explicit ViewFS/local routes are rejected, but a provider inherited from FE/BE default resources can preserve an hdfs:// path while delegating locally, so repeated URI/map admission is insufficient. Loading only FE defaults cannot establish BE behavior. This remains unresolved pending validation at the actual consumer/configuration boundary. No blanket rejection of valid remote ViewFS or generic Hadoop/catalog behavior change is included, and the PR description now records this limitation. Follow-up commit: aad01e0da033db903a8cd088cd733ee0623c8709. 73 selected FE tests passed with fresh XML, no failures/errors/skips and unchanged inputs; targeted Checkstyle passed. -- 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]
