[Feature][Connector V2][File] Add config of 'file_filter_pattern', wh…

…ich used for filtering files.

docs/en/connector-v2/source/CosFile.md

@@ -55,6 +55,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
 | schema                    | config  | no       | -                   |
 | common-options            |         | no       | -                   |
 | sheet_name                | string  | no       | -                   |
+| file_filter_pattern       | string  | no       | *                   |
 
 ### path [string]
 
@@ -247,6 +248,10 @@ Source plugin common parameters, please refer to [Source Common Options](common-
 
 Reader the sheet of the workbook,Only used when file_format is excel.
 
+### file_filter_pattern [string]
+
+Filter pattern, which used for filtering files.
+
 ## Example
 
 ```hocon

docs/en/connector-v2/source/HdfsFile.md

@@ -53,6 +53,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
 | schema                    | config  | no       | -                   |
 | common-options            |         | no       | -                   |
 | sheet_name                | string  | no       | -                   |
+| file_filter_pattern       | string  | no       | *                   |
 
 ### path [string]
 
@@ -245,6 +246,10 @@ Source plugin common parameters, please refer to [Source Common Options](common-
 
 Reader the sheet of the workbook,Only used when file_format is excel.
 
+### file_filter_pattern [string]
+
+Filter pattern, which used for filtering files.
+
 ## Example
 
 ```hocon

docs/en/connector-v2/source/LocalFile.md

@@ -49,6 +49,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
 | schema                    | config  | no       | -                   |
 | common-options            |         | no       | -                   |
 | sheet_name                | string  | no       | -                   |
+| file_filter_pattern       | string  | no       | *                   |
 
 ### path [string]
 
@@ -225,6 +226,10 @@ Source plugin common parameters, please refer to [Source Common Options](common-
 
 Reader the sheet of the workbook,Only used when file_format is excel.
 
+### file_filter_pattern [string]
+
+Filter pattern, which used for filtering files.
+
 ## Example
 
 ```hocon

docs/en/connector-v2/source/OssFile.md

@@ -56,6 +56,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
 | schema                    | config  | no       | -                   |
 | common-options            |         | no       | -                   |
 | sheet_name                | string  | no       | -                   |
+| file_filter_pattern       | string  | no       | *                   |
 
 ### path [string]
 
@@ -282,6 +283,10 @@ Reader the sheet of the workbook,Only used when file_format is excel.
 
 ```
 
+### file_filter_pattern [string]
+
+Filter pattern, which used for filtering files.
+
 ## Changelog
 
 ### 2.2.0-beta 2022-09-26

docs/en/connector-v2/source/OssJindoFile.md

@@ -56,6 +56,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
 | schema                    | config  | no       | -                   |
 | common-options            |         | no       | -                   |
 | sheet_name                | string  | no       | -                   |
+| file_filter_pattern       | string  | no       | *                   |
 
 ### path [string]
 
@@ -248,6 +249,10 @@ Source plugin common parameters, please refer to [Source Common Options](common-
 
 Reader the sheet of the workbook,Only used when file_format is excel.
 
+### file_filter_pattern [string]
+
+Filter pattern, which used for filtering files.
+
 ## Example
 
 ```hocon

docs/en/connector-v2/source/S3File.md

@@ -57,6 +57,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
 | schema                          | config  | no       | -                                                     |
 | common-options                  |         | no       | -                                                     |
 | sheet_name                      | string  | no       | -                                                     |
+| file_filter_pattern             | string  | no       | *                                                     |
 
 ### path [string]
 
@@ -299,6 +300,10 @@ Reader the sheet of the workbook,Only used when file_format is excel.
 
 ```
 
+### file_filter_pattern [string]
+
+Filter pattern, which used for filtering files.
+
 ## Changelog
 
 ### 2.3.0-beta 2022-10-20

docs/en/connector-v2/source/SftpFile.md

@@ -47,6 +47,7 @@ If you use SeaTunnel Engine, It automatically integrated the hadoop jar when you
 | schema                    | config  | no       | -                   |
 | common-options            |         | no       | -                   |
 | sheet_name                | string  | no       | -                   |
+| file_filter_pattern       | string  | no       | *                   |
 
 ### host [string]
 
@@ -226,6 +227,10 @@ Source plugin common parameters, please refer to [Source Common Options](common-
 
 Reader the sheet of the workbook,Only used when file_format is excel.
 
+### file_filter_pattern [string]
+
+Filter pattern, which used for filtering files.
+
 ## Example
 
 ```hocon

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseSourceConfig.java

@@ -112,4 +112,11 @@ public class BaseSourceConfig {
                     .stringType()
                     .noDefaultValue()
                     .withDescription("To be read sheet name,only valid for excel files");
+
+    public static final Option<String> FILE_FILTER_PATTERN =
+            Options.key("file_filter_pattern")
+                    .stringType()
+                    .defaultValue("*")
+                    .withDescription(
+                            "File pattern. The connector will filter some files base on the pattern.");
 }

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java

@@ -43,6 +43,9 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import static org.apache.parquet.avro.AvroReadSupport.READ_INT96_AS_FIXED;
 import static org.apache.parquet.avro.AvroSchemaConverter.ADD_LIST_ELEMENT_RECORDS;
@@ -74,6 +77,8 @@ public abstract class AbstractReadStrategy implements ReadStrategy {
     protected long skipHeaderNumber = BaseSourceConfig.SKIP_HEADER_ROW_NUMBER.defaultValue();
     protected boolean isKerberosAuthorization = false;
 
+    protected Pattern pattern;
+
     @Override
     public void init(HadoopConf conf) {
         this.hadoopConf = conf;
@@ -126,7 +131,7 @@ public List<String> getFileNamesByPath(HadoopConf hadoopConf, String path) throw
                 fileNames.addAll(getFileNamesByPath(hadoopConf, fileStatus.getPath().toString()));
                 continue;
             }
-            if (fileStatus.isFile()) {
+            if (fileStatus.isFile() && filterFileByPattern(fileStatus)) {
                 // filter '_SUCCESS' file
                 if (!fileStatus.getPath().getName().equals("_SUCCESS")
                         && !fileStatus.getPath().getName().startsWith(".")) {
@@ -146,6 +151,12 @@ public List<String> getFileNamesByPath(HadoopConf hadoopConf, String path) throw
                 }
             }
         }
+
+        if (fileNames.isEmpty()) {
+            throw new IllegalArgumentException(
+                    "Got no file, please check the configuration parameters such as: [file_filter_pattern]");
+        }
+
         return fileNames;
     }
 
@@ -166,6 +177,11 @@ public void setPluginConfig(Config pluginConfig) {
         if (pluginConfig.hasPath(BaseSourceConfig.READ_COLUMNS.key())) {
             readColumns.addAll(pluginConfig.getStringList(BaseSourceConfig.READ_COLUMNS.key()));
         }
+        if (pluginConfig.hasPath(BaseSourceConfig.FILE_FILTER_PATTERN.key())) {
+            String filterPattern =
+                    pluginConfig.getString(BaseSourceConfig.FILE_FILTER_PATTERN.key());
+            this.pattern = Pattern.compile(Matcher.quoteReplacement(filterPattern));
+        }
     }
 
     @Override
@@ -214,4 +230,11 @@ protected SeaTunnelRowType mergePartitionTypes(String path, SeaTunnelRowType sea
         // return merge row type
         return new SeaTunnelRowType(newFieldNames, newFieldTypes);
     }
+
+    protected boolean filterFileByPattern(FileStatus fileStatus) {
+        if (Objects.nonNull(pattern)) {
+            return pattern.matcher(fileStatus.getPath().getName()).matches();
+        }
+        return true;
+    }
 }

seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java

@@ -60,6 +60,7 @@ public OptionRule optionRule() {
                 .optional(BaseSourceConfig.DATE_FORMAT)
                 .optional(BaseSourceConfig.DATETIME_FORMAT)
                 .optional(BaseSourceConfig.TIME_FORMAT)
+                .optional(BaseSourceConfig.FILE_FILTER_PATTERN)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java

@@ -60,6 +60,7 @@ public OptionRule optionRule() {
                 .optional(BaseSourceConfig.DATE_FORMAT)
                 .optional(BaseSourceConfig.DATETIME_FORMAT)
                 .optional(BaseSourceConfig.TIME_FORMAT)
+                .optional(BaseSourceConfig.FILE_FILTER_PATTERN)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java

@@ -57,6 +57,7 @@ public OptionRule optionRule() {
                 .optional(BaseSourceConfig.DATE_FORMAT)
                 .optional(BaseSourceConfig.DATETIME_FORMAT)
                 .optional(BaseSourceConfig.TIME_FORMAT)
+                .optional(BaseSourceConfig.FILE_FILTER_PATTERN)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java

@@ -60,6 +60,7 @@ public OptionRule optionRule() {
                 .optional(BaseSourceConfig.DATE_FORMAT)
                 .optional(BaseSourceConfig.DATETIME_FORMAT)
                 .optional(BaseSourceConfig.TIME_FORMAT)
+                .optional(BaseSourceConfig.FILE_FILTER_PATTERN)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java

@@ -56,6 +56,7 @@ public OptionRule optionRule() {
                 .optional(BaseSourceConfig.DATE_FORMAT)
                 .optional(BaseSourceConfig.DATETIME_FORMAT)
                 .optional(BaseSourceConfig.TIME_FORMAT)
+                .optional(BaseSourceConfig.FILE_FILTER_PATTERN)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java

@@ -60,6 +60,7 @@ public OptionRule optionRule() {
                 .optional(BaseSourceConfig.DATE_FORMAT)
                 .optional(BaseSourceConfig.DATETIME_FORMAT)
                 .optional(BaseSourceConfig.TIME_FORMAT)
+                .optional(BaseSourceConfig.FILE_FILTER_PATTERN)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java

@@ -65,6 +65,7 @@ public OptionRule optionRule() {
                 .optional(BaseSourceConfig.DATE_FORMAT)
                 .optional(BaseSourceConfig.DATETIME_FORMAT)
                 .optional(BaseSourceConfig.TIME_FORMAT)
+                .optional(BaseSourceConfig.FILE_FILTER_PATTERN)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java

@@ -60,6 +60,7 @@ public OptionRule optionRule() {
                 .optional(BaseSourceConfig.DATE_FORMAT)
                 .optional(BaseSourceConfig.DATETIME_FORMAT)
                 .optional(BaseSourceConfig.TIME_FORMAT)
+                .optional(BaseSourceConfig.FILE_FILTER_PATTERN)
                 .build();
     }