luoyuxia commented on code in PR #22249:
URL: https://github.com/apache/flink/pull/22249#discussion_r1236760683
##########
docs/content/docs/connectors/table/filesystem.md:
##########
@@ -50,6 +50,10 @@ CREATE TABLE MyUserTable (
-- section for more details
'partition.default-name' = '...', -- optional: default partition name in
case the dynamic partition
-- column value is null/empty string
+ 'source.path.regex-pattern' = '...', -- optional: regex pattern to filter
files under the directory
Review Comment:
```suggestion
'source.path.regex-pattern' = '...', -- optional: regex pattern to filter
files to read under the directory
```
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/enumerate/BlockSplittingRecursiveAllDirEnumerator.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.Internal;
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.compression.StandardDeCompressors;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.function.Predicate;
+
+/**
+ * This {@code FileEnumerator} enumerates all files under the given paths
recursively except the
+ * hidden directories, and creates a separate split for each file block.
+ *
+ * <p>Please note that file blocks are only exposed by some file systems, such
as HDFS. File systems
+ * that do not expose block information will not create multiple file splits
per file, but keep the
+ * files as one source split.
+ *
+ * <p>Files with suffixes corresponding to known compression formats (for
example '.gzip', '.bz2',
+ * ...) will not be split. See {@link StandardDeCompressors} for a list of
known formats and
+ * suffixes.
+ *
+ * <p>Compared to {@link BlockSplittingRecursiveEnumerator}, this enumerator
will enumerate all
+ * files even through its parent directory is filtered out by the file filter.
+ */
+@Internal
+public class BlockSplittingRecursiveAllDirEnumerator extends
BlockSplittingRecursiveEnumerator {
+
+ /** The filter used to skip hidden directories. */
+ private final DefaultFileFilter hiddenDirFilter = new DefaultFileFilter();
+
+ /**
+ * Creates a new enumerator that enumerates all files whose file path
matches the regex except
+ * hidden files. Hidden files are considered files where the filename
starts with '.' or with
+ * '_'.
+ *
+ * <p>The enumerator does not split files that have a suffix corresponding
to a known
+ * compression format (for example '.gzip', '.bz2', '.xy', '.zip', ...).
See {@link
+ * StandardDeCompressors} for details.
+ */
+ public BlockSplittingRecursiveAllDirEnumerator(String pathPattern) {
+ this(
+ new RegexFileFilter(pathPattern),
+ StandardDeCompressors.getCommonSuffixes().toArray(new
String[0]));
+ }
+
+ /**
+ * Creates a new enumerator that uses the given predicate as a filter for
file paths, and avoids
+ * splitting files with the given extension (typically to avoid splitting
compressed files).
+ */
+ public BlockSplittingRecursiveAllDirEnumerator(
+ final Predicate<Path> fileFilter, final String[]
nonSplittableFileSuffixes) {
+ super(fileFilter, nonSplittableFileSuffixes);
+ }
+
+ @Override
+ protected void addSplitsForPath(
+ FileStatus fileStatus, FileSystem fs, ArrayList<FileSourceSplit>
target)
+ throws IOException {
+ if (fileStatus.isDir()) {
+ if (!hiddenDirFilter.test(fileStatus.getPath())) {
+ return;
+ }
+ final FileStatus[] containedFiles =
fs.listStatus(fileStatus.getPath());
+ for (FileStatus containedStatus : containedFiles) {
+ // If this dir matches the regex, add all files under it as a
split (not include the
Review Comment:
plase remove these comments since it's out of date
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/enumerate/BlockSplittingRecursiveAllDirEnumerator.java:
##########
@@ -0,0 +1,97 @@
+/*
+ * 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.Internal;
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.compression.StandardDeCompressors;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.function.Predicate;
+
+/**
+ * This {@code FileEnumerator} enumerates all files under the given paths
recursively except the
+ * hidden directories, and creates a separate split for each file block.
+ *
+ * <p>Please note that file blocks are only exposed by some file systems, such
as HDFS. File systems
+ * that do not expose block information will not create multiple file splits
per file, but keep the
+ * files as one source split.
+ *
+ * <p>Files with suffixes corresponding to known compression formats (for
example '.gzip', '.bz2',
+ * ...) will not be split. See {@link StandardDeCompressors} for a list of
known formats and
+ * suffixes.
+ *
+ * <p>Compared to {@link BlockSplittingRecursiveEnumerator}, this enumerator
will enumerate all
+ * files even through its parent directory is filtered out by the file filter.
+ */
+@Internal
+public class BlockSplittingRecursiveAllDirEnumerator extends
BlockSplittingRecursiveEnumerator {
+
+ /** The filter used to skip hidden directories. */
+ private final DefaultFileFilter hiddenDirFilter = new DefaultFileFilter();
+
+ /**
+ * Creates a new enumerator that enumerates all files whose file path
matches the regex except
+ * hidden files. Hidden files are considered files where the filename
starts with '.' or with
+ * '_'.
+ *
+ * <p>The enumerator does not split files that have a suffix corresponding
to a known
+ * compression format (for example '.gzip', '.bz2', '.xy', '.zip', ...).
See {@link
+ * StandardDeCompressors} for details.
+ */
+ public BlockSplittingRecursiveAllDirEnumerator(String pathPattern) {
+ this(
+ new RegexFileFilter(pathPattern),
+ StandardDeCompressors.getCommonSuffixes().toArray(new
String[0]));
+ }
+
+ /**
+ * Creates a new enumerator that uses the given predicate as a filter for
file paths, and avoids
+ * splitting files with the given extension (typically to avoid splitting
compressed files).
+ */
+ public BlockSplittingRecursiveAllDirEnumerator(
+ final Predicate<Path> fileFilter, final String[]
nonSplittableFileSuffixes) {
+ super(fileFilter, nonSplittableFileSuffixes);
+ }
+
+ @Override
+ protected void addSplitsForPath(
+ FileStatus fileStatus, FileSystem fs, ArrayList<FileSourceSplit>
target)
+ throws IOException {
+ if (fileStatus.isDir()) {
+ if (!hiddenDirFilter.test(fileStatus.getPath())) {
+ return;
+ }
+ final FileStatus[] containedFiles =
fs.listStatus(fileStatus.getPath());
+ for (FileStatus containedStatus : containedFiles) {
+ // If this dir matches the regex, add all files under it as a
split (not include the
Review Comment:
At least I haven't seen the logic that `If this dir matches the regex,.....`
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/enumerate/NonSplittingRecursiveAllDirEnumerator.java:
##########
@@ -0,0 +1,84 @@
+/*
+ * 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.Internal;
+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.IOException;
+import java.util.ArrayList;
+import java.util.function.Predicate;
+
+/**
+ * This {@code FileEnumerator} enumerates all files under the given paths
recursively except the
+ * hidden directories. Each file matched the given regex pattern becomes one
split; this enumerator
+ * does not split files into smaller "block" units.
+ *
+ * <p>The default instantiation of this enumerator filters files with the
common hidden file
+ * prefixes '.' and '_'. A custom file filter can be specified.
+ *
+ * <p>Compared to {@link NonSplittingRecursiveEnumerator}, this enumerator
will enumerate all files
+ * even through its parent directory is filtered out by the file filter.
+ */
+@Internal
+public class NonSplittingRecursiveAllDirEnumerator extends
NonSplittingRecursiveEnumerator {
+ /** The filter used to skip hidden directories. */
+ private final DefaultFileFilter hiddenDirFilter = new DefaultFileFilter();
+
+ /**
+ * Creates a NonSplittingRegexEnumerator that enumerates all files whose
file path matches the
+ * regex except hidden files. Hidden files are considered files where the
filename starts with
+ * '.' or with '_'.
+ */
+ public NonSplittingRecursiveAllDirEnumerator(String pathRegexPattern) {
+ this(new RegexFileFilter(pathRegexPattern));
+ }
+
+ /**
+ * Creates a NonSplittingRegexEnumerator that enumerates all files whose
file path matches the
+ * regex. Support to use given custom predicate as a filter for file paths.
+ */
+ public NonSplittingRecursiveAllDirEnumerator(Predicate<Path> fileFilter) {
+ super(fileFilter);
+ }
+
+ @Override
+ protected void addSplitsForPath(
+ FileStatus fileStatus, FileSystem fs, ArrayList<FileSourceSplit>
target)
+ throws IOException {
+ if (fileStatus.isDir()) {
+ if (!hiddenDirFilter.test(fileStatus.getPath())) {
+ return;
+ }
+ final FileStatus[] containedFiles =
fs.listStatus(fileStatus.getPath());
+ for (FileStatus containedStatus : containedFiles) {
+ // If this dir matches the regex, add all files under it as a
split (not include the
Review Comment:
dito
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemConnectorOptions.java:
##########
@@ -68,6 +68,17 @@ public class FileSystemConnectorOptions {
+ "The statistics reporting is a heavy
operation in some cases,"
+ "this config allows users to choose the
statistics type according to different situations.");
+ public static final ConfigOption<String> SOURCE_REGEX_PATTERN =
Review Comment:
```suggestion
public static final ConfigOption<String> SOURCE_PATH_REGEX_PATTERN =
```
?
--
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]
