OIiveirra commented on code in PR #69526: URL: https://github.com/apache/doris/pull/69526#discussion_r4369910954
########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/HadoopTvfAdmission.java: ########## @@ -0,0 +1,312 @@ +// 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.filesystem.spi.GlobPattern; +import org.apache.doris.nereids.exceptions.AnalysisException; + +import org.apache.hadoop.fs.FileSystem.TvfGuard; + +import java.io.IOException; +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.HashSet; +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 final Access access; + private final TvfGuard.Route routes; + 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; + } + if (nativeConsumer && isRoutingKey(asciiCaseFold(nativeStringPrefix(entry.getKey()).trim()))) { + checkNativeCharacters(access, entry.getKey()); + checkNativeCharacters(access, entry.getValue()); + } + 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(); + } + } + routes = new TvfGuard.Route(properties, access == Access.HADOOP_READ); + } + + /** 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) { + check(access, path, normalized, true); + } + + /** A declared CSV schema skips the schema RPC; listing and data admission still apply. */ + public static void check(Access access, String path, Map<String, String> normalized, + boolean needsSchemaDiscovery) { + 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); + javaConsumer.checkRoute(javaPath, javaPath.getPath(), new HashSet<>()); + if (access == Access.HADOOP_READ && hasListingGlob(path)) { + URI listingPath = hadoopPathUri(GlobPattern.expandNumericRanges(path)); + javaConsumer.routes.checkGlobRoute(listingPath, listingPath.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. + String nameNode = nativeReadNameNode(path, 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. + if (needsSchemaDiscovery) { + nativeConsumer.checkNativeRead(nativeNameNode(access, normalized), path); + } + } + } + } catch (IOException | URISyntaxException | IllegalArgumentException e) { + new HadoopTvfAdmission(access, normalized, false).reject(); + } + } + + private static String nativeReadNameNode(String path, Map<String, String> normalized) { + String nameNode = LocationPath.of(path).getFsIdentifier(); + if (nameNode.isEmpty()) { + nameNode = nativeNameNode(Access.HADOOP_READ, normalized); + } + return nameNode; + } + + /** Reserved internal policy must not be selected by SQL/resource property aliases. */ + public static void removeRuntimePolicy(Map<String, String> properties) { + properties.keySet().removeIf(key -> asciiCaseFold(nativeStringPrefix(key)).trim() + .startsWith("fs.doris.tvf.")); + } + + private static boolean hasListingGlob(String path) { + return path.indexOf('*') >= 0 || path.indexOf('?') >= 0 + || path.indexOf('[') >= 0 || path.indexOf('{') >= 0; + } + + public static Map<String, String> protectJava(Access access, String path, Map<String, String> properties) { + Map<String, String> guarded = new HashMap<>(properties); + removeRuntimePolicy(guarded); + if (access == Access.HADOOP_READ || !Config.enable_local_tvf) { + try { + URI uri = hadoopPathUri(path); + java.util.List<TvfGuard.Rule> rules; + if (access == Access.HADOOP_READ && hasListingGlob(path)) { + URI listingPath = hadoopPathUri(GlobPattern.expandNumericRanges(path)); + rules = TvfGuard.globRules(listingPath, listingPath.getPath()); + } else { + rules = java.util.Collections.singletonList(new TvfGuard.Rule(uri.toString(), uri.getPath(), true)); Review Comment: Fixed in a9c0597cbbdaa85e1d79cb710ba83599456e79c8. JAVA_DELETE now serializes an exact-path rule, matching DFSFileSystem.delete and ViewFS resolve(true), so literal brace directories no longer inspect unrelated descendant mounts. Actual selected local and nested-local targets remain rejected. Focused controls cover literal remote cleanup beside an unrelated local mount and the selected-local negative cases. The fix is synchronized across the three PRs. Validation: final standard JDK17 FE selected tests: 106 passed, zero failures/errors/skips with fresh XML and unchanged source. Standard Clang21/ASAN HdfsMgrTest: 12 passed; native source/test bytes remain unchanged. Actual-file Checkstyle passed. Full community CI is not claimed. -- 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]
