github-actions[bot] commented on code in PR #69526: URL: https://github.com/apache/doris/pull/69526#discussion_r4369701009
########## 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: [P2] Preserve exact-path semantics in cleanup policies `protectJava(JAVA_DELETE, ...)` reaches this branch but serializes `descendants=true`, so runtime admission treats a literal brace directory as a glob. With LOCAL disabled, `INSERT INTO hdfs('file_path'='viewfs://cluster/data/literal{tag}/prefix_', 'format'='csv', 'delete_existing_files'='true', 'fs.defaultFS'='viewfs://cluster', 'fs.viewfs.mounttable.cluster.link./data'='hdfs://nn/base', 'fs.viewfs.mounttable.cluster.link./data/local'='file:///tmp') SELECT 1` passes binding and both static route checks, which select remote `/data`. The runtime rule then uses `/data/` as its glob prefix and rejects the unrelated `/data/local` mount, preventing cleanup and the insert. `BindSink` permits braces, and `DFSFileSystem.delete` passes the literal `Path` to `hfs.delete`; this operation never expands braces. Encode JAVA_DELETE with exact-path semantics and add a remote literal-brace-directory control beside a local mount. ########## fe/be-java-extensions/hadoop-deps/src/main/java/org/apache/hadoop/fs/FileSystem.java: ########## @@ -3642,13 +3653,670 @@ public static Class<? extends FileSystem> getFileSystemClass(String scheme, * @return the initialized filesystem. * @throws IOException problems loading or initializing the FileSystem */ + // DORIS-PATCH: TVF policy is checked against the frozen configuration used by the consumer. + public static int getDorisTvfGuardVersion() { + return 1; + } + + public static final class TvfGuard { + public static final String POLICY_KEY = "fs.doris.tvf.remote.paths"; + public static final String NATIVE_POLICY = "native"; + public static final String JAVA_READ_POLICY = "java-read"; + public static final String ROOT_KEY = "fs.doris.tvf.root"; + public static final String PATH_KEY = "fs.doris.tvf.path"; + public static final String CONTEXT_KEY = "fs.doris.tvf.context"; + private static final ThreadLocal<Context> LAST_NATIVE_CONTEXT = new ThreadLocal<>(); + + private TvfGuard() { + } + + public static final class Rule { + private final String filesystem; + private final String path; + private final boolean descendants; + + public Rule(String filesystem, String path, boolean descendants) { + this.filesystem = filesystem; + this.path = path; + this.descendants = descendants; + } + } + + public static String encode(List<Rule> rules) { + if (FileSystem.getDorisTvfGuardVersion() != 1) { + throw new IllegalStateException("TVF Hadoop runtime policy version mismatch"); + } + try { + java.io.ByteArrayOutputStream bytes = new java.io.ByteArrayOutputStream(); + java.io.DataOutputStream out = new java.io.DataOutputStream(bytes); + out.writeInt(1); + out.writeInt(rules.size()); + for (Rule rule : rules) { + out.writeUTF(rule.filesystem); + out.writeUTF(rule.path); + out.writeBoolean(rule.descendants); + } + out.flush(); + return java.util.Base64.getEncoder().encodeToString(bytes.toByteArray()); + } catch (IOException e) { + throw new IllegalArgumentException("Cannot encode TVF Hadoop route policy", e); + } + } + + /** Routes touched by Globber, without constructing a filesystem or listing data. */ + public static List<Rule> globRules(URI filesystem, String pattern) throws IOException { + List<Rule> rules = new ArrayList<>(); + rules.add(new Rule(filesystem.toString(), pattern, true)); + if (pattern.indexOf('\\') < 0) { Review Comment: [P2] Expand slash-containing braces before admitting read discovery With LOCAL disabled, `SELECT * FROM hdfs('uri'='viewfs://cluster/remote/{../local}/*.csv', 'fs.defaultFS'='viewfs://cluster', 'fs.viewfs.mounttable.cluster.link./remote'='hdfs://nn/base', 'fs.viewfs.mounttable.cluster.link./local'='file:///tmp', 'format'='csv', 'csv_schema'='id:int')` passes FE and frozen-runtime admission: the pattern has no backslash, so this early return leaves only `/remote` checked. Hadoop's [GlobExpander](https://github.com/apache/hadoop/blob/rel/release-3.4.2/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/GlobExpander.java#L60-L139) removes the singleton slash-containing braces, and [Globber](https://github.com/apache/hadoop/blob/rel/release-3.4.2/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Globber.java#L200-L218) normalizes the flattened pattern through `new Path` to `/local/*.csv` before listing. ViewFS consequently lists FE-local `/tmp` before the returned-file checks. Empty matches return the declared s chema; matched paths are rejected afterward, permitting local metadata probing outside the constrained LOCAL route. Apply expansion/normalization even without backslashes and cover this singleton-brace case. This differs from thread 4368483369's fixed escaped-component input. -- 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]
