[FLINK-6417][connector/filesystem] Add opt-in glob enumeration for FileSource

docs/content/docs/connectors/datastream/filesystem.md

@@ -247,6 +247,58 @@ new HiveSource<>(
 {{< /tab >}}
 {{< /tabs >}}
 
+### Opt-in Glob Enumeration
+
+The default enumerators treat input paths literally, including names containing `*`, `?`, or `[`.
+To select paths using glob patterns, install a `GlobFileEnumerator` explicitly:
+
+```java
+FileSource<String> source =
+        FileSource.forRecordStreamFormat(
+                        new TextLineInputFormat(),
+                        new Path("hdfs:///data/partition-*/file-*.txt"))
+                .setFileEnumerator(
+                        () -> new GlobFileEnumerator(new NonSplittingRecursiveEnumerator()))
+                .build();
+```
+
+Choose a delegate appropriate for your format. `NonSplittingRecursiveEnumerator` keeps each file
+in one split; `BlockSplittingRecursiveEnumerator` can split formats that support reading file blocks.
+The glob enumerator discovers and deduplicates concrete files, then calls the delegate once to filter
+and split them. It does not pass directories to the delegate, so the delegate's custom directory filters
+or directory-specific subclass methods do not control traversal. To customize directory traversal,
+use the `GlobFileEnumerator(FileEnumerator, Predicate<Path>)` constructor.
+
+Patterns apply to path segments, not URI schemes or authorities:
+
+- `*` matches zero or more characters within one segment.
+- `?` matches one character within one segment.
+- `[abc]`, `[a-z]`, and `[!a-z]` match a character set, range, or negated range.
+- A whole segment `**` matches zero or more directory levels. For example, `data/**/*.txt`
+  includes both `data/file.txt` and `data/nested/file.txt`.
+- Matching is case-sensitive. Braces are literal, not alternatives.
+- Use `[*]`, `[?]`, `[[]`, and `[]]` for literal special characters. For example,
+  `report[[]2026].txt` selects `report[2026].txt`. Backslash escaping is unavailable because
+  Flink's `Path` normalizes backslashes to directory separators.
+
+A matched directory is read recursively. Thus, if `*.csv` matches a directory named `archive.csv`,
+its descendants are included even when their names do not end in `.csv`; use the delegate's file
+filter if a filename restriction must also apply to those descendants.
+By default, traversal skips directories whose names start with `.` or `_`; the delegate controls
+hidden-file filtering. Directory filtering starts at the fixed search prefix, not its ancestors.
+
+Discovery starts at the longest prefix without glob syntax and descends only through matching
+segments. It uses the configured Flink filesystem's status and listing operations, requiring access
+to the searched directories. On object stores, directory listings correspond to prefix listings.
+Broad patterns, especially `**`, can still be expensive; use a narrow fixed prefix where possible.
+With continuous discovery, this work is repeated each discovery interval.
+
+A glob with no matches, including a missing fixed search prefix, produces no files. A missing
+literal input still fails. Invalid patterns, permission errors, listing failures, and failures
+while creating splits are not treated as empty matches. Files selected through overlapping patterns
+are passed to the delegate only once. This option is for the DataStream `FileSource`; it does not
+change legacy `FileInputFormat` or SQL filesystem path handling.
+
 ### Current Limitations
 
 Watermarking does not work very well for large backlogs of files. This is because watermarks eagerly advance within a file, and the next file might contain data later than the watermark.
@@ -1025,4 +1077,3 @@ being efficient, the `FileSink` also uses the [Multi-part Upload](https://help.a
 feature of OSS(similar with S3).
 
 {{< top >}}
-

flink-architecture-tests/flink-architecture-tests-production/src/main/java/org/apache/flink/architecture/rules/ConnectorRules.java

@@ -58,24 +58,22 @@ private static DescribedPredicate<JavaClass> areEnclosedInPublicClasses() {
                 .as("are enclosed in public classes");
     }
 
+    static DescribedPredicate<JavaClass> areAllowedDependencies() {
+        return areFlinkClassesThatResideOutsideOfConnectorPackagesAndArePublic()
+                .or(JavaClass.Predicates.resideOutsideOfPackages("org.apache.flink.."))
+                .or(JavaClass.Predicates.resideInAnyPackage(CONNECTOR_PACKAGES))
+                .or(JavaClass.Predicates.resideInAnyPackage(UTIL_PACKAGES))
+                .onResultOf(JavaClass::getBaseComponentType);
+    }
+
     @ArchTest
     public static final ArchRule CONNECTOR_CLASSES_ONLY_DEPEND_ON_PUBLIC_API =
             freeze(
                     javaClassesThat(resideInAnyPackage(CONNECTOR_PACKAGES))
                             .and()
                             .areNotAnnotatedWith(Deprecated.class)
                             .should()
-                            .onlyDependOnClassesThat(
-                                    areFlinkClassesThatResideOutsideOfConnectorPackagesAndArePublic()
-                                            .or(
-                                                    JavaClass.Predicates.resideOutsideOfPackages(
-                                                            "org.apache.flink.."))
-                                            .or(
-                                                    JavaClass.Predicates.resideInAnyPackage(
-                                                            CONNECTOR_PACKAGES))
-                                            .or(
-                                                    JavaClass.Predicates.resideInAnyPackage(
-                                                            UTIL_PACKAGES)))
+                            .onlyDependOnClassesThat(areAllowedDependencies())
                             .as(
                                     "Connector production code must depend only on public API when outside of connector packages"));
 }

flink-architecture-tests/flink-architecture-tests-production/src/test/java/org/apache/flink/architecture/rules/ConnectorRulesTest.java

@@ -0,0 +1,93 @@
+/*
+ * 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.flink.architecture.rules;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.Public;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.io.GlobFilePathFilter;
+import org.apache.flink.core.fs.Path;
+
+import com.tngtech.archunit.core.domain.JavaClass;
+import com.tngtech.archunit.core.importer.ClassFileImporter;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests connector dependency classification, including array component types. */
+class ConnectorRulesTest {
+
+    private static final JavaClass DEPENDENCIES =
+            new ClassFileImporter().importClass(Dependencies.class);
+
+    @ParameterizedTest
+    @CsvSource({
+        "publicScalar, true",
+        "publicArray, true",
+        "publicMatrix, true",
+        "evolvingArray, true",
+        "evolvingMatrix, true",
+        "nestedPublicArray, true",
+        "pathArray, true",
+        "internalScalar, false",
+        "internalArray, false",
+        "internalMatrix, false",
+        "unannotatedArray, false",
+        "internalFilterArray, false",
+        "externalArray, true",
+        "primitiveArray, true"
+    })
+    void testAllowedDependency(String field, boolean allowed) {
+        assertThat(
+                        ConnectorRules.areAllowedDependencies()
+                                .test(DEPENDENCIES.getField(field).getRawType()))
+                .isEqualTo(allowed);
+    }
+
+    private static class Dependencies {
+        private PublicType publicScalar;
+        private PublicType[] publicArray;
+        private PublicType[][] publicMatrix;
+        private EvolvingType[] evolvingArray;
+        private EvolvingType[][] evolvingMatrix;
+        private PublicType.Nested[] nestedPublicArray;
+        private Path[] pathArray;
+        private InternalType internalScalar;
+        private InternalType[] internalArray;
+        private InternalType[][] internalMatrix;
+        private UnannotatedType[] unannotatedArray;
+        private GlobFilePathFilter[] internalFilterArray;
+        private String[] externalArray;
+        private int[] primitiveArray;
+    }
+
+    @Public
+    private static class PublicType {
+        private static class Nested {}
+    }
+
+    @PublicEvolving
+    private static class EvolvingType {}
+
+    @Internal
+    private static class InternalType {}
+
+    private static class UnannotatedType {}
+}

flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/enumerate/GlobFileEnumerator.java

@@ -0,0 +1,172 @@
+/*
+ * 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.flink.connector.file.src.enumerate;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A file enumerator that explicitly enables glob expansion in FileSource input paths.
+ *
+ * <p>Within a path segment, {@code *} matches zero or more characters, {@code ?} matches one
+ * character, and character classes such as {@code [abc]}, {@code [a-z]}, and {@code [!a-z]} match
+ * one character. A segment consisting of {@code **} matches zero or more directory levels. Matching
+ * is case-sensitive and independent of the local operating system. Braces are literal; use {@code
+ * [*]}, {@code [?]}, {@code [[]}, and {@code []]} for literal glob characters. Backslash escaping
+ * is not supported because {@link Path} normalizes backslashes to separators.
+ *
+ * <p>This enumerator owns directory traversal, including recursive traversal of matched
+ * directories. It passes unique, qualified <em>file</em> paths to the supplied delegate once per
+ * enumeration, preserving the parallelism hint and all returned splits. The delegate controls file
+ * filtering and splitting; it does not receive directories. Custom directory filters and
+ * directory-specific subclass hooks on the delegate therefore do not govern glob traversal. Use the
+ * separate directory filter to control traversal instead.
+ *
+ * <p>Enumeration starts at each input's longest prefix without glob syntax. Missing glob prefixes
+ * and patterns with no matches yield no files. Missing non-glob inputs, listing failures, and
+ * failures in the delegate are propagated. No discovery results are cached between invocations.
+ * Broad patterns, especially {@code **}, may require listing large directory trees.
+ *
+ * <p>Install this enumerator explicitly through the file source builder's {@code setFileEnumerator}
+ * method. Choose a delegate appropriate for the format, for example {@link
+ * NonSplittingRecursiveEnumerator} for a non-splittable format. Existing file enumerators do not
+ * expand globs.
+ */
+@PublicEvolving
+public final class GlobFileEnumerator implements FileEnumerator {
+
+    private final FileEnumerator delegate;
+    private final Predicate<Path> directoryFilter;
+
+    /**
+     * Creates an enumerator that skips directories whose names start with '.' or '_'.
+     *
+     * @param delegate enumerator receiving concrete files for filtering and splitting
+     */
+    @PublicEvolving
+    public GlobFileEnumerator(FileEnumerator delegate) {
+        this(delegate, new DefaultFileFilter());
+    }
+
+    /**
+     * Creates an enumerator with an explicit directory traversal policy.
+     *
+     * <p>The predicate is applied once per visited directory per invocation, starting at each
+     * input's fixed prefix. Rejected directories are not listed. Ancestors above that prefix are
+     * not visited. Hidden-file filtering remains the delegate's responsibility.
+     *
+     * @param delegate enumerator receiving concrete files, never directories
+     * @param directoryFilter predicate accepting directories to traverse
+     */
+    @PublicEvolving
+    public GlobFileEnumerator(FileEnumerator delegate, Predicate<Path> directoryFilter) {
+        this.delegate = checkNotNull(delegate);
+        this.directoryFilter = checkNotNull(directoryFilter);
+    }
+
+    @Override
+    @PublicEvolving
+    public Collection<FileSourceSplit> enumerateSplits(Path[] paths, int minDesiredSplits)
+            throws IOException {
+        // Validate all patterns before touching any filesystem.
+        final List<GlobPattern> patterns = new ArrayList<>(paths.length);
+        for (Path path : paths) {
+            patterns.add(new GlobPattern(checkNotNull(path)));
+        }
+
+        final Discovery discovery = new Discovery();
+        for (GlobPattern pattern : patterns) {
+            final FileSystem fs = pattern.getRoot().getFileSystem();
+            final Path root = pattern.getRoot().makeQualified(fs);
+            final FileStatus status;
+            try {
+                status = fs.getFileStatus(root);
+            } catch (FileNotFoundException e) {
+                if (pattern.hasGlob()) {
+                    continue;
+                }
+                throw e;
+            }
+            // Even ** cannot consume a path separator after a regular file.
+            if (status.isDir() || !pattern.hasGlob()) {
+                discovery.expand(fs, status, pattern, pattern.initialPositions());
+            }
+        }
+        return delegate.enumerateSplits(discovery.files.toArray(new Path[0]), minDesiredSplits);
+    }
+
+    /** Per-invocation state; continuous discovery must also see files created in later calls. */
+    private final class Discovery {
+
+        private final Set<Path> files = new LinkedHashSet<>();
+        private final Set<Path> collectedDirectories = new HashSet<>();
+        private final Map<Path, Boolean> acceptedDirectories = new HashMap<>();
+
+        private void expand(
+                FileSystem fs, FileStatus status, GlobPattern pattern, Set<Integer> positions)
+                throws IOException {
+            final Path path = status.getPath().makeQualified(fs);
+            if (status.isDir() && !acceptDirectory(path)) {
+                return;
+            }
+            if (pattern.isComplete(positions)) {
+                collectFiles(fs, status);
+            } else if (status.isDir() && !collectedDirectories.contains(path)) {
+                for (FileStatus child : fs.listStatus(path)) {
+                    final Set<Integer> next = pattern.matchChild(child, positions);
+                    if (!next.isEmpty()) {
+                        expand(fs, child, pattern, next);
+                    }
+                }
+            }
+        }
+
+        private void collectFiles(FileSystem fs, FileStatus status) throws IOException {
+            final Path path = status.getPath().makeQualified(fs);
+            if (!status.isDir()) {
+                files.add(path);
+            } else if (acceptDirectory(path) && collectedDirectories.add(path)) {
+                for (FileStatus child : fs.listStatus(path)) {
+                    collectFiles(fs, child);
+                }
+            }
+        }
+
+        private boolean acceptDirectory(Path path) {
+            return acceptedDirectories.computeIfAbsent(path, directoryFilter::test);
+        }
+    }
+}