This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12817-89a5598eb5e258f30248009572c58d0deba24d60 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 449225292833a7f8c5e0b0a9169a28b855245c86 Author: chovygo <[email protected]> AuthorDate: Sun Oct 18 06:25:03 2026 +0000 [Fix][Connector-V2] Declare missing source options in file source factories (#12817) Co-authored-by: zhaowei <[email protected]> --- docs/en/connectors/source/GcsFile.md | 6 +++ docs/en/connectors/source/OssFile.md | 1 + docs/en/connectors/source/OssJindoFile.md | 2 + docs/en/connectors/source/SmbFile.md | 7 +++ docs/zh/connectors/source/GcsFile.md | 6 +++ docs/zh/connectors/source/OssFile.md | 1 + docs/zh/connectors/source/OssJindoFile.md | 2 + docs/zh/connectors/source/SmbFile.md | 7 +++ .../file/bos/source/BosFileSourceFactory.java | 9 ++++ .../seatunnel/file/bos/BosFileFactoryTest.java | 31 ++++++++++++ .../file/cos/source/CosFileSourceFactory.java | 9 ++++ .../seatunnel/file/cos/CosFileFactoryTest.java | 31 ++++++++++++ .../file/ftp/source/FtpFileSourceFactory.java | 9 ++++ .../seatunnel/file/ftp/FtpFileFactoryTest.java | 29 ++++++++++++ .../file/gcs/source/GcsFileSourceFactory.java | 9 ++++ .../file/gcs/GcsFileSourceFactoryTest.java | 28 +++++++++++ .../file/hdfs/source/HdfsFileSourceFactory.java | 9 ++++ .../seatunnel/file/hdfs/HdfsFileFactoryTest.java | 29 ++++++++++++ .../oss/jindo/source/OssFileSourceFactory.java | 10 ++++ .../connectors/test/OssJindoFactoryTest.java | 52 ++++++++++++++++++++ .../file/local/source/LocalFileSourceFactory.java | 9 ++++ .../seatunnel/file/local/LocalFileFactoryTest.java | 36 ++++++++++++++ .../file/obs/source/ObsFileSourceFactory.java | 11 +++++ .../seatunnel/file/obs/ObsFileFactoryTest.java | 55 +++++++++++++++++++++- .../file/oss/source/OssFileSourceFactory.java | 9 ++++ .../seatunnel/file/oss/OssFileFactoryTest.java | 31 ++++++++++++ .../file/s3/source/S3FileSourceFactory.java | 9 ++++ .../seatunnel/file/s3/S3FileFactoryTest.java | 29 ++++++++++++ .../file/sftp/source/SftpFileSourceFactory.java | 11 +++++ .../seatunnel/file/sftp/SftpFileFactoryTest.java | 44 +++++++++++++++++ .../file/smb/source/SmbFileSourceFactory.java | 11 +++++ .../seatunnel/file/smb/SmbFileFactoryTest.java | 44 +++++++++++++++++ 32 files changed, 585 insertions(+), 1 deletion(-) diff --git a/docs/en/connectors/source/GcsFile.md b/docs/en/connectors/source/GcsFile.md index 260b3147ed..b95c30d397 100644 --- a/docs/en/connectors/source/GcsFile.md +++ b/docs/en/connectors/source/GcsFile.md @@ -74,6 +74,8 @@ The explicit `service_account_key_file` option takes precedence over the corresp | enable_file_split | boolean | no | `false` | Enables logical splitting for uncompressed text, CSV, JSON, and Parquet files. | | file_split_size | long | conditional | `134217728` | Split size in bytes when `enable_file_split=true`. | | null_format | string | no | - | Text representation of null values. | +| binary_chunk_size | int | no | 1024 | Only used when file_format_type is binary. The chunk size (in bytes) for reading binary files. Larger values may improve performance for large files but use more memory. | +| binary_complete_file_mode | boolean | no | false | Only used when file_format_type is binary. Whether to read the complete file as a single chunk instead of splitting into chunks. | | quote_char | string | no | `"` | CSV quote character. | | escape_char | string | no | - | CSV escape character. | | sheet_name | string | no | - | Excel worksheet to read. | @@ -81,6 +83,7 @@ The explicit `service_account_key_file` option takes precedence over the corresp | poi_excel_max_file_size | long | no | `52428800` | Maximum Excel file size in bytes for the POI engine. | | xml_row_tag | string | conditional | - | XML element representing one row. | | xml_use_attr_format | boolean | conditional | - | Whether XML values are read from attributes. | +| csv_use_header_line | boolean | no | false | Whether to use the header line to parse the file, only used when the file_format is `csv` and the file contains the header line that match RFC 4180. | | discovery_mode | string | no | `once` | `once` or `continuous`. Continuous mode currently requires update sync and binary format. | | scan_interval | string | no | `10S` | Poll interval for continuous discovery. | | start_mode | string | no | `earliest` | `earliest` processes existing files; `latest` starts with later changes. | @@ -95,6 +98,9 @@ The explicit `service_account_key_file` option takes precedence over the corresp | backup_path | string | conditional | - | Required for `post_sync_action=backup`. Must not overlap the source path. | | retention_max_age | string | no | - | Maximum age for SeaTunnel backup objects. | | retention_check_interval | string | no | `1H` | Backup retention scan interval. | +| file_filter_modified_start | string | no | - | File modification time filter. The connector will filter files based on the last modification start time (inclusive). Default format: `yyyy-MM-dd HH:mm:ss`. | +| file_filter_modified_end | string | no | - | File modification time filter. The connector will filter files based on the last modification end time (exclusive). Default format: `yyyy-MM-dd HH:mm:ss`. | +| sort_files_by_modification_time | boolean | no | false | Sort files by modification time in descending order. Enable this when reading evolving schemas to ensure schema inference uses the latest file. | | common-options | | no | - | See [Source Common Options](../common-options/source-common-options.md). | ## Example diff --git a/docs/en/connectors/source/OssFile.md b/docs/en/connectors/source/OssFile.md index 8d937d0d09..2e3b8357cc 100644 --- a/docs/en/connectors/source/OssFile.md +++ b/docs/en/connectors/source/OssFile.md @@ -210,6 +210,7 @@ If you assign file type to `parquet` `orc`, schema option not required, connecto | xml_use_attr_format | boolean | no | - | Specifies whether to process data using the tag attribute format, only used when file_format is xml. | | csv_use_header_line | boolean | no | false | Whether to use the header line to parse the file, only used when the file_format is `csv` and the file contains the header line that match RFC 4180 | | compress_codec | string | no | none | Which compress codec the files used. | +| archive_compress_codec | string | no | none | Which archive compress codec the files used. Supported: `none` `zip` `tar` `tar.gz` `gz`. | | encoding | string | no | UTF-8 | File encoding, only used when `file_format_type` is `json`, `text`, `csv`, or `xml`. | | null_format | string | no | - | Only used when file_format_type is text. null_format to define which strings can be represented as null. e.g: `\N` | | binary_chunk_size | int | no | 1024 | Only used when file_format_type is binary. The chunk size (in bytes) for reading binary files. Default is 1024 bytes. Larger values may improve performance for large files but use more memory. | diff --git a/docs/en/connectors/source/OssJindoFile.md b/docs/en/connectors/source/OssJindoFile.md index ffd62381c1..af07373807 100644 --- a/docs/en/connectors/source/OssJindoFile.md +++ b/docs/en/connectors/source/OssJindoFile.md @@ -87,6 +87,8 @@ It only supports hadoop version **2.9.X+**. | archive_compress_codec | string | no | none | | encoding | string | no | UTF-8 | | null_format | string | no | - | +| binary_chunk_size | int | no | 1024 | +| binary_complete_file_mode | boolean | no | false | | common-options | | no | - | | file_filter_modified_start | string | no | - | | file_filter_modified_end | string | no | - | diff --git a/docs/en/connectors/source/SmbFile.md b/docs/en/connectors/source/SmbFile.md index 94ae3bdeac..f24e4cddea 100644 --- a/docs/en/connectors/source/SmbFile.md +++ b/docs/en/connectors/source/SmbFile.md @@ -99,9 +99,13 @@ The File does not have a specific type list, and we can indicate which SeaTunnel | sheet_name | String | No | - | Reader the sheet of the workbook, Only used when file_format is excel. | | xml_row_tag | String | No | - | Specifies the tag name of the data rows within the XML file, only used when file_format is xml. | | xml_use_attr_format | Boolean | No | - | Specifies whether to process data using the tag attribute format, only used when file_format is xml. | +| csv_use_header_line | boolean | no | false | Whether to use the header line to parse the file, only used when the file_format is `csv` and the file contains the header line that match RFC 4180. | | compress_codec | String | No | None | The compress codec of files | +| archive_compress_codec | String | No | none | Which archive compress codec the files used. Supported: `none` `zip` `tar` `tar.gz` `gz`. | | encoding | String | No | UTF-8 | The encoding of the file to read | | null_format | String | No | - | Only used when file_format_type is text. Define which strings can be represented as null, e.g. `\N`. | +| binary_chunk_size | int | no | 1024 | Only used when file_format_type is binary. The chunk size (in bytes) for reading binary files. Larger values may improve performance for large files but use more memory. | +| binary_complete_file_mode | boolean | no | false | Only used when file_format_type is binary. Whether to read the complete file as a single chunk instead of splitting into chunks. | | filename_extension | String | No | - | Filter filename extension, which used for filtering files with specific extension. Example: `csv` `.txt` `json` `.xml`. | | excel_engine | String | No | POI | Only used when file_format is excel. Supported engines are `POI` and `EasyExcel`. | | poi_excel_max_file_size | Long | No | 52428800 | Only used when file_format is excel and excel_engine is POI. The maximum Excel file size in bytes (default 50 MB). | @@ -123,6 +127,9 @@ The File does not have a specific type list, and we can indicate which SeaTunnel | retention_max_age | String | No | - | Optional retention age for backup files, only valid when `post_sync_action=backup`. | | retention_check_interval | String | No | 1H | Retention scan interval, only effective when `post_sync_action=backup` and `retention_max_age` is configured. | | recursive_file_scan | Boolean | No | true | Whether to scan subdirectories recursively. If `false`, subdirectories will be ignored. | +| file_filter_modified_start | string | no | - | File modification time filter. The connector will filter files based on the last modification start time (inclusive). Default format: `yyyy-MM-dd HH:mm:ss`. | +| file_filter_modified_end | string | no | - | File modification time filter. The connector will filter files based on the last modification end time (exclusive). Default format: `yyyy-MM-dd HH:mm:ss`. | +| sort_files_by_modification_time | boolean | no | false | Sort files by modification time in descending order. Enable this when reading evolving schemas to ensure schema inference uses the latest file. | | common-options | | No | - | Source plugin common parameters, please refer to [Source Common Options](../common-options/source-common-options.md) for details. | ## How to Create a SMB Data Synchronization Job diff --git a/docs/zh/connectors/source/GcsFile.md b/docs/zh/connectors/source/GcsFile.md index 0aa7f4a74e..e60bc57e53 100644 --- a/docs/zh/connectors/source/GcsFile.md +++ b/docs/zh/connectors/source/GcsFile.md @@ -64,6 +64,8 @@ import ChangeLog from '../changelog/connector-file-gcs.md'; | enable_file_split | boolean | 否 | `false` | 为未压缩的 text、CSV、JSON 和 Parquet 文件启用逻辑切分。 | | file_split_size | long | 条件必填 | `134217728` | `enable_file_split=true` 时的切分大小，单位为字节。 | | null_format | string | 否 | - | 表示空值的文本。 | +| binary_chunk_size | int | 否 | 1024 | 仅在 file_format_type 为 binary 时使用。读取二进制文件的块大小（以字节为单位）。较大的值可能会提高大文件的性能，但会使用更多内存。 | +| binary_complete_file_mode | boolean | 否 | false | 仅在 file_format_type 为 binary 时使用。是否将完整文件作为单个块读取，而不是分割成块。 | | quote_char | string | 否 | `"` | CSV 引号字符。 | | escape_char | string | 否 | - | CSV 转义字符。 | | sheet_name | string | 否 | - | 要读取的 Excel 工作表。 | @@ -71,6 +73,7 @@ import ChangeLog from '../changelog/connector-file-gcs.md'; | poi_excel_max_file_size | long | 否 | `52428800` | POI 引擎可读取的最大 Excel 文件字节数。 | | xml_row_tag | string | 条件必填 | - | 表示一行数据的 XML 元素。 | | xml_use_attr_format | boolean | 条件必填 | - | 是否从 XML 属性读取值。 | +| csv_use_header_line | boolean | 否 | false | 是否使用标题行来解析文件，仅在 file_format 为 `csv` 且文件包含符合 RFC 4180 的标题行时使用。 | | discovery_mode | string | 否 | `once` | `once` 或 `continuous`。连续发现当前要求 update 同步和 binary 格式。 | | scan_interval | string | 否 | `10S` | 连续发现的轮询间隔。 | | start_mode | string | 否 | `earliest` | `earliest` 处理已有文件，`latest` 从后续变更开始。 | @@ -85,6 +88,9 @@ import ChangeLog from '../changelog/connector-file-gcs.md'; | backup_path | string | 条件必填 | - | `post_sync_action=backup` 时必填，且不能与源路径重叠。 | | retention_max_age | string | 否 | - | SeaTunnel 备份对象的最大保留时间。 | | retention_check_interval | string | 否 | `1H` | 备份保留扫描间隔。 | +| file_filter_modified_start | string | 否 | - | 按文件最后修改时间筛选文件的起始时间（包含该时间）。默认格式：`yyyy-MM-dd HH:mm:ss`。 | +| file_filter_modified_end | string | 否 | - | 按文件最后修改时间筛选文件的结束时间（不包含该时间）。默认格式：`yyyy-MM-dd HH:mm:ss`。 | +| sort_files_by_modification_time | boolean | 否 | false | 是否按修改时间降序排序文件。启用此选项后，在读取不断演化的 schema 时可确保 schema 推断使用最新的文件。 | | common-options | | 否 | - | 参见 [Source 通用配置](../common-options/source-common-options.md)。 | ## 示例 diff --git a/docs/zh/connectors/source/OssFile.md b/docs/zh/connectors/source/OssFile.md index f7231c21bb..a3b5915e25 100644 --- a/docs/zh/connectors/source/OssFile.md +++ b/docs/zh/connectors/source/OssFile.md @@ -210,6 +210,7 @@ schema { | xml_row_tag | string | 否 | - | 指定XML文件中数据行的标签名称，仅在file_format为xml时使用。 | | xml_use_attr_format | boolean | 否 | - | 指定是否使用标签属性格式处理数据，仅在file_format为xml时使用。 | | compress_codec | string | 否 | none | 文件使用的压缩编解码器。 | +| archive_compress_codec | string | 否 | none | 文件使用的归档压缩编解码器。支持：`none` `zip` `tar` `tar.gz` `gz`。 | | encoding | string | 否 | UTF-8 | 文件编码，仅在 `file_format_type` 为 `json`、`text`、`csv` 或 `xml` 时使用。 | | null_format | string | 否 | - | 仅在file_format_type为text时使用。null_format用于定义哪些字符串可以表示为null。例如：`\N` | | binary_chunk_size | int | 否 | 1024 | 仅在file_format_type为binary时使用。读取二进制文件的块大小（以字节为单位）。默认为1024字节。较大的值可能会提高大文件的性能，但会使用更多内存。 | diff --git a/docs/zh/connectors/source/OssJindoFile.md b/docs/zh/connectors/source/OssJindoFile.md index b94af55993..a872c98592 100644 --- a/docs/zh/connectors/source/OssJindoFile.md +++ b/docs/zh/connectors/source/OssJindoFile.md @@ -87,6 +87,8 @@ import ChangeLog from '../changelog/connector-file-oss-jindo.md'; | archive_compress_codec | string | 否 | none | 读取归档文件时使用的归档压缩格式。 | | encoding | string | 否 | UTF-8 | 读取 `json`、`text`、`csv` 或 `xml` 文件时使用的字符编码。 | | null_format | string | 否 | - | 仅用于 `text` 格式，指定应解析为 `null` 的字符串。 | +| binary_chunk_size | int | 否 | 1024 | 仅在 file_format_type 为 binary 时使用。读取二进制文件的块大小（以字节为单位）。默认为 1024 字节。较大的值可能会提高大文件的性能，但会使用更多内存。 | +| binary_complete_file_mode | boolean | 否 | false | 仅在 file_format_type 为 binary 时使用。是否将完整文件作为单个块读取，而不是分割成块。 | | common-options | | 否 | - | Source 插件通用参数，详见 [Source Common Options](../common-options/source-common-options.md)。 | | file_filter_modified_start | string | 否 | - | 按文件最后修改时间筛选文件的起始时间（包含该时间）。 | | file_filter_modified_end | string | 否 | - | 按文件最后修改时间筛选文件的结束时间（不包含该时间）。 | diff --git a/docs/zh/connectors/source/SmbFile.md b/docs/zh/connectors/source/SmbFile.md index 02f45542dd..6c42be4837 100644 --- a/docs/zh/connectors/source/SmbFile.md +++ b/docs/zh/connectors/source/SmbFile.md @@ -99,9 +99,13 @@ import ChangeLog from '../changelog/connector-file-smb.md'; | sheet_name | String | 否 | - | 读取工作簿中的 sheet，仅在 file_format 为 excel 时使用 | | xml_row_tag | String | 否 | - | 指定 XML 文件中数据行的标签名，仅在 file_format 为 xml 时使用 | | xml_use_attr_format | Boolean | 否 | - | 指定是否使用标签属性格式处理数据，仅在 file_format 为 xml 时使用 | +| csv_use_header_line | boolean | 否 | false | 是否使用标题行来解析文件，仅在 file_format 为 `csv` 且文件包含符合 RFC 4180 的标题行时使用 | | compress_codec | String | 否 | None | 文件的压缩编解码器 | +| archive_compress_codec | String | 否 | none | 文件使用的归档压缩编解码器。支持：`none` `zip` `tar` `tar.gz` `gz`。 | | encoding | String | 否 | UTF-8 | 读取文件时使用的编码 | | null_format | String | 否 | - | 仅在 file_format_type 为 text 时使用。定义哪些字符串可以表示为 null，例如 `\N` | +| binary_chunk_size | int | 否 | 1024 | 仅在 file_format_type 为 binary 时使用。读取二进制文件的块大小（以字节为单位）。较大的值可能会提高大文件的性能，但会使用更多内存。 | +| binary_complete_file_mode | boolean | 否 | false | 仅在 file_format_type 为 binary 时使用。是否将完整文件作为单个块读取，而不是分割成块。 | | filename_extension | String | 否 | - | 文件扩展名过滤，用于过滤特定扩展名的文件。例如：`csv` `.txt` `json` `.xml` | | excel_engine | String | 否 | POI | 仅在 file_format 为 excel 时使用。支持的引擎为 `POI` 和 `EasyExcel` | | poi_excel_max_file_size | Long | 否 | 52428800 | 仅在 file_format 为 excel 且 excel_engine 为 POI 时使用。POI 引擎可读取的最大 Excel 文件大小（默认 50 MB） | @@ -123,6 +127,9 @@ import ChangeLog from '../changelog/connector-file-smb.md'; | retention_max_age | String | 否 | - | 备份文件的可选保留期限，仅在 `post_sync_action=backup` 时有效 | | retention_check_interval | String | 否 | 1H | 保留扫描间隔，仅在配置了 `post_sync_action=backup` 和 `retention_max_age` 时生效 | | recursive_file_scan | Boolean | 否 | true | 是否递归扫描子目录。如果为 `false`，将忽略子目录 | +| file_filter_modified_start | string | 否 | - | 按文件最后修改时间筛选文件的起始时间（包含该时间）。默认格式：`yyyy-MM-dd HH:mm:ss`。 | +| file_filter_modified_end | string | 否 | - | 按文件最后修改时间筛选文件的结束时间（不包含该时间）。默认格式：`yyyy-MM-dd HH:mm:ss`。 | +| sort_files_by_modification_time | boolean | 否 | false | 是否按修改时间降序排序文件。启用此选项后，在读取不断演化的 schema 时可确保 schema 推断使用最新的文件。 | | common-options | | 否 | - | 源插件通用参数，请参阅 [Source Common Options](../common-options/source-common-options.md) | ## 如何创建 SMB 数据同步作业 diff --git a/seatunnel-connectors-v2/connector-file/connector-file-bos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/bos/source/BosFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-bos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/bos/source/BosFileSourceFactory.java index 9da0434b51..d3e90a1719 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-bos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/bos/source/BosFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-bos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/bos/source/BosFileSourceFactory.java @@ -106,6 +106,15 @@ public class BosFileSourceFactory implements TableSourceFactory { .optional(FileBaseSourceOptions.QUOTE_CHAR) .optional(FileBaseSourceOptions.ESCAPE_CHAR) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-bos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/bos/BosFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-bos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/bos/BosFileFactoryTest.java index 9071d46c97..6123a990b4 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-bos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/bos/BosFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-bos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/bos/BosFileFactoryTest.java @@ -17,8 +17,10 @@ package org.apache.seatunnel.connectors.seatunnel.file.bos; +import org.apache.seatunnel.api.configuration.util.OptionRule; import org.apache.seatunnel.connectors.seatunnel.file.bos.sink.BosFileSinkFactory; import org.apache.seatunnel.connectors.seatunnel.file.bos.source.BosFileSourceFactory; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -30,4 +32,33 @@ public class BosFileFactoryTest { Assertions.assertNotNull((new BosFileSourceFactory()).optionRule()); Assertions.assertNotNull((new BosFileSinkFactory()).optionRule()); } + + @Test + void consumedOptionsAreDeclared() { + OptionRule optionRule = (new BosFileSourceFactory()).optionRule(); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE)); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)); + } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java index 26ca989daa..fe6422e75c 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java @@ -106,6 +106,15 @@ public class CosFileSourceFactory implements TableSourceFactory { .optional(FileBaseSourceOptions.QUOTE_CHAR) .optional(FileBaseSourceOptions.ESCAPE_CHAR) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/cos/CosFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/cos/CosFileFactoryTest.java index 6691f5b1f2..2ebf19a486 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-cos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/cos/CosFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-cos/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/cos/CosFileFactoryTest.java @@ -17,6 +17,8 @@ package org.apache.seatunnel.connectors.seatunnel.file.cos; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; import org.apache.seatunnel.connectors.seatunnel.file.cos.sink.CosFileSinkFactory; import org.apache.seatunnel.connectors.seatunnel.file.cos.source.CosFileSourceFactory; @@ -30,4 +32,33 @@ public class CosFileFactoryTest { Assertions.assertNotNull((new CosFileSourceFactory()).optionRule()); Assertions.assertNotNull((new CosFileSinkFactory()).optionRule()); } + + @Test + void consumedOptionsAreDeclared() { + OptionRule optionRule = (new CosFileSourceFactory()).optionRule(); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE)); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)); + } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java index 269abc69a8..e965ffca14 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java @@ -143,6 +143,15 @@ public class FtpFileSourceFactory implements TableSourceFactory { FilePostSyncAction.BACKUP, FileBaseSourceOptions.BACKUP_PATH) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java index 82c3229843..9c14684786 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/FtpFileFactoryTest.java @@ -108,6 +108,35 @@ class FtpFileFactoryTest { Assertions.assertDoesNotThrow(() -> validate(backupConfig, optionRule)); } + @Test + void consumedOptionsAreDeclared() { + OptionRule optionRule = (new FtpFileSourceFactory()).optionRule(); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE)); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)); + } + private static Map<String, Object> sourceConfig() { Map<String, Object> config = new HashMap<>(); config.put(FileBaseOptions.FILE_PATH.key(), "/source"); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java index 9f50be47ac..945cb2c03f 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/source/GcsFileSourceFactory.java @@ -154,6 +154,15 @@ public class GcsFileSourceFactory implements TableSourceFactory { FilePostSyncAction.BACKUP, FileBaseSourceOptions.BACKUP_PATH) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java index 677b3f0ebd..843066c922 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-gcs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/gcs/GcsFileSourceFactoryTest.java @@ -73,6 +73,34 @@ class GcsFileSourceFactoryTest { assertTrue(optionRuleContains(optionRule, FileBaseSourceOptions.FILE_SPLIT_SIZE)); } + @Test + void consumedOptionsAreDeclared() { + OptionRule optionRule = new GcsFileSourceFactory().optionRule(); + assertTrue(optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS)); + assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE)); + assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE)); + assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)); + assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START)); + assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END)); + assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)); + } + private static Map<String, Object> sourceConfig() { Map<String, Object> config = new HashMap<>(); config.put(GcsFileSourceOptions.FILE_PATH.key(), "/source"); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java index e3bc352336..b90a7e8aeb 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java @@ -155,6 +155,15 @@ public class HdfsFileSourceFactory implements TableSourceFactory { .optional(HdfsFileSourceOptions.ESCAPE_CHAR) .optional(ConnectorCommonOptions.METALAKE_TYPE) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java index f52d8abe22..13098145a8 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-hadoop/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/hdfs/HdfsFileFactoryTest.java @@ -125,6 +125,35 @@ class HdfsFileFactoryTest { Assertions.assertDoesNotThrow(() -> validate(config, optionRule)); } + @Test + void consumedOptionsAreDeclared() { + OptionRule optionRule = (new HdfsFileSourceFactory()).optionRule(); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE)); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)); + } + private static Map<String, Object> sourceConfig() { Map<String, Object> config = new HashMap<>(); config.put(FileBaseOptions.FILE_PATH.key(), "/source"); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java index c833d2b041..f94430a48f 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/jindo/source/OssFileSourceFactory.java @@ -87,6 +87,7 @@ public class OssFileSourceFactory implements TableSourceFactory { .optional(FileBaseSourceOptions.TIME_FORMAT_LEGACY) .optional(FileBaseSourceOptions.FILE_FILTER_PATTERN) .optional(FileBaseSourceOptions.COMPRESS_CODEC) + .optional(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC) .optional(FileBaseSourceOptions.NULL_FORMAT) .optional(FileBaseSourceOptions.FILENAME_EXTENSION) .optional(FileBaseSourceOptions.READ_COLUMNS) @@ -105,6 +106,15 @@ public class OssFileSourceFactory implements TableSourceFactory { .optional(FileBaseSourceOptions.QUOTE_CHAR) .optional(FileBaseSourceOptions.ESCAPE_CHAR) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/test/java/org/apache/seatunnel/connectors/test/OssJindoFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/test/java/org/apache/seatunnel/connectors/test/OssJindoFactoryTest.java index c014f1d492..f5984d9bec 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/test/java/org/apache/seatunnel/connectors/test/OssJindoFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-jindo-oss/src/test/java/org/apache/seatunnel/connectors/test/OssJindoFactoryTest.java @@ -17,16 +17,68 @@ package org.apache.seatunnel.connectors.test; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseOptions; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; import org.apache.seatunnel.connectors.seatunnel.file.oss.jindo.sink.OssFileSinkFactory; import org.apache.seatunnel.connectors.seatunnel.file.oss.jindo.source.OssFileSourceFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; + public class OssJindoFactoryTest { @Test public void testOptionRule() { Assertions.assertNotNull((new OssFileSourceFactory()).optionRule()); Assertions.assertNotNull((new OssFileSinkFactory()).optionRule()); } + + @Test + void consumedOptionsAcceptedByValidateUnknownKeys() { + // These options are consumed by AbstractReadStrategy / CsvReadStrategy / + // BinaryReadStrategy at runtime. They MUST be declared in the factory's + // optionRule() so that ConfigValidator.validateUnknownKeys (called by + // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not + // reject them as "unknown option keys". + Map<String, Object> consumedOptions = new LinkedHashMap<>(); + consumedOptions.put( + FileBaseSourceOptions.READ_PARTITIONS.key(), + Collections.singletonList("year=2024")); + consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(), true); + consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(), 2048); + consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(), true); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(), "2024-01-01 00:00:00"); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(), "2024-12-31 23:59:59"); + consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true); + consumedOptions.put(FileBaseSourceOptions.COMPRESS_CODEC.key(), "gzip"); + consumedOptions.put(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC.key(), "zip"); + + OptionRule optionRule = (new OssFileSourceFactory()).optionRule(); + + for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) { + Map<String, Object> config = sourceConfig(); + config.put(entry.getKey(), entry.getValue()); + + Assertions.assertDoesNotThrow( + () -> + ConfigValidator.validateUnknownKeys( + ReadonlyConfig.fromMap(config), optionRule, "OssFile"), + "validateUnknownKeys should accept declared option: " + entry.getKey()); + } + } + + private static Map<String, Object> sourceConfig() { + Map<String, Object> config = new HashMap<>(); + config.put(FileBaseOptions.FILE_PATH.key(), "/source"); + return config; + } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java index de77146f4a..b6ef0900ba 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-local/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/local/source/LocalFileSourceFactory.java @@ -150,6 +150,15 @@ public class LocalFileSourceFactory implements TableSourceFactory { FilePostSyncAction.BACKUP, FileBaseSourceOptions.BACKUP_PATH) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java index c4c40323f9..20f32463c8 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-local/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/local/LocalFileFactoryTest.java @@ -30,7 +30,9 @@ import org.apache.seatunnel.connectors.seatunnel.file.local.source.LocalFileSour import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; class LocalFileFactoryTest { @@ -118,6 +120,40 @@ class LocalFileFactoryTest { Assertions.assertDoesNotThrow(() -> validate(backupConfig, optionRule)); } + @Test + void consumedOptionsAcceptedByValidateUnknownKeys() { + // These 7 options are consumed by AbstractReadStrategy / CsvReadStrategy / + // BinaryReadStrategy at runtime. They MUST be declared in the factory's + // optionRule() so that ConfigValidator.validateUnknownKeys (called by + // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not + // reject them as "unknown option keys". + Map<String, Object> consumedOptions = new LinkedHashMap<>(); + consumedOptions.put( + FileBaseSourceOptions.READ_PARTITIONS.key(), + Collections.singletonList("year=2024")); + consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(), true); + consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(), 2048); + consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(), true); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(), "2024-01-01 00:00:00"); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(), "2024-12-31 23:59:59"); + consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true); + + OptionRule optionRule = (new LocalFileSourceFactory()).optionRule(); + + for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) { + Map<String, Object> config = sourceConfig(); + config.put(entry.getKey(), entry.getValue()); + + Assertions.assertDoesNotThrow( + () -> + ConfigValidator.validateUnknownKeys( + ReadonlyConfig.fromMap(config), optionRule, "LocalFile"), + "validateUnknownKeys should accept declared option: " + entry.getKey()); + } + } + private static Map<String, Object> sourceConfig() { Map<String, Object> config = new HashMap<>(); config.put(FileBaseOptions.FILE_PATH.key(), "/source"); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java index 55e068f5d8..23876525b3 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/obs/source/ObsFileSourceFactory.java @@ -76,6 +76,8 @@ public class ObsFileSourceFactory implements TableSourceFactory { .optional(FileBaseSourceOptions.DATE_FORMAT_LEGACY) .optional(FileBaseSourceOptions.DATETIME_FORMAT_LEGACY) .optional(FileBaseSourceOptions.TIME_FORMAT_LEGACY) + .optional(FileBaseSourceOptions.COMPRESS_CODEC) + .optional(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC) .optional(FileBaseSourceOptions.NULL_FORMAT) .optional(FileBaseSourceOptions.FILENAME_EXTENSION) .optional(FileBaseSourceOptions.READ_COLUMNS) @@ -94,6 +96,15 @@ public class ObsFileSourceFactory implements TableSourceFactory { .optional(FileBaseSourceOptions.QUOTE_CHAR) .optional(FileBaseSourceOptions.ESCAPE_CHAR) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/obs/ObsFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/obs/ObsFileFactoryTest.java index dc7c34525a..d87df875ac 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-obs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/obs/ObsFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-obs/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/obs/ObsFileFactoryTest.java @@ -17,17 +17,70 @@ package org.apache.seatunnel.connectors.seatunnel.file.obs; +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.configuration.util.ConfigValidator; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseOptions; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; import org.apache.seatunnel.connectors.seatunnel.file.obs.sink.ObsFileSinkFactory; import org.apache.seatunnel.connectors.seatunnel.file.obs.source.ObsFileSourceFactory; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; + public class ObsFileFactoryTest { @Test void optionRule() { - Assertions.assertNotNull((new ObsFileSourceFactory()).optionRule()); + OptionRule optionRule = (new ObsFileSourceFactory()).optionRule(); + Assertions.assertNotNull(optionRule); Assertions.assertNotNull((new ObsFileSinkFactory()).optionRule()); } + + @Test + void consumedOptionsAcceptedByValidateUnknownKeys() { + // These options are consumed by AbstractReadStrategy / CsvReadStrategy / + // BinaryReadStrategy at runtime. They MUST be declared in the factory's + // optionRule() so that ConfigValidator.validateUnknownKeys (called by + // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not + // reject them as "unknown option keys". + Map<String, Object> consumedOptions = new LinkedHashMap<>(); + consumedOptions.put( + FileBaseSourceOptions.READ_PARTITIONS.key(), + Collections.singletonList("year=2024")); + consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(), true); + consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(), 2048); + consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(), true); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(), "2024-01-01 00:00:00"); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(), "2024-12-31 23:59:59"); + consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true); + consumedOptions.put(FileBaseSourceOptions.COMPRESS_CODEC.key(), "gzip"); + consumedOptions.put(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC.key(), "zip"); + + OptionRule optionRule = (new ObsFileSourceFactory()).optionRule(); + + for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) { + Map<String, Object> config = sourceConfig(); + config.put(entry.getKey(), entry.getValue()); + + Assertions.assertDoesNotThrow( + () -> + ConfigValidator.validateUnknownKeys( + ReadonlyConfig.fromMap(config), optionRule, "ObsFile"), + "validateUnknownKeys should accept declared option: " + entry.getKey()); + } + } + + private static Map<String, Object> sourceConfig() { + Map<String, Object> config = new HashMap<>(); + config.put(FileBaseOptions.FILE_PATH.key(), "/source"); + return config; + } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java index e664d1f82d..634e0ce627 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java @@ -141,6 +141,15 @@ public class OssFileSourceFactory implements TableSourceFactory { FilePostSyncAction.BACKUP, FileBaseSourceOptions.BACKUP_PATH) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java index fe7b37c4d2..0c6042fd72 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-oss/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/oss/OssFileFactoryTest.java @@ -17,6 +17,8 @@ package org.apache.seatunnel.connectors.seatunnel.file.oss; +import org.apache.seatunnel.api.configuration.util.OptionRule; +import org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions; import org.apache.seatunnel.connectors.seatunnel.file.oss.sink.OssFileSinkFactory; import org.apache.seatunnel.connectors.seatunnel.file.oss.source.OssFileSourceFactory; @@ -30,4 +32,33 @@ public class OssFileFactoryTest { Assertions.assertNotNull((new OssFileSourceFactory()).optionRule()); Assertions.assertNotNull((new OssFileSinkFactory()).optionRule()); } + + @Test + void consumedOptionsAreDeclared() { + OptionRule optionRule = (new OssFileSourceFactory()).optionRule(); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE)); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)); + } } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java index 02a85e7cd4..03cb29e75c 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/s3/source/S3FileSourceFactory.java @@ -174,6 +174,15 @@ public class S3FileSourceFactory implements TableSourceFactory, SupportSourceDry FilePostSyncAction.BACKUP, FileBaseSourceOptions.BACKUP_PATH) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java index 38d21352d6..3ff80a2184 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-s3/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/s3/S3FileFactoryTest.java @@ -126,6 +126,35 @@ class S3FileFactoryTest { Assertions.assertDoesNotThrow(() -> validate(config, optionRule)); } + @Test + void consumedOptionsAreDeclared() { + OptionRule optionRule = new S3FileSourceFactory().optionRule(); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.READ_PARTITIONS)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.CSV_USE_HEADER_LINE)); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.BINARY_CHUNK_SIZE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_START)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.FILE_FILTER_MODIFIED_END)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME)); + } + private static Map<String, Object> sinkConfig() { Map<String, Object> config = new HashMap<>(); config.put(S3FileBaseOptions.FILE_PATH.key(), "/tmp/seatunnel"); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java index 160b260dc5..768c8f75ed 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/source/SftpFileSourceFactory.java @@ -88,6 +88,8 @@ public class SftpFileSourceFactory implements TableSourceFactory { .optional(FileBaseSourceOptions.DATETIME_FORMAT_LEGACY) .optional(FileBaseSourceOptions.TIME_FORMAT_LEGACY) .optional(FileBaseSourceOptions.FILE_FILTER_PATTERN) + .optional(FileBaseSourceOptions.COMPRESS_CODEC) + .optional(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC) .optional(FileBaseSourceOptions.NULL_FORMAT) .optional(FileBaseSourceOptions.FILENAME_EXTENSION) .optional(FileBaseSourceOptions.READ_COLUMNS) @@ -131,6 +133,15 @@ public class SftpFileSourceFactory implements TableSourceFactory { FilePostSyncAction.BACKUP, FileBaseSourceOptions.BACKUP_PATH) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java index 93d31661ea..272523b688 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/SftpFileFactoryTest.java @@ -32,7 +32,9 @@ import org.apache.seatunnel.connectors.seatunnel.file.sftp.system.SFTPFileSystem import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; class SftpFileFactoryTest { @@ -73,6 +75,12 @@ class SftpFileFactoryTest { optionRule .getOptionalOptions() .contains(FileBaseSourceOptions.RETENTION_CHECK_INTERVAL)); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.COMPRESS_CODEC)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC)); Assertions.assertTrue( optionRule.getOptionalOptions().stream() .anyMatch(option -> "keyfile".equals(option.key()))); @@ -134,6 +142,42 @@ class SftpFileFactoryTest { hadoopConf.getExtraOptions().get(SFTPFileSystem.FS_SFTP_KEYFILE)); } + @Test + void consumedOptionsAcceptedByValidateUnknownKeys() { + // These options are consumed by AbstractReadStrategy / CsvReadStrategy / + // BinaryReadStrategy at runtime. They MUST be declared in the factory's + // optionRule() so that ConfigValidator.validateUnknownKeys (called by + // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not + // reject them as "unknown option keys". + Map<String, Object> consumedOptions = new LinkedHashMap<>(); + consumedOptions.put( + FileBaseSourceOptions.READ_PARTITIONS.key(), + Collections.singletonList("year=2024")); + consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(), true); + consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(), 2048); + consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(), true); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(), "2024-01-01 00:00:00"); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(), "2024-12-31 23:59:59"); + consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true); + consumedOptions.put(FileBaseSourceOptions.COMPRESS_CODEC.key(), "gzip"); + consumedOptions.put(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC.key(), "zip"); + + OptionRule optionRule = (new SftpFileSourceFactory()).optionRule(); + + for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) { + Map<String, Object> config = sourceConfig(); + config.put(entry.getKey(), entry.getValue()); + + Assertions.assertDoesNotThrow( + () -> + ConfigValidator.validateUnknownKeys( + ReadonlyConfig.fromMap(config), optionRule, "SftpFile"), + "validateUnknownKeys should accept declared option: " + entry.getKey()); + } + } + private static Map<String, Object> sourceConfig() { Map<String, Object> config = new HashMap<>(); config.put(FileBaseOptions.FILE_PATH.key(), "/source"); diff --git a/seatunnel-connectors-v2/connector-file/connector-file-smb/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/smb/source/SmbFileSourceFactory.java b/seatunnel-connectors-v2/connector-file/connector-file-smb/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/smb/source/SmbFileSourceFactory.java index 7a7281a77d..c109039f83 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-smb/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/smb/source/SmbFileSourceFactory.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-smb/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/smb/source/SmbFileSourceFactory.java @@ -89,6 +89,8 @@ public class SmbFileSourceFactory implements TableSourceFactory { .optional(FileBaseSourceOptions.DATETIME_FORMAT_LEGACY) .optional(FileBaseSourceOptions.TIME_FORMAT_LEGACY) .optional(FileBaseSourceOptions.FILE_FILTER_PATTERN) + .optional(FileBaseSourceOptions.COMPRESS_CODEC) + .optional(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC) .optional(FileBaseSourceOptions.NULL_FORMAT) .optional(FileBaseSourceOptions.FILENAME_EXTENSION) .optional(FileBaseSourceOptions.READ_COLUMNS) @@ -132,6 +134,15 @@ public class SmbFileSourceFactory implements TableSourceFactory { FilePostSyncAction.BACKUP, FileBaseSourceOptions.BACKUP_PATH) .optional(FileBaseSourceOptions.RECURSIVE_FILE_SCAN) + .optional( + FileBaseSourceOptions.READ_PARTITIONS, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START, + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END, + FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME) + .optional( + FileBaseSourceOptions.CSV_USE_HEADER_LINE, + FileBaseSourceOptions.BINARY_CHUNK_SIZE, + FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE) .build(); } diff --git a/seatunnel-connectors-v2/connector-file/connector-file-smb/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/smb/SmbFileFactoryTest.java b/seatunnel-connectors-v2/connector-file/connector-file-smb/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/smb/SmbFileFactoryTest.java index 635174b771..6f50d8651f 100644 --- a/seatunnel-connectors-v2/connector-file/connector-file-smb/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/smb/SmbFileFactoryTest.java +++ b/seatunnel-connectors-v2/connector-file/connector-file-smb/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/smb/SmbFileFactoryTest.java @@ -34,7 +34,9 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -76,6 +78,12 @@ class SmbFileFactoryTest { optionRule .getOptionalOptions() .contains(FileBaseSourceOptions.RETENTION_CHECK_INTERVAL)); + Assertions.assertTrue( + optionRule.getOptionalOptions().contains(FileBaseSourceOptions.COMPRESS_CODEC)); + Assertions.assertTrue( + optionRule + .getOptionalOptions() + .contains(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC)); } @Test @@ -224,6 +232,42 @@ class SmbFileFactoryTest { Assertions.assertEquals("smb", hadoopConf.getSchema()); } + @Test + void consumedOptionsAcceptedByValidateUnknownKeys() { + // These options are consumed by AbstractReadStrategy / CsvReadStrategy / + // BinaryReadStrategy at runtime. They MUST be declared in the factory's + // optionRule() so that ConfigValidator.validateUnknownKeys (called by + // --check / --dry-run at SeaTunnelConfValidateCommand:325) does not + // reject them as "unknown option keys". + Map<String, Object> consumedOptions = new LinkedHashMap<>(); + consumedOptions.put( + FileBaseSourceOptions.READ_PARTITIONS.key(), + Collections.singletonList("year=2024")); + consumedOptions.put(FileBaseSourceOptions.CSV_USE_HEADER_LINE.key(), true); + consumedOptions.put(FileBaseSourceOptions.BINARY_CHUNK_SIZE.key(), 2048); + consumedOptions.put(FileBaseSourceOptions.BINARY_COMPLETE_FILE_MODE.key(), true); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_START.key(), "2024-01-01 00:00:00"); + consumedOptions.put( + FileBaseSourceOptions.FILE_FILTER_MODIFIED_END.key(), "2024-12-31 23:59:59"); + consumedOptions.put(FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.key(), true); + consumedOptions.put(FileBaseSourceOptions.COMPRESS_CODEC.key(), "gzip"); + consumedOptions.put(FileBaseSourceOptions.ARCHIVE_COMPRESS_CODEC.key(), "zip"); + + OptionRule optionRule = (new SmbFileSourceFactory()).optionRule(); + + for (Map.Entry<String, Object> entry : consumedOptions.entrySet()) { + Map<String, Object> config = sourceConfig(); + config.put(entry.getKey(), entry.getValue()); + + Assertions.assertDoesNotThrow( + () -> + ConfigValidator.validateUnknownKeys( + ReadonlyConfig.fromMap(config), optionRule, "SmbFile"), + "validateUnknownKeys should accept declared option: " + entry.getKey()); + } + } + private static Map<String, Object> sourceConfig() { Map<String, Object> config = new HashMap<>(); config.put(FileBaseOptions.FILE_PATH.key(), "/source");
