github-actions[bot] commented on code in PR #69528:
URL: https://github.com/apache/doris/pull/69528#discussion_r4369693987
##########
fe/be-java-extensions/hadoop-deps/src/main/java/org/apache/hadoop/fs/FileSystem.java:
##########
@@ -3622,13 +3633,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) {
+ return rules;
+ }
+ for (String expanded : GlobExpander.expand(pattern)) {
+ Path flattened = new Path(expanded.isEmpty() ? Path.CUR_DIR :
expanded);
+ String path = flattened.toUri().getPath();
+ Path candidate = new Path(flattened.isAbsolute() ? "/" : Path.CUR_DIR);
+ for (String component : path.split("/")) {
+ if (component.isEmpty()) {
+ continue;
+ }
+ GlobFilter filter = new GlobFilter(component);
+ if (filter.hasPattern()) {
+ // Globber lists this parent before it knows the matching child
names. Keep
+ // prior wildcard components as a conservative subtree route,
including when
+ // a later escaped '..' moves the final candidate outside that
subtree.
+ rules.add(new Rule(filesystem.toString(), new Path(candidate,
"*").toUri().getPath(), true));
+ }
+ component = component.replaceAll("\\\\(.)", "$1");
+ candidate = new Path(candidate, component);
+ }
+ rules.add(new Rule(filesystem.toString(), candidate.toUri().getPath(),
true));
+ }
+ return rules;
+ }
+
+ private static IOException rejected() {
+ return new IOException("Hadoop TVF requires a proven remote filesystem
route. "
+ + "Use local() with LOCAL TVF privileges and secure-path
restrictions.");
+ }
+
+ private static Configuration freeze(Configuration conf) throws IOException
{
+ // Service providers (notably HDFS) register their default/site
resources on first load.
+ // Load them before copying raw values, as an ordinary FileSystem lookup
does.
+ if (!FILE_SYSTEMS_LOADED) {
+ loadFileSystems();
+ }
+ if (conf instanceof FrozenConfiguration) {
+ return conf;
+ }
+ return new FrozenConfiguration(conf);
+ }
+
+ private static Configuration prepare(Configuration conf) throws
IOException {
+ String policy = conf.getRaw(POLICY_KEY);
+ if (policy == null || policy.isEmpty()) {
+ return conf;
+ }
+ Configuration frozen = freeze(conf);
+ Context context = ((FrozenConfiguration) frozen).context;
+ if (NATIVE_POLICY.equals(policy)) {
+ context.checkNative(frozen.getRaw(ROOT_KEY), frozen.getRaw(PATH_KEY));
+ } else if (JAVA_READ_POLICY.equals(policy)) {
+ context.checkJava(new Path(frozen.getRaw(PATH_KEY)));
+ } else {
+ context.checkRules(policy);
+ }
+ return frozen;
+ }
+
+ public static void check(URI ignored, Configuration conf) throws
IOException {
+ prepare(conf);
+ }
+
+ public static void clearNativeContext() {
+ LAST_NATIVE_CONTEXT.remove();
+ }
+
+ public static Context takeNativeContext() {
+ Context result = LAST_NATIVE_CONTEXT.get();
+ LAST_NATIVE_CONTEXT.remove();
+ return result;
+ }
+
+ public static Configuration prepareJavaRead(Configuration conf, Path path,
String marker) throws IOException {
+ if (marker == null || marker.isEmpty()) {
+ return conf;
+ }
+ if (!NATIVE_POLICY.equals(marker)) {
+ throw rejected();
+ }
+ Configuration request = new Configuration(conf);
+ request.set(POLICY_KEY, JAVA_READ_POLICY);
+ request.set(ROOT_KEY, path.toUri().toString());
+ request.set(PATH_KEY, path.toString());
+ return prepare(request);
+ }
+
+ public static void checkJavaRead(FileSystem fs, Path path, String marker)
throws IOException {
+ if (marker == null || marker.isEmpty()) {
+ return;
+ }
+ if (!NATIVE_POLICY.equals(marker)) {
+ throw rejected();
+ }
+ // The marker is request-owned, never inferred from an unmarked cached
filesystem.
+ Configuration conf = fs.getConf();
+ if (JAVA_READ_POLICY.equals(conf.getRaw(POLICY_KEY))) {
+ // S3A derives a Configuration for bucket options. Its copy preserves
our internal
+ // marker and raw snapshot values, but not the private Configuration
subclass.
+ ((FrozenConfiguration) freeze(conf)).context.checkJava(path);
+ } else {
+ throw rejected();
+ }
+ }
+
+ private static final class FrozenConfiguration extends Configuration {
+ private final Context context;
+
+ private FrozenConfiguration(Configuration source) throws IOException {
+ super(false);
+ setClassLoader(source.getClassLoader());
+ java.util.SortedMap<String, String> values = new java.util.TreeMap<>();
+ for (Map.Entry<String, String> entry : source) {
+ String key = entry.getKey();
+ String value = source.getRaw(key);
+ set(key, value);
+ if (!PATH_KEY.equals(key) && !ROOT_KEY.equals(key) &&
!CONTEXT_KEY.equals(key)) {
+ values.put(key, value);
+ }
+ }
+ // ViewFS's private InnerCache folds authority case (A/a). Guarded
children must
+ // instead use FileSystem's exact-authority/profile cache below.
+ set("fs.viewfs.enable.inner.cache", "false");
Review Comment:
[P2] Release ViewFS child clients when disabling its inner cache
For a remote mount `/data -> s3a://bucket/base`, repeated
`hdfs('uri'='viewfs://cluster/data/fileN.csv',
'fs.defaultFS'='viewfs://cluster', 'format'='csv', 'csv_schema'='id:int', ...)`
calls with unchanged credentials now create a distinct S3A client for each
path: `protectJava()` includes the listing URI/path in the policy, and
`Cache.Key` distinguishes those policies. This override routes children into
Hadoop's global cache, but `DFSFileSystem` only disables HDFS caching; [the
pinned ViewFS close
path](https://github.com/apache/doris-thirdparty/blob/hadoop-3.4.2.2-for-doris/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java#L1828-L1866)
closes children only when their scheme's cache is disabled. Default-cached S3A
children therefore survive closing/collecting the outer filesystem and remain
in the global HashMap without eviction, accumulating clients, HTTP pools and
memory for the FE lifetime. Preserve exact authority/profile isolat
ion while adding bounded ownership and cleanup for these request-scoped
children, with a many-distinct-paths lifecycle regression.
--
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]
