This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 6de6f1b953 [Improve][Connector-V2][File] Optimize file discovery and
update comparison for sync_mode=update (#11312)
6de6f1b953 is described below
commit 6de6f1b95365a99919ca5fdae622b22132a2050d
Author: yzeng1618 <[email protected]>
AuthorDate: Fri Jul 24 19:58:11 2026 +0800
[Improve][Connector-V2][File] Optimize file discovery and update comparison
for sync_mode=update (#11312)
Co-authored-by: zengyi <[email protected]>
---
docs/en/connectors/source/FtpFile.md | 10 +
docs/en/connectors/source/HdfsFile.md | 10 +
docs/en/connectors/source/LocalFile.md | 10 +
docs/en/connectors/source/SftpFile.md | 10 +
docs/zh/connectors/source/FtpFile.md | 10 +
docs/zh/connectors/source/HdfsFile.md | 10 +
docs/zh/connectors/source/LocalFile.md | 10 +
docs/zh/connectors/source/SftpFile.md | 10 +
.../file/config/FileBaseSourceOptions.java | 18 ++
.../file/hadoop/FileStatusListingSession.java | 41 +++
.../file/hadoop/HadoopFileSystemProxy.java | 55 ++++
.../seatunnel/file/hadoop/StreamingFileSystem.java | 27 ++
.../file/source/reader/AbstractReadStrategy.java | 234 ++++++++++------
.../file/source/reader/FileDiscoveryScanner.java | 144 ++++++++++
.../source/reader/UpdateFileMetadataLoader.java | 295 +++++++++++++++++++++
.../source/reader/AbstractReadStrategyTest.java | 18 ++
.../source/reader/FileDiscoveryScannerTest.java | 166 ++++++++++++
.../source/reader/UpdateCompareConfigTest.java | 86 ++++++
.../reader/UpdateFileMetadataLoaderTest.java | 230 ++++++++++++++++
.../file/ftp/source/FtpFileSourceFactory.java | 4 +-
.../file/ftp/system/SeaTunnelFTPFileSystem.java | 101 ++++++-
.../seatunnel/file/ftp/FtpFileFactoryTest.java | 8 +
.../ftp/system/SeaTunnelFTPFileSystemTest.java | 32 +++
.../file/hdfs/source/HdfsFileSourceFactory.java | 4 +-
.../seatunnel/file/hdfs/HdfsFileFactoryTest.java | 8 +
.../file/local/source/LocalFileSourceFactory.java | 4 +-
.../seatunnel/file/local/LocalFileFactoryTest.java | 8 +
.../file/sftp/source/SftpFileSourceFactory.java | 4 +-
.../seatunnel/file/sftp/system/SFTPFileSystem.java | 94 ++++++-
.../seatunnel/file/sftp/SftpFileFactoryTest.java | 8 +
30 files changed, 1571 insertions(+), 98 deletions(-)
diff --git a/docs/en/connectors/source/FtpFile.md
b/docs/en/connectors/source/FtpFile.md
index 9da83a14ea..f89d8a87bc 100644
--- a/docs/en/connectors/source/FtpFile.md
+++ b/docs/en/connectors/source/FtpFile.md
@@ -88,6 +88,8 @@ If you use SeaTunnel Engine, It automatically integrated the
hadoop jar when you
| target_hadoop_conf | map | no | -
|
| update_strategy | string | no | distcp
|
| compare_mode | string | no | len_mtime
|
+| update_compare_parallelism | int | no | 8
|
+| update_compare_bulk_threshold | int | no | 0
|
| common-options | | no | -
|
| file_filter_modified_start | string | no | -
|
| file_filter_modified_end | string | no | -
|
@@ -545,6 +547,14 @@ Only used when `sync_mode=update`. Supported values:
`distcp` (default), `strict
Only used when `sync_mode=update`. Supported values: `len_mtime` (default),
`checksum` (only valid when `update_strategy=strict`).
+### update_compare_parallelism [int]
+
+Maximum parallelism for sparse target metadata lookups in `sync_mode=update`.
The default is `8`; valid values are `1` through `64`; values outside this
range are rejected during configuration validation. The maximum number of
submitted-but-incomplete lookups is eight times this value.
+
+### update_compare_bulk_threshold [int]
+
+A positive value switches comparison to one directory listing when the
candidate count under a target parent reaches the threshold. The default `0`
disables automatic bulk listing and uses bounded point lookups, avoiding an
unexpectedly expensive target directory scan. This behavior applies to all
target filesystems. Source filters are applied while entries are listed to
reduce peak metadata memory.
+
### file_filter_modified_start [string]
File modification time filter. The connector will filter some files base on
the last modification start time (include start time). The default data format
is `yyyy-MM-dd HH:mm:ss`.
diff --git a/docs/en/connectors/source/HdfsFile.md
b/docs/en/connectors/source/HdfsFile.md
index fd28d44d68..12b051df1b 100644
--- a/docs/en/connectors/source/HdfsFile.md
+++ b/docs/en/connectors/source/HdfsFile.md
@@ -90,6 +90,8 @@ Read data from hdfs file system.
| target_hadoop_conf | map | no | -
| Only used when `sync_mode=update`. Extra Hadoop configuration for target
filesystem. You can set `fs.defaultFS` in this map to override target
defaultFS.
|
| update_strategy | string | no | distcp
| Only used when `sync_mode=update`. Supported values: `distcp` (default),
`strict`.
|
| compare_mode | string | no | len_mtime
| Only used when `sync_mode=update`. Supported values: `len_mtime` (default),
`checksum` (only valid when `update_strategy=strict`).
|
+| update_compare_parallelism | int | no | 8
| Maximum parallelism for sparse target metadata lookups. Valid range: `1-64`.
|
+| update_compare_bulk_threshold | int | no | 0
| A positive value switches comparison to one directory listing when the
candidate count under one target parent reaches the threshold. `0` disables
automatic bulk listing.
|
| common-options | | no | -
| Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
|
| file_filter_modified_start | string | no | -
| File modification time filter. The connector will filter some files base on
the last modification start time (include start time). The default data format
is `yyyy-MM-dd HH:mm:ss`.
|
| file_filter_modified_end | string | no | -
| File modification time filter. The connector will filter some files base on
the last modification end time (not include end time). The default data format
is `yyyy-MM-dd HH:mm:ss`.
|
@@ -311,6 +313,14 @@ Only used when `sync_mode=update`. Supported values:
`len_mtime` (default), `che
- `len_mtime`: SKIP only when both `len` and `mtime` are equal, otherwise COPY.
- `checksum`: SKIP only when `len` is equal and Hadoop `getFileChecksum` is
equal, otherwise COPY (only valid when `update_strategy=strict`).
+### update_compare_parallelism [int]
+
+Maximum parallelism for sparse target metadata lookups in `sync_mode=update`.
The default is `8`; valid values are `1` through `64`; values outside this
range are rejected during configuration validation. The maximum number of
submitted-but-incomplete lookups is eight times this value.
+
+### update_compare_bulk_threshold [int]
+
+A positive value switches comparison to one directory listing when the
candidate count under a target parent reaches the threshold. The default `0`
disables automatic bulk listing and uses bounded point lookups, avoiding an
unexpectedly expensive target directory scan. This behavior applies to all
target filesystems. Source filters are applied while entries are listed to
reduce peak metadata memory.
+
### enable_file_split [boolean]
Turn on the file splitting function, the default is false. It can be selected
when the file type is csv, text, json, parquet and non-compressed format.
diff --git a/docs/en/connectors/source/LocalFile.md
b/docs/en/connectors/source/LocalFile.md
index abda9935fd..1832575c2a 100644
--- a/docs/en/connectors/source/LocalFile.md
+++ b/docs/en/connectors/source/LocalFile.md
@@ -85,6 +85,8 @@ If you use SeaTunnel Engine, It automatically integrated the
hadoop jar when you
| target_hadoop_conf | map | no | -
|
| update_strategy | string | no | distcp
|
| compare_mode | string | no | len_mtime
|
+| update_compare_parallelism | int | no | 8
|
+| update_compare_bulk_threshold | int | no | 0
|
| common-options | | no | -
|
| tables_configs | list | no | used to define a multiple
table task |
| file_filter_modified_start | string | no | -
|
@@ -513,6 +515,14 @@ Only used when `sync_mode=update`. Supported values:
`distcp` (default), `strict
Only used when `sync_mode=update`. Supported values: `len_mtime` (default),
`checksum` (only valid when `update_strategy=strict`).
+### update_compare_parallelism [int]
+
+Maximum parallelism for sparse target metadata lookups in `sync_mode=update`.
The default is `8`; valid values are `1` through `64`; values outside this
range are rejected during configuration validation. The maximum number of
submitted-but-incomplete lookups is eight times this value.
+
+### update_compare_bulk_threshold [int]
+
+A positive value switches comparison to one directory listing when the
candidate count under a target parent reaches the threshold. The default `0`
disables automatic bulk listing and uses bounded point lookups, avoiding an
unexpectedly expensive target directory scan. This behavior applies to all
target filesystems. Source filters are applied while entries are listed to
reduce peak metadata memory.
+
### file_filter_modified_start [string]
File modification time filter. The connector will filter some files base on
the last modification start time (include start time). The default data format
is `yyyy-MM-dd HH:mm:ss`.
diff --git a/docs/en/connectors/source/SftpFile.md
b/docs/en/connectors/source/SftpFile.md
index 1c64f11ec7..dd3ad14b61 100644
--- a/docs/en/connectors/source/SftpFile.md
+++ b/docs/en/connectors/source/SftpFile.md
@@ -117,6 +117,8 @@ The File does not have a specific type list, and we can
indicate which SeaTunnel
| target_hadoop_conf | map | no | -
| Only used when `sync_mode=update`. Extra Hadoop configuration for target
filesystem. You can set `fs.defaultFS` in this map to override target
defaultFS.
|
| update_strategy | string | no | distcp
| Only used when `sync_mode=update`. Supported values: `distcp` (default),
`strict`.
|
| compare_mode | string | no | len_mtime
| Only used when `sync_mode=update`. Supported values: `len_mtime`
(default), `checksum` (only valid when `update_strategy=strict`).
|
+| update_compare_parallelism | int | no | 8
| Maximum parallelism for sparse target metadata lookups. Valid range:
`1-64`.
|
+| update_compare_bulk_threshold | int | no | 0
| A positive value switches comparison to one directory listing when the
candidate count under one target parent reaches the threshold. `0` disables
automatic bulk listing.
|
| common-options | | No | -
| Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
|
| file_filter_modified_start | string | no | -
| File modification time filter. The connector will filter some files base
on the last modification start time (include start time). The default data
format is `yyyy-MM-dd HH:mm:ss`.
|
| file_filter_modified_end | string | no | -
| File modification time filter. The connector will filter some files base
on the last modification end time (not include end time). The default data
format is `yyyy-MM-dd HH:mm:ss`.
|
@@ -385,6 +387,14 @@ Only used when `sync_mode=update`. Supported values:
`distcp` (default), `strict
Only used when `sync_mode=update`. Supported values: `len_mtime` (default),
`checksum` (only valid when `update_strategy=strict`).
+### update_compare_parallelism [int]
+
+Maximum parallelism for sparse target metadata lookups in `sync_mode=update`.
The default is `8`; valid values are `1` through `64`; values outside this
range are rejected during configuration validation. The maximum number of
submitted-but-incomplete lookups is eight times this value.
+
+### update_compare_bulk_threshold [int]
+
+A positive value switches comparison to one directory listing when the
candidate count under a target parent reaches the threshold. The default `0`
disables automatic bulk listing and uses bounded point lookups, avoiding an
unexpectedly expensive target directory scan. This behavior applies to all
target filesystems. Source filters are applied while entries are listed to
reduce peak metadata memory. A flat SFTP directory must still return all
entries because SFTP `READDIR` has no server- [...]
+
### quote_char [string]
A single character that encloses CSV fields, allowing fields with commas, line
breaks, or quotes to be read correctly.
diff --git a/docs/zh/connectors/source/FtpFile.md
b/docs/zh/connectors/source/FtpFile.md
index 74c6c81af9..d62be8eb62 100644
--- a/docs/zh/connectors/source/FtpFile.md
+++ b/docs/zh/connectors/source/FtpFile.md
@@ -85,6 +85,8 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
| target_hadoop_conf | map | 否 | - |
| update_strategy | string | 否 | distcp |
| compare_mode | string | 否 | len_mtime |
+| update_compare_parallelism | int | 否 | 8 |
+| update_compare_bulk_threshold | int | 否 | 0 |
| common-options | | 否 | - |
| file_filter_modified_start | string | 否 | - |
| file_filter_modified_end | string | 否 | - |
@@ -516,6 +518,14 @@ compare_mode = "len_mtime"
仅在 `sync_mode=update` 时使用。支持：`len_mtime`（默认）、`checksum`（仅在
`update_strategy=strict` 时可用）。
+### update_compare_parallelism [int]
+
+`sync_mode=update` 对稀疏目标文件执行元数据点查时的最大并发数。默认值为 `8`，有效范围为 `1` 到
`64`；范围外的值会在配置校验阶段被拒绝；已提交但未完成的任务上限为该值的 8 倍。
+
+### update_compare_bulk_threshold [int]
+
+设置正数后，同一目标父目录的候选文件达到该阈值时，SeaTunnel 改为只枚举一次目标目录。默认值 `0`
表示关闭自动批量枚举，使用有界并发点查，避免意外扫描庞大的目标目录。该行为适用于所有目标文件系统。源端会边枚举边过滤，以降低元数据峰值内存。
+
### file_filter_modified_start
按照最后修改时间过滤文件。 要过滤的开始时间(包括改时间),时间格式是：`yyyy-MM-dd HH:mm:ss`。
diff --git a/docs/zh/connectors/source/HdfsFile.md
b/docs/zh/connectors/source/HdfsFile.md
index f6a9d1126b..f81b575d59 100644
--- a/docs/zh/connectors/source/HdfsFile.md
+++ b/docs/zh/connectors/source/HdfsFile.md
@@ -90,6 +90,8 @@ import ChangeLog from '../changelog/connector-file-hadoop.md';
| target_hadoop_conf | map | 否 | - | 仅在
`sync_mode=update` 时使用。目标端 Hadoop 配置（可选），可在其中设置 `fs.defaultFS` 覆盖目标 defaultFS。
|
| update_strategy | string | 否 | distcp | 仅在
`sync_mode=update` 时使用。支持：`distcp`（默认）、`strict`。
|
| compare_mode | string | 否 | len_mtime | 仅在
`sync_mode=update` 时使用。支持：`len_mtime`（默认）、`checksum`（仅在
`update_strategy=strict` 时可用）。
|
+| update_compare_parallelism | int | 否 | 8 |
稀疏目标元数据点查的最大并发数，有效范围为 `1-64`。
|
+| update_compare_bulk_threshold | int | 否 | 0 |
设置正数后，同一目标父目录达到该候选数时切换为单次目录枚举；`0` 表示关闭自动批量枚举。
|
| common-options | | 否 | - |
数据源插件通用参数，请参阅 [数据源通用选项](../common-options/source-common-options.md) 了解详情。
|
| file_filter_modified_start | string | 否 | - |
按照最后修改时间过滤文件。 要过滤的开始时间(包括改时间),时间格式是：`yyyy-MM-dd HH:mm:ss`
|
| file_filter_modified_end | string | 否 | - |
按照最后修改时间过滤文件。 要过滤的结束时间(不包括改时间),时间格式是：`yyyy-MM-dd HH:mm:ss`
|
@@ -311,6 +313,14 @@ abc.*
- `len_mtime`：`len` 与 `mtime` 都相同才 SKIP，否则 COPY。
- `checksum`：要求 `len` 相同且 Hadoop `getFileChecksum` 相同才 SKIP，否则 COPY（仅在
`update_strategy=strict` 时生效）。
+### update_compare_parallelism [int]
+
+`sync_mode=update` 对稀疏目标文件执行元数据点查时的最大并发数。默认值为 `8`，有效范围为 `1` 到
`64`；范围外的值会在配置校验阶段被拒绝；已提交但未完成的任务上限为该值的 8 倍。
+
+### update_compare_bulk_threshold [int]
+
+设置正数后，同一目标父目录的候选文件达到该阈值时，SeaTunnel 改为只枚举一次目标目录。默认值 `0`
表示关闭自动批量枚举，使用有界并发点查，避免意外扫描庞大的目标目录。该行为适用于所有目标文件系统。源端会边枚举边过滤，以降低元数据峰值内存。
+
### enable_file_split [boolean]
开启大文件拆分功能，默认 false。仅支持 `csv`/`text`/`json`/`parquet`
且非压缩格式（`compress_codec=none` 且 `archive_compress_codec=none`）。
diff --git a/docs/zh/connectors/source/LocalFile.md
b/docs/zh/connectors/source/LocalFile.md
index ca1031c3a4..25a7e44c3a 100644
--- a/docs/zh/connectors/source/LocalFile.md
+++ b/docs/zh/connectors/source/LocalFile.md
@@ -85,6 +85,8 @@ import ChangeLog from '../changelog/connector-file-local.md';
| target_hadoop_conf | map | 否 | - |
| update_strategy | string | 否 | distcp |
| compare_mode | string | 否 | len_mtime |
+| update_compare_parallelism | int | 否 | 8 |
+| update_compare_bulk_threshold | int | 否 | 0 |
| common-options | | 否 | - |
| tables_configs | list | 否 | 用于定义多表任务 |
| file_filter_modified_start | string | 否 | - |
@@ -513,6 +515,14 @@ compare_mode = "len_mtime"
仅在 `sync_mode=update` 时使用。支持：`len_mtime`（默认）、`checksum`（仅在
`update_strategy=strict` 时可用）。
+### update_compare_parallelism [int]
+
+`sync_mode=update` 对稀疏目标文件执行元数据点查时的最大并发数。默认值为 `8`，有效范围为 `1` 到
`64`；范围外的值会在配置校验阶段被拒绝；已提交但未完成的任务上限为该值的 8 倍。
+
+### update_compare_bulk_threshold [int]
+
+设置正数后，同一目标父目录的候选文件达到该阈值时，SeaTunnel 改为只枚举一次目标目录。默认值 `0`
表示关闭自动批量枚举，使用有界并发点查，避免意外扫描庞大的目标目录。该行为适用于所有目标文件系统。源端会边枚举边过滤，以降低元数据峰值内存。
+
### file_filter_modified_start
按照最后修改时间过滤文件。 要过滤的开始时间(包括改时间),时间格式是：`yyyy-MM-dd HH:mm:ss`。
diff --git a/docs/zh/connectors/source/SftpFile.md
b/docs/zh/connectors/source/SftpFile.md
index 555427c1c5..c833c7a71b 100644
--- a/docs/zh/connectors/source/SftpFile.md
+++ b/docs/zh/connectors/source/SftpFile.md
@@ -117,6 +117,8 @@ import ChangeLog from '../changelog/connector-file-sftp.md';
| target_hadoop_conf | map | 否 | - | 仅在
`sync_mode=update` 时使用。目标端 Hadoop 配置（可选），可在其中设置 `fs.defaultFS` 覆盖目标 defaultFS。
|
| update_strategy | string | 否 | distcp | 仅在
`sync_mode=update` 时使用。支持：`distcp`（默认）、`strict`。
|
| compare_mode | string | 否 | len_mtime | 仅在
`sync_mode=update` 时使用。支持：`len_mtime`（默认）、`checksum`（仅在
`update_strategy=strict` 时可用）。
|
+| update_compare_parallelism | int | 否 | 8 |
稀疏目标元数据点查的最大并发数，有效范围为 `1-64`。
|
+| update_compare_bulk_threshold | int | 否 | 0 |
设置正数后，同一目标父目录达到该候选数时切换为单次目录枚举；`0` 表示关闭自动批量枚举。
|
| common-options | | 否 | - |
数据源插件通用参数，请参考[数据源通用选项](../common-options/source-common-options.md)了解详情。
|
| file_filter_modified_start | string | 否 | - |
按照最后修改时间过滤文件。 要过滤的开始时间(包括改时间),时间格式是：`yyyy-MM-dd HH:mm:ss`
|
| file_filter_modified_end | string | 否 | - |
按照最后修改时间过滤文件。 要过滤的结束时间(不包括改时间),时间格式是：`yyyy-MM-dd HH:mm:ss`
|
@@ -385,6 +387,14 @@ compare_mode = "len_mtime"
仅在 `sync_mode=update` 时使用。支持：`len_mtime`（默认）、`checksum`（仅在
`update_strategy=strict` 时可用）。
+### update_compare_parallelism [int]
+
+`sync_mode=update` 对稀疏目标文件执行元数据点查时的最大并发数。默认值为 `8`，有效范围为 `1` 到
`64`；范围外的值会在配置校验阶段被拒绝；已提交但未完成的任务上限为该值的 8 倍。
+
+### update_compare_bulk_threshold [int]
+
+设置正数后，同一目标父目录的候选文件达到该阈值时，SeaTunnel 改为只枚举一次目标目录。默认值 `0`
表示关闭自动批量枚举，使用有界并发点查，避免意外扫描庞大的目标目录。该行为适用于所有目标文件系统。源端会边枚举边过滤，以降低元数据峰值内存；但 SFTP
平铺目录仍需返回全部条目，因为标准 `READDIR` 不支持服务端按修改时间过滤。
+
### schema [config]
#### fields [Config]
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseSourceOptions.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseSourceOptions.java
index 8828eef0c7..8a5229ae3a 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileBaseSourceOptions.java
@@ -211,6 +211,24 @@ public class FileBaseSourceOptions extends FileBaseOptions
{
.withDescription(
"Compare mode when sync_mode=update. Supported
values: len_mtime, checksum. "
+ "checksum uses Hadoop
FileSystem#getFileChecksum, only valid when update_strategy=strict.");
+
+ public static final Option<Integer> UPDATE_COMPARE_PARALLELISM =
+ Options.key("update_compare_parallelism")
+ .intType()
+ .defaultValue(8)
+ .withDescription(
+ "Maximum parallelism for sparse target metadata
lookups in sync_mode=update. "
+ + "The valid range is 1 to 64.");
+
+ public static final Option<Integer> UPDATE_COMPARE_BULK_THRESHOLD =
+ Options.key("update_compare_bulk_threshold")
+ .intType()
+ .defaultValue(0)
+ .withDescription(
+ "Number of candidates under one target parent
directory that switches "
+ + "sync_mode=update comparison from point
lookups to one directory listing. "
+ + "The default 0 disables automatic bulk
listing for non-FTP/SFTP targets; "
+ + "positive values enable it.");
public static final Option<String> QUOTE_CHAR =
Options.key("quote_char")
.stringType()
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileStatusListingSession.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileStatusListingSession.java
new file mode 100644
index 0000000000..a697989e2a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/FileStatusListingSession.java
@@ -0,0 +1,41 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.file.hadoop;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/** A closeable, reusable session that emits directory metadata without
materializing an array. */
+public interface FileStatusListingSession extends Closeable {
+
+ /** Returns metadata for the scan root without enumerating its parent
directory. */
+ FileStatus getFileStatus(Path path) throws IOException;
+
+ /** Emits direct children of one directory to the consumer as metadata
becomes available. */
+ void list(Path directory, FileStatusConsumer consumer) throws IOException;
+
+ /** A checked consumer used by protocol callbacks while listing directory
entries. */
+ @FunctionalInterface
+ interface FileStatusConsumer {
+ /** Handles one direct child from the active directory listing. */
+ void accept(FileStatus status) throws IOException;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java
index a6578ea2db..07997b62b6 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java
@@ -219,10 +219,60 @@ public class HadoopFileSystemProxy implements
Serializable, Closeable {
return execute(() -> getFileSystem().listStatus(new Path(filePath)));
}
+ /** Opens one reusable directory-listing session for a complete discovery
pass. */
+ public FileStatusListingSession openFileStatusListingSession() throws
IOException {
+ FileSystem fs = getFileSystem();
+ if (fs instanceof StreamingFileSystem) {
+ return ((StreamingFileSystem) fs).openFileStatusListingSession();
+ }
+ return new HadoopListingSession();
+ }
+
public FileStatus getFileStatus(String filePath) throws IOException {
return execute(() -> getFileSystem().getFileStatus(new
Path(filePath)));
}
+ private final class HadoopListingSession implements
FileStatusListingSession {
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ return execute(() -> getFileSystem().getFileStatus(path));
+ }
+
+ @Override
+ public void list(Path directory, FileStatusConsumer consumer) throws
IOException {
+ execute(
+ () -> {
+ FileSystem fs = getFileSystem();
+ int emitted = 0;
+ try {
+ RemoteIterator<? extends FileStatus> iterator;
+ if ("s3a".equalsIgnoreCase(fs.getScheme())) {
+ iterator = fs.listLocatedStatus(directory);
+ } else {
+ iterator = fs.listStatusIterator(directory);
+ }
+ while (iterator.hasNext()) {
+ consumer.accept(iterator.next());
+ emitted++;
+ }
+ } catch (UnsupportedOperationException e) {
+ if (emitted > 0) {
+ throw e;
+ }
+ for (FileStatus status : fs.listStatus(directory))
{
+ consumer.accept(status);
+ }
+ }
+ return Void.class;
+ });
+ }
+
+ @Override
+ public void close() {
+ // The proxy owns the Hadoop FileSystem lifecycle.
+ }
+ }
+
public FileChecksum getFileChecksum(String filePath) throws IOException {
return execute(() -> getFileSystem().getFileChecksum(new
Path(filePath)));
}
@@ -255,6 +305,11 @@ public class HadoopFileSystemProxy implements
Serializable, Closeable {
return fileSystem;
}
+ /** Returns the scheme of the initialized target or source file system. */
+ public String getScheme() {
+ return getFileSystem().getScheme();
+ }
+
@SneakyThrows
public <T> T doWithHadoopAuth(HadoopLoginFactory.LoginFunction<T>
loginFunction) {
if (configuration == null) {
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/StreamingFileSystem.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/StreamingFileSystem.java
new file mode 100644
index 0000000000..23a25975c2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/StreamingFileSystem.java
@@ -0,0 +1,27 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.file.hadoop;
+
+import java.io.IOException;
+
+/** Internal extension implemented by file systems with a native streaming
directory API. */
+public interface StreamingFileSystem {
+
+ /** Opens a session that is reused for one complete source or target
metadata scan. */
+ FileStatusListingSession openFileStatusListingSession() throws IOException;
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java
index 01cfb40e0a..ae1a17f32b 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java
@@ -40,6 +40,7 @@ import
org.apache.seatunnel.connectors.seatunnel.file.config.FileSyncMode;
import
org.apache.seatunnel.connectors.seatunnel.file.config.FileUpdateStrategy;
import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileStatusListingSession;
import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.HadoopFileSystemProxy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplit;
@@ -60,7 +61,6 @@ import lombok.extern.slf4j.Slf4j;
import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
-import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
@@ -127,6 +127,10 @@ public abstract class AbstractReadStrategy implements
ReadStrategy {
protected FileUpdateStrategy updateStrategy =
FileBaseSourceOptions.UPDATE_STRATEGY.defaultValue();
protected FileCompareMode compareMode =
FileBaseSourceOptions.COMPARE_MODE.defaultValue();
+ protected int updateCompareParallelism =
+ FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM.defaultValue();
+ protected int updateCompareBulkThreshold =
+ FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD.defaultValue();
protected Map<String, String> targetHadoopConf;
protected transient HadoopFileSystemProxy targetHadoopFileSystemProxy;
protected transient boolean shareTargetFileSystemProxy;
@@ -135,11 +139,6 @@ public abstract class AbstractReadStrategy implements
ReadStrategy {
protected boolean sortFilesByModTime =
FileBaseSourceOptions.SORT_FILES_BY_MOD_TIME.defaultValue();
- private static final class UpdateModeStats {
- private long scanned;
- private long skipped;
- }
-
@Override
public void init(HadoopConf conf) {
this.hadoopConf = conf;
@@ -162,86 +161,126 @@ public abstract class AbstractReadStrategy implements
ReadStrategy {
@Override
public List<String> getFileNamesByPath(String path) throws IOException {
- List<FileInfo> fileInfoList = new ArrayList<>();
- UpdateModeStats updateModeStats = enableUpdateSync ? new
UpdateModeStats() : null;
- collectFileInfoByPath(path, fileInfoList, updateModeStats);
+ List<FileStatus> candidates = new ArrayList<>();
+ FileDiscoveryScanner.ScanStats scanStats;
+ try (FileStatusListingSession session =
+ hadoopFileSystemProxy.openFileStatusListingSession()) {
+ scanStats =
+ FileDiscoveryScanner.scan(
+ new Path(path),
+ recursiveFileScan,
+ session,
+ this::isSourceCandidate,
+ candidates::add);
+ } catch (IOException e) {
+ Path sourcePath = new Path(path);
+ String protocol =
+ sourcePath.toUri().getScheme() == null
+ ? hadoopFileSystemProxy.getScheme()
+ : sourcePath.toUri().getScheme();
+ throw new IOException(
+ "Failed during source_listing setup/scan for protocol="
+ + protocol
+ + ", path="
+ + maskUriUserInfo(path),
+ e);
+ }
+
+ List<FileInfo> fileInfoList = new ArrayList<>(candidates.size());
+ UpdateFileMetadataLoader.Result updateResult = null;
+ long skipped = 0;
+ if (enableUpdateSync) {
+ if (targetHadoopFileSystemProxy == null) {
+ initTargetHadoopFileSystemProxy();
+ }
+ List<UpdateFileMetadataLoader.Request> requests = new
ArrayList<>(candidates.size());
+ for (int i = 0; i < candidates.size(); i++) {
+ String sourceFilePath = candidates.get(i).getPath().toString();
+ String relativePath = resolveRelativePath(sourceRootPath,
sourceFilePath);
+ requests.add(
+ new UpdateFileMetadataLoader.Request(
+ i, buildTargetFilePath(targetPath,
relativePath)));
+ }
+ updateResult =
+ UpdateFileMetadataLoader.load(
+ requests,
+ targetHadoopFileSystemProxy,
+ updateCompareParallelism,
+ updateCompareBulkThreshold);
+ for (int i = 0; i < candidates.size(); i++) {
+ FileStatus source = candidates.get(i);
+ FileStatus target = updateResult.getStatuses().get(i);
+ String targetFilePath = requests.get(i).getTargetPath();
+ if (shouldSyncFileInUpdateMode(source, target,
targetFilePath)) {
+ fileInfoList.add(
+ new FileInfo(
+ source.getPath().toString(),
source.getModificationTime()));
+ } else {
+ skipped++;
+ }
+ }
+ } else {
+ for (FileStatus candidate : candidates) {
+ fileInfoList.add(
+ new FileInfo(
+ candidate.getPath().toString(),
candidate.getModificationTime()));
+ }
+ }
// Sort by modification time in descending order if enabled
if (sortFilesByModTime) {
fileInfoList.sort(Comparator.comparingLong(FileInfo::getModifyTime).reversed());
}
- if (updateModeStats != null) {
- log.info(
- "Update sync mode statistics: scanned={}, skipped={},
to_sync={}",
- updateModeStats.scanned,
- updateModeStats.skipped,
- updateModeStats.scanned - updateModeStats.skipped);
- }
-
for (FileInfo fileInfo : fileInfoList) {
this.fileNames.add(fileInfo.getFileName());
}
+ log.info(
+ "File metadata pipeline statistics: phase=source_construction,
source_entries={}, filtered={}, candidates={}, source_directories={},
target_bulk_directories={}, target_point_lookups={}, peak_point_concurrency={},
peak_point_in_flight={}, skipped={}, to_sync={}, source_listing_ms={},
source_filtering_ms={}, target_bulk_listing_ms={}, target_point_lookup_ms={}",
+ scanStats.getSourceEntries(),
+ scanStats.getFiltered(),
+ scanStats.getCandidates(),
+ scanStats.getListedDirectories(),
+ updateResult == null ? 0 :
updateResult.getBulkListedDirectories(),
+ updateResult == null ? 0 : updateResult.getPointLookups(),
+ updateResult == null ? 0 : updateResult.getPeakConcurrency(),
+ updateResult == null ? 0 : updateResult.getPeakInFlight(),
+ skipped,
+ fileInfoList.size(),
+ scanStats.getListingNanos() / 1_000_000,
+ scanStats.getFilteringNanos() / 1_000_000,
+ updateResult == null ? 0 : updateResult.getBulkListingNanos()
/ 1_000_000,
+ updateResult == null ? 0 : updateResult.getPointLookupNanos()
/ 1_000_000);
+
return
fileInfoList.stream().map(FileInfo::getFileName).collect(Collectors.toList());
}
- private void collectFileInfoByPath(
- String path, List<FileInfo> fileInfoList, UpdateModeStats
updateModeStats)
- throws IOException {
- FileStatus[] stats = hadoopFileSystemProxy.listStatus(path);
- for (FileStatus fileStatus : stats) {
- if (fileStatus.isDirectory() && recursiveFileScan) {
- // skip hidden tmp directory, such as .hive-staging_hive
- if (!fileStatus.getPath().getName().startsWith(".")) {
- collectFileInfoByPath(
- fileStatus.getPath().toString(), fileInfoList,
updateModeStats);
- }
- continue;
- }
- if (!fileStatus.isFile()
- || !filterFileByPattern(fileStatus)
- || fileStatus.getLen() <= 0) {
- continue;
- }
-
- // filter '_SUCCESS' file and hidden files
- String fileName = fileStatus.getPath().getName();
- if (fileName.equals("_SUCCESS")
- || fileName.startsWith(".")
- || !filterFileByModificationDate(fileStatus)) {
- continue;
- }
-
- String filePath = fileStatus.getPath().toString();
- if (StringUtils.isNotEmpty(filenameExtension)
- && !filePath.endsWith(filenameExtension)) {
- continue;
- }
-
- if (!readPartitions.isEmpty()) {
- boolean partitionMatched = false;
- for (String readPartition : readPartitions) {
- if (filePath.contains(readPartition)) {
- partitionMatched = true;
- break;
- }
- }
- if (!partitionMatched) {
- continue;
+ private boolean isSourceCandidate(FileStatus fileStatus) {
+ String fileName = fileStatus.getPath().getName();
+ if (!fileStatus.isFile()
+ || fileName.equals("_SUCCESS")
+ || fileName.startsWith(".")
+ || fileStatus.getLen() <= 0
+ || !filterFileByModificationDate(fileStatus)) {
+ return false;
+ }
+ String filePath = fileStatus.getPath().toString();
+ if (StringUtils.isNotEmpty(filenameExtension) &&
!filePath.endsWith(filenameExtension)) {
+ return false;
+ }
+ if (!filterFileByPattern(fileStatus)) {
+ return false;
+ }
+ if (!readPartitions.isEmpty()) {
+ for (String readPartition : readPartitions) {
+ if (filePath.contains(readPartition)) {
+ return true;
}
}
-
- if (updateModeStats != null) {
- updateModeStats.scanned++;
- }
- if (shouldSyncFileInUpdateMode(fileStatus)) {
- FileInfo fileInfo = new FileInfo(filePath,
fileStatus.getModificationTime());
- fileInfoList.add(fileInfo);
- } else if (updateModeStats != null) {
- updateModeStats.skipped++;
- }
+ return false;
}
+ return true;
}
private Date getFileModifiedDate(String modifiedDate) {
@@ -632,14 +671,19 @@ public abstract class AbstractReadStrategy implements
ReadStrategy {
@Override
public void close() throws IOException {
+ if (targetHadoopFileSystemProxy != null &&
!shareTargetFileSystemProxy) {
+ closeFileSystemProxy(targetHadoopFileSystemProxy, "target");
+ }
+ if (hadoopFileSystemProxy != null) {
+ closeFileSystemProxy(hadoopFileSystemProxy, "source");
+ }
+ }
+
+ private void closeFileSystemProxy(HadoopFileSystemProxy proxy, String
role) {
try {
- if (targetHadoopFileSystemProxy != null &&
!shareTargetFileSystemProxy) {
- targetHadoopFileSystemProxy.close();
- }
- if (hadoopFileSystemProxy != null) {
- hadoopFileSystemProxy.close();
- }
- } catch (Exception ignore) {
+ proxy.close();
+ } catch (Exception e) {
+ log.warn("Failed to close {} file system proxy", role, e);
}
}
@@ -695,6 +739,27 @@ public abstract class AbstractReadStrategy implements
ReadStrategy {
+ " is not supported when
update_strategy=distcp.");
}
+ updateCompareParallelism =
+
pluginConfig.hasPath(FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM.key())
+ ? pluginConfig.getInt(
+
FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM.key())
+ :
FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM.defaultValue();
+ if (updateCompareParallelism < 1 || updateCompareParallelism > 64) {
+ throw new FileConnectorException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ "update_compare_parallelism must be between 1 and 64.");
+ }
+ updateCompareBulkThreshold =
+
pluginConfig.hasPath(FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD.key())
+ ? pluginConfig.getInt(
+
FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD.key())
+ :
FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD.defaultValue();
+ if (updateCompareBulkThreshold < 0) {
+ throw new FileConnectorException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ "update_compare_bulk_threshold must be greater than or
equal to 0.");
+ }
+
if
(pluginConfig.hasPath(FileBaseSourceOptions.TARGET_HADOOP_CONF.key())) {
ConfigObject configObject =
pluginConfig.getObject(FileBaseSourceOptions.TARGET_HADOOP_CONF.key());
@@ -770,21 +835,14 @@ public abstract class AbstractReadStrategy implements
ReadStrategy {
}
}
- private boolean shouldSyncFileInUpdateMode(FileStatus sourceFileStatus)
throws IOException {
+ private boolean shouldSyncFileInUpdateMode(
+ FileStatus sourceFileStatus, FileStatus targetFileStatus, String
targetFilePath)
+ throws IOException {
if (!enableUpdateSync) {
return true;
}
- if (targetHadoopFileSystemProxy == null) {
- initTargetHadoopFileSystemProxy();
- }
String sourceFilePath = sourceFileStatus.getPath().toString();
- String relativePath = resolveRelativePath(sourceRootPath,
sourceFilePath);
- String targetFilePath = buildTargetFilePath(targetPath, relativePath);
-
- FileStatus targetFileStatus;
- try {
- targetFileStatus =
targetHadoopFileSystemProxy.getFileStatus(targetFilePath);
- } catch (FileNotFoundException e) {
+ if (targetFileStatus == null) {
return true;
}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/FileDiscoveryScanner.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/FileDiscoveryScanner.java
new file mode 100644
index 0000000000..e778376536
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/FileDiscoveryScanner.java
@@ -0,0 +1,144 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.file.source.reader;
+
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileStatusListingSession;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+@Slf4j
+final class FileDiscoveryScanner {
+ private static final long PROGRESS_INTERVAL = 10_000L;
+
+ private FileDiscoveryScanner() {}
+
+ /** Traverses directories explicitly and applies the filter in the listing
callback. */
+ static ScanStats scan(
+ Path root,
+ boolean recursive,
+ FileStatusListingSession session,
+ FileFilter filter,
+ FileStatusListingSession.FileStatusConsumer candidateConsumer)
+ throws IOException {
+ long started = System.nanoTime();
+ ScanStats stats = new ScanStats();
+ FileStatus rootStatus;
+ try {
+ rootStatus = session.getFileStatus(root);
+ } catch (IOException e) {
+ throw new IOException(
+ "Failed during source_listing root_stat for protocol="
+ + scheme(root)
+ + ", path="
+ + mask(root),
+ e);
+ }
+ if (rootStatus.isFile()) {
+ stats.sourceEntries++;
+ process(rootStatus, filter, candidateConsumer, stats);
+ } else {
+ Deque<Path> directories = new ArrayDeque<>();
+ directories.push(rootStatus.getPath());
+ while (!directories.isEmpty()) {
+ Path directory = directories.pop();
+ try {
+ session.list(
+ directory,
+ status -> {
+ stats.sourceEntries++;
+ if (stats.sourceEntries % PROGRESS_INTERVAL ==
0) {
+ log.info(
+ "File source listing progress:
phase=source_construction, entries={}, candidates={}",
+ stats.sourceEntries,
+ stats.candidates);
+ }
+ if (status.isDirectory()) {
+ if (recursive &&
!status.getPath().getName().startsWith(".")) {
+ directories.addLast(status.getPath());
+ }
+ return;
+ }
+ process(status, filter, candidateConsumer,
stats);
+ });
+ stats.listedDirectories++;
+ } catch (IOException e) {
+ throw new IOException(
+ "Failed during source_listing for protocol="
+ + scheme(directory)
+ + ", path="
+ + mask(directory),
+ e);
+ }
+ }
+ }
+ stats.listingNanos = System.nanoTime() - started -
stats.filteringNanos;
+ return stats;
+ }
+
+ private static void process(
+ FileStatus status,
+ FileFilter filter,
+ FileStatusListingSession.FileStatusConsumer candidateConsumer,
+ ScanStats stats)
+ throws IOException {
+ long filterStarted = System.nanoTime();
+ boolean accepted = filter.test(status);
+ stats.filteringNanos += System.nanoTime() - filterStarted;
+ if (accepted) {
+ candidateConsumer.accept(status);
+ stats.candidates++;
+ } else {
+ stats.filtered++;
+ }
+ }
+
+ private static String scheme(Path path) {
+ return path.toUri().getScheme() == null ? "default" :
path.toUri().getScheme();
+ }
+
+ private static String mask(Path path) {
+ java.net.URI uri = path.toUri();
+ if (uri.getUserInfo() == null || uri.getAuthority() == null) {
+ return path.toString();
+ }
+ return path.toString().replace(uri.getUserInfo() + "@", "***@");
+ }
+
+ @FunctionalInterface
+ interface FileFilter {
+ boolean test(FileStatus status) throws IOException;
+ }
+
+ @Getter
+ static final class ScanStats {
+ private long sourceEntries;
+ private long filtered;
+ private long candidates;
+ private long listedDirectories;
+ private long listingNanos;
+ private long filteringNanos;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateFileMetadataLoader.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateFileMetadataLoader.java
new file mode 100644
index 0000000000..af4e1eb548
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateFileMetadataLoader.java
@@ -0,0 +1,295 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.file.source.reader;
+
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileStatusListingSession;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.HadoopFileSystemProxy;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+
+import lombok.Getter;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+final class UpdateFileMetadataLoader {
+ private static final int IN_FLIGHT_MULTIPLIER = 8;
+ private static final AtomicInteger LOOKUP_THREAD_SEQUENCE = new
AtomicInteger();
+
+ private UpdateFileMetadataLoader() {}
+
+ /**
+ * Loads target metadata using one directory list for dense groups and
bounded point lookups.
+ */
+ static Result load(
+ List<Request> requests,
+ HadoopFileSystemProxy target,
+ int parallelism,
+ int bulkThreshold)
+ throws IOException {
+ FileStatus[] ordered = new FileStatus[requests.size()];
+ Map<Path, List<Request>> groups = new LinkedHashMap<>();
+ List<Request> pointRequests = new ArrayList<>();
+ for (Request request : requests) {
+ Path parent = new Path(request.targetPath).getParent();
+ if (parent == null || parent.toString().isEmpty()) {
+ pointRequests.add(request);
+ } else {
+ groups.computeIfAbsent(parent, ignored -> new
ArrayList<>()).add(request);
+ }
+ }
+
+ List<Map.Entry<Path, List<Request>>> bulkGroups = new ArrayList<>();
+ for (Map.Entry<Path, List<Request>> group : groups.entrySet()) {
+ if (bulkThreshold == 0 || group.getValue().size() < bulkThreshold)
{
+ pointRequests.addAll(group.getValue());
+ } else {
+ bulkGroups.add(group);
+ }
+ }
+
+ long bulkNanos = 0;
+ if (!bulkGroups.isEmpty()) {
+ try (FileStatusListingSession session =
target.openFileStatusListingSession()) {
+ for (Map.Entry<Path, List<Request>> group : bulkGroups) {
+ long started = System.nanoTime();
+ Map<String, Request> wanted = new HashMap<>();
+ for (Request request : group.getValue()) {
+ wanted.put(new Path(request.targetPath).getName(),
request);
+ }
+ try {
+ session.list(
+ group.getKey(),
+ status -> {
+ Request request =
wanted.get(status.getPath().getName());
+ if (request != null) {
+ ordered[request.order] = status;
+ }
+ });
+ } catch (FileNotFoundException ignored) {
+ // A missing target directory means every source file
in the group is new.
+ } catch (IOException e) {
+ throw new IOException(
+ "Failed during target_bulk_listing for path="
+ + mask(group.getKey()),
+ e);
+ } finally {
+ bulkNanos += System.nanoTime() - started;
+ }
+ }
+ }
+ }
+
+ PointResult pointResult = loadPoints(pointRequests, target, ordered,
parallelism);
+ return new Result(
+ Arrays.asList(ordered),
+ bulkGroups.size(),
+ pointRequests.size(),
+ pointResult.peakConcurrency,
+ pointResult.peakInFlight,
+ bulkNanos,
+ pointResult.elapsedNanos);
+ }
+
+ private static PointResult loadPoints(
+ List<Request> requests,
+ HadoopFileSystemProxy target,
+ FileStatus[] ordered,
+ int parallelism)
+ throws IOException {
+ if (requests.isEmpty()) {
+ return new PointResult(0, 0, 0);
+ }
+ ExecutorService executor =
+ Executors.newFixedThreadPool(
+ parallelism,
+ runnable -> {
+ Thread thread =
+ new Thread(
+ runnable,
+ "seatunnel-file-update-lookup-"
+ +
LOOKUP_THREAD_SEQUENCE.incrementAndGet());
+ thread.setDaemon(true);
+ return thread;
+ });
+ CompletionService<Lookup> completion = new
ExecutorCompletionService<>(executor);
+ Set<Future<Lookup>> inFlight = new HashSet<>();
+ int maxInFlight = parallelism * IN_FLIGHT_MULTIPLIER;
+ int submitted = 0;
+ int completed = 0;
+ int peakInFlight = 0;
+ AtomicInteger active = new AtomicInteger();
+ AtomicInteger peakConcurrency = new AtomicInteger();
+ long started = System.nanoTime();
+ IOException failure = null;
+ try {
+ while (completed < requests.size()) {
+ while (submitted < requests.size() && submitted - completed <
maxInFlight) {
+ Request request = requests.get(submitted++);
+ inFlight.add(
+ completion.submit(
+ () -> {
+ int current = active.incrementAndGet();
+
peakConcurrency.accumulateAndGet(current, Math::max);
+ try {
+ return new Lookup(
+ request,
+
target.getFileStatus(request.targetPath));
+ } catch (FileNotFoundException
ignored) {
+ return new Lookup(request, null);
+ } catch (IOException e) {
+ throw new IOException(
+ "Target metadata lookup
failed for path="
+ + mask(new
Path(request.targetPath)),
+ e);
+ } finally {
+ active.decrementAndGet();
+ }
+ }));
+ peakInFlight = Math.max(peakInFlight, submitted -
completed);
+ }
+ try {
+ Future<Lookup> completedFuture = completion.take();
+ inFlight.remove(completedFuture);
+ Lookup lookup = completedFuture.get();
+ ordered[lookup.request.order] = lookup.status;
+ completed++;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ failure = new IOException("Interrupted during
target_point_lookup", e);
+ throw failure;
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ failure = new IOException("Failed during
target_point_lookup", cause);
+ throw failure;
+ }
+ }
+ return new PointResult(
+ peakConcurrency.get(), peakInFlight, System.nanoTime() -
started);
+ } finally {
+ for (Future<Lookup> future : inFlight) {
+ future.cancel(true);
+ }
+ executor.shutdownNow();
+ try {
+ if (!executor.awaitTermination(30, TimeUnit.SECONDS)) {
+ IOException cleanupFailure =
+ new IOException("Timed out closing
target_point_lookup workers");
+ if (failure != null) {
+ failure.addSuppressed(cleanupFailure);
+ } else {
+ throw cleanupFailure;
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ if (failure != null) {
+ failure.addSuppressed(e);
+ } else {
+ throw new IOException("Interrupted closing
target_point_lookup workers", e);
+ }
+ }
+ }
+ }
+
+ @Getter
+ static final class Request {
+ private final int order;
+ private final String targetPath;
+
+ Request(int order, String targetPath) {
+ this.order = order;
+ this.targetPath = targetPath;
+ }
+ }
+
+ @Getter
+ static final class Result {
+ private final List<FileStatus> statuses;
+ private final int bulkListedDirectories;
+ private final int pointLookups;
+ private final int peakConcurrency;
+ private final int peakInFlight;
+ private final long bulkListingNanos;
+ private final long pointLookupNanos;
+
+ Result(
+ List<FileStatus> statuses,
+ int bulkListedDirectories,
+ int pointLookups,
+ int peakConcurrency,
+ int peakInFlight,
+ long bulkListingNanos,
+ long pointLookupNanos) {
+ this.statuses = statuses;
+ this.bulkListedDirectories = bulkListedDirectories;
+ this.pointLookups = pointLookups;
+ this.peakConcurrency = peakConcurrency;
+ this.peakInFlight = peakInFlight;
+ this.bulkListingNanos = bulkListingNanos;
+ this.pointLookupNanos = pointLookupNanos;
+ }
+ }
+
+ private static final class Lookup {
+ private final Request request;
+ private final FileStatus status;
+
+ private Lookup(Request request, FileStatus status) {
+ this.request = request;
+ this.status = status;
+ }
+ }
+
+ private static final class PointResult {
+ private final int peakConcurrency;
+ private final int peakInFlight;
+ private final long elapsedNanos;
+
+ private PointResult(int peakConcurrency, int peakInFlight, long
elapsedNanos) {
+ this.peakConcurrency = peakConcurrency;
+ this.peakInFlight = peakInFlight;
+ this.elapsedNanos = elapsedNanos;
+ }
+ }
+
+ private static String mask(Path path) {
+ java.net.URI uri = path.toUri();
+ if (uri.getUserInfo() == null || uri.getAuthority() == null) {
+ return path.toString();
+ }
+ return path.toString().replace(uri.getUserInfo() + "@", "***@");
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategyTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategyTest.java
index a8b74b2818..4c54e583b3 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategyTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategyTest.java
@@ -26,6 +26,7 @@ import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import
org.apache.seatunnel.connectors.seatunnel.file.config.FileBaseSourceOptions;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.HadoopFileSystemProxy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.split.FileSourceSplit;
import org.apache.seatunnel.connectors.seatunnel.file.util.LocalFileSystemConf;
@@ -46,6 +47,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
+import org.mockito.Mockito;
import lombok.extern.slf4j.Slf4j;
@@ -66,6 +68,22 @@ import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME
@Slf4j
public class AbstractReadStrategyTest {
+ @Test
+ void testCloseSourceProxyWhenTargetProxyCloseFails() throws Exception {
+ HadoopFileSystemProxy targetProxy =
Mockito.mock(HadoopFileSystemProxy.class);
+ HadoopFileSystemProxy sourceProxy =
Mockito.mock(HadoopFileSystemProxy.class);
+ Mockito.doThrow(new IOException("target close
failed")).when(targetProxy).close();
+
+ BinaryReadStrategy strategy = new BinaryReadStrategy();
+ strategy.targetHadoopFileSystemProxy = targetProxy;
+ strategy.hadoopFileSystemProxy = sourceProxy;
+
+ strategy.close();
+
+ Mockito.verify(targetProxy).close();
+ Mockito.verify(sourceProxy).close();
+ }
+
@Test
void testSafeSliceUsesSeekForSeekableStream() throws Exception {
byte[] data = "0123456789".getBytes(StandardCharsets.UTF_8);
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/FileDiscoveryScannerTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/FileDiscoveryScannerTest.java
new file mode 100644
index 0000000000..1443bfe94f
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/FileDiscoveryScannerTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.file.source.reader;
+
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileStatusListingSession;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class FileDiscoveryScannerTest {
+
+ @Test
+ void shouldFilterEntriesAsTheyAreListed() throws Exception {
+ int total = 100_000;
+ List<FileStatus> accepted = new ArrayList<>();
+ GeneratingListingSession session = new GeneratingListingSession(total);
+
+ FileDiscoveryScanner.ScanStats stats =
+ FileDiscoveryScanner.scan(
+ new Path("file:///source"),
+ false,
+ session,
+ status -> status.getPath().getName().endsWith("000"),
+ accepted::add);
+
+ Assertions.assertEquals(total, stats.getSourceEntries());
+ Assertions.assertEquals(100, stats.getCandidates());
+ Assertions.assertEquals(100, accepted.size());
+ Assertions.assertTrue(session.callbackWasConsumedBeforeNextEntry);
+ }
+
+ @Test
+ void shouldUseExplicitDirectoryTraversalAndListEachDirectoryOnce() throws
Exception {
+ Map<String, List<FileStatus>> entries = new HashMap<>();
+ entries.put(
+ "file:/source",
+ Arrays.asList(directory("file:///source/day=01"),
file("file:///source/root.bin")));
+ entries.put(
+ "file:/source/day=01",
+
Collections.singletonList(file("file:///source/day=01/a.bin")));
+ RecordingListingSession session = new RecordingListingSession(entries);
+ List<FileStatus> accepted = new ArrayList<>();
+
+ FileDiscoveryScanner.scan(
+ new Path("file:///source"), true, session, status -> true,
accepted::add);
+
+ Assertions.assertEquals(2, accepted.size());
+ Assertions.assertEquals(
+ Arrays.asList("file:/source", "file:/source/day=01"),
session.listed);
+ }
+
+ @Test
+ void shouldCountRootFileAsSourceEntry() throws Exception {
+ FileStatus rootFile = file("file:///source/file.bin");
+ FileStatusListingSession session =
+ new FileStatusListingSession() {
+ @Override
+ public FileStatus getFileStatus(Path path) {
+ return rootFile;
+ }
+
+ @Override
+ public void list(Path directory, FileStatusConsumer
consumer) {
+ Assertions.fail("A root file must not be listed as a
directory");
+ }
+
+ @Override
+ public void close() {}
+ };
+
+ FileDiscoveryScanner.ScanStats stats =
+ FileDiscoveryScanner.scan(
+ rootFile.getPath(), true, session, status -> true,
ignored -> {});
+
+ Assertions.assertEquals(1, stats.getSourceEntries());
+ Assertions.assertEquals(1, stats.getCandidates());
+ }
+
+ private static FileStatus file(String path) {
+ return new FileStatus(1, false, 1, 1, 1, new Path(path));
+ }
+
+ private static FileStatus directory(String path) {
+ return new FileStatus(0, true, 1, 1, 1, new Path(path));
+ }
+
+ private static final class GeneratingListingSession implements
FileStatusListingSession {
+ private final int total;
+ private boolean callbackWasConsumedBeforeNextEntry = true;
+ private int consumed;
+
+ private GeneratingListingSession(int total) {
+ this.total = total;
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) {
+ return directory(path.toString());
+ }
+
+ @Override
+ public void list(Path directory, FileStatusConsumer consumer) throws
IOException {
+ for (int i = 0; i < total; i++) {
+ int before = consumed;
+ consumer.accept(file(directory + "/file-" +
String.format("%06d", i)));
+ consumed++;
+ callbackWasConsumedBeforeNextEntry &= consumed == before + 1;
+ }
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ private static final class RecordingListingSession implements
FileStatusListingSession {
+ private final Map<String, List<FileStatus>> entries;
+ private final List<String> listed = new ArrayList<>();
+
+ private RecordingListingSession(Map<String, List<FileStatus>> entries)
{
+ this.entries = entries;
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) {
+ return directory(path.toString());
+ }
+
+ @Override
+ public void list(Path directory, FileStatusConsumer consumer) throws
IOException {
+ listed.add(directory.toString());
+ for (FileStatus status :
+ entries.getOrDefault(directory.toString(),
Collections.emptyList())) {
+ consumer.accept(status);
+ }
+ }
+
+ @Override
+ public void close() {}
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateCompareConfigTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateCompareConfigTest.java
new file mode 100644
index 0000000000..9fef221345
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateCompareConfigTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.file.source.reader;
+
+import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
+
+import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+class UpdateCompareConfigTest {
+
+ @Test
+ void shouldRejectPointLookupParallelismOutsideRange() throws Exception {
+ for (int invalid : new int[] {0, 65}) {
+ Map<String, Object> config = updateConfig();
+ config.put("update_compare_parallelism", invalid);
+ try (BinaryReadStrategy strategy = new BinaryReadStrategy()) {
+ Assertions.assertThrows(
+ FileConnectorException.class,
+ () ->
strategy.setPluginConfig(ConfigFactory.parseMap(config)));
+ }
+ }
+ }
+
+ @Test
+ void shouldRejectNegativeBulkThreshold() throws Exception {
+ Map<String, Object> config = updateConfig();
+ config.put("update_compare_bulk_threshold", -1);
+ try (BinaryReadStrategy strategy = new BinaryReadStrategy()) {
+ Assertions.assertThrows(
+ FileConnectorException.class,
+ () ->
strategy.setPluginConfig(ConfigFactory.parseMap(config)));
+ }
+ }
+
+ @Test
+ void shouldAcceptDefaultsAndBoundaryValues() throws Exception {
+ Map<String, Object> config = updateConfig();
+ config.put("update_compare_parallelism", 64);
+ config.put("update_compare_bulk_threshold", 0);
+ try (BinaryReadStrategy strategy = new BinaryReadStrategy()) {
+ Assertions.assertDoesNotThrow(
+ () ->
strategy.setPluginConfig(ConfigFactory.parseMap(config)));
+ Assertions.assertEquals(64, strategy.updateCompareParallelism);
+ Assertions.assertEquals(0, strategy.updateCompareBulkThreshold);
+ }
+ }
+
+ @Test
+ void shouldDisableAutomaticBulkComparisonByDefault() throws Exception {
+ try (BinaryReadStrategy strategy = new BinaryReadStrategy()) {
+ Assertions.assertDoesNotThrow(
+ () ->
strategy.setPluginConfig(ConfigFactory.parseMap(updateConfig())));
+ Assertions.assertEquals(0, strategy.updateCompareBulkThreshold);
+ }
+ }
+
+ private static Map<String, Object> updateConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("path", "/source");
+ config.put("file_format_type", "binary");
+ config.put("sync_mode", "update");
+ config.put("target_path", "/target");
+ return config;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateFileMetadataLoaderTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateFileMetadataLoaderTest.java
new file mode 100644
index 0000000000..b3b2049b28
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/UpdateFileMetadataLoaderTest.java
@@ -0,0 +1,230 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.file.source.reader;
+
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileStatusListingSession;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.HadoopFileSystemProxy;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+class UpdateFileMetadataLoaderTest {
+
+ @Test
+ void shouldListDenseTargetDirectoryOnlyOnce() throws Exception {
+ HadoopFileSystemProxy proxy =
Mockito.mock(HadoopFileSystemProxy.class);
+ AtomicInteger listCount = new AtomicInteger();
+ FileStatusListingSession session =
+ new FileStatusListingSession() {
+ @Override
+ public FileStatus getFileStatus(Path path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void list(Path directory, FileStatusConsumer
consumer)
+ throws java.io.IOException {
+ listCount.incrementAndGet();
+ for (int i = 0; i < 10_000; i++) {
+ consumer.accept(file(directory + "/file-" + i));
+ }
+ }
+
+ @Override
+ public void close() {}
+ };
+ Mockito.when(proxy.openFileStatusListingSession()).thenReturn(session);
+
+ List<UpdateFileMetadataLoader.Request> requests = new ArrayList<>();
+ for (int i = 0; i < 10_000; i++) {
+ requests.add(new UpdateFileMetadataLoader.Request(i,
"s3a://bucket/day/file-" + i));
+ }
+
+ UpdateFileMetadataLoader.Result result =
+ UpdateFileMetadataLoader.load(requests, proxy, 8, 64);
+
+ Assertions.assertEquals(1, listCount.get());
+ Assertions.assertEquals(1, result.getBulkListedDirectories());
+ Assertions.assertEquals(0, result.getPointLookups());
+ Assertions.assertEquals(10_000, result.getStatuses().size());
+ Mockito.verify(proxy,
Mockito.never()).getFileStatus(Mockito.anyString());
+ }
+
+ @Test
+ void shouldBoundSparsePointLookupsAndKeepRequestOrder() throws Exception {
+ HadoopFileSystemProxy proxy =
Mockito.mock(HadoopFileSystemProxy.class);
+ AtomicInteger active = new AtomicInteger();
+ AtomicInteger peak = new AtomicInteger();
+ Mockito.when(proxy.getFileStatus(Mockito.anyString()))
+ .thenAnswer(
+ invocation -> {
+ int current = active.incrementAndGet();
+ peak.accumulateAndGet(current, Math::max);
+ try {
+ Thread.sleep(5);
+ return file(invocation.getArgument(0));
+ } finally {
+ active.decrementAndGet();
+ }
+ });
+
+ List<UpdateFileMetadataLoader.Request> requests = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ requests.add(new UpdateFileMetadataLoader.Request(i,
"s3a://bucket/day-" + i + "/f"));
+ }
+
+ UpdateFileMetadataLoader.Result result =
+ UpdateFileMetadataLoader.load(requests, proxy, 4, 64);
+
+ Assertions.assertEquals(100, result.getPointLookups());
+ Assertions.assertTrue(peak.get() <= 4);
+ Assertions.assertTrue(result.getPeakConcurrency() <= 4);
+ Assertions.assertTrue(result.getPeakInFlight() <= 32);
+ for (int i = 0; i < requests.size(); i++) {
+ Assertions.assertEquals(
+ requests.get(i).getTargetPath(),
+ result.getStatuses().get(i).getPath().toString());
+ }
+ }
+
+ @Test
+ void shouldHonorDisabledBulkComparisonForFtpAndSftpTargets() throws
Exception {
+ HadoopFileSystemProxy proxy =
Mockito.mock(HadoopFileSystemProxy.class);
+ FileStatusListingSession session =
Mockito.mock(FileStatusListingSession.class);
+ Mockito.when(proxy.openFileStatusListingSession()).thenReturn(session);
+ Mockito.when(proxy.getFileStatus(Mockito.anyString()))
+ .thenAnswer(invocation -> file(invocation.getArgument(0)));
+ List<UpdateFileMetadataLoader.Request> requests =
+ Collections.singletonList(
+ new UpdateFileMetadataLoader.Request(0,
"sftp://host/path/file.bin";));
+
+ UpdateFileMetadataLoader.Result result =
+ UpdateFileMetadataLoader.load(requests, proxy, 8, 0);
+
+ Assertions.assertEquals(0, result.getBulkListedDirectories());
+ Assertions.assertEquals(1, result.getPointLookups());
+ Mockito.verify(proxy).getFileStatus("sftp://host/path/file.bin";);
+ Mockito.verify(session, Mockito.never()).list(Mockito.any(),
Mockito.any());
+ }
+
+ @Test
+ void shouldUsePointLookupsWhenAutomaticBulkComparisonIsDisabled() throws
Exception {
+ HadoopFileSystemProxy proxy =
Mockito.mock(HadoopFileSystemProxy.class);
+ Mockito.when(proxy.getFileStatus(Mockito.anyString()))
+ .thenAnswer(invocation -> file(invocation.getArgument(0)));
+ List<UpdateFileMetadataLoader.Request> requests = new ArrayList<>();
+ for (int i = 0; i < 100; i++) {
+ requests.add(new UpdateFileMetadataLoader.Request(i,
"s3a://bucket/day/file-" + i));
+ }
+
+ UpdateFileMetadataLoader.Result result =
+ UpdateFileMetadataLoader.load(requests, proxy, 4, 0);
+
+ Assertions.assertEquals(0, result.getBulkListedDirectories());
+ Assertions.assertEquals(100, result.getPointLookups());
+ Mockito.verify(proxy, Mockito.never()).openFileStatusListingSession();
+ }
+
+ @Test
+ void shouldUseNamedDaemonThreadsForPointLookups() throws Exception {
+ HadoopFileSystemProxy proxy =
Mockito.mock(HadoopFileSystemProxy.class);
+ AtomicReference<Thread> lookupThread = new AtomicReference<>();
+ Mockito.when(proxy.getFileStatus(Mockito.anyString()))
+ .thenAnswer(
+ invocation -> {
+ lookupThread.set(Thread.currentThread());
+ return file(invocation.getArgument(0));
+ });
+
+ UpdateFileMetadataLoader.load(
+ Collections.singletonList(
+ new UpdateFileMetadataLoader.Request(0,
"s3a://bucket/day/file")),
+ proxy,
+ 1,
+ 0);
+
+ Assertions.assertNotNull(lookupThread.get());
+ Assertions.assertTrue(lookupThread.get().isDaemon());
+ Assertions.assertTrue(
+
lookupThread.get().getName().startsWith("seatunnel-file-update-lookup-"));
+ }
+
+ @Test
+ void shouldUsePointLookupWhenRelativeTargetHasNoParent() throws Exception {
+ HadoopFileSystemProxy proxy =
Mockito.mock(HadoopFileSystemProxy.class);
+
Mockito.when(proxy.getFileStatus("file.bin")).thenReturn(file("file.bin"));
+
+ UpdateFileMetadataLoader.Result result =
+ UpdateFileMetadataLoader.load(
+ Collections.singletonList(
+ new UpdateFileMetadataLoader.Request(0,
"file.bin")),
+ proxy,
+ 1,
+ 1);
+
+ Assertions.assertEquals(1, result.getPointLookups());
+ Assertions.assertEquals(0, result.getBulkListedDirectories());
+ Mockito.verify(proxy, Mockito.never()).openFileStatusListingSession();
+ }
+
+ @Test
+ void shouldCancelPointLookupsAndReleaseWorkersAfterFirstFailure() throws
Exception {
+ HadoopFileSystemProxy proxy =
Mockito.mock(HadoopFileSystemProxy.class);
+ AtomicInteger active = new AtomicInteger();
+ Mockito.when(proxy.getFileStatus(Mockito.anyString()))
+ .thenAnswer(
+ invocation -> {
+ String path = invocation.getArgument(0);
+ if (path.endsWith("/fail")) {
+ throw new java.io.IOException("target
unavailable");
+ }
+ active.incrementAndGet();
+ try {
+ Thread.sleep(30_000);
+ return file(path);
+ } finally {
+ active.decrementAndGet();
+ }
+ });
+ List<UpdateFileMetadataLoader.Request> requests = new ArrayList<>();
+ requests.add(new UpdateFileMetadataLoader.Request(0,
"s3a://bucket/a/fail"));
+ for (int i = 1; i < 20; i++) {
+ requests.add(
+ new UpdateFileMetadataLoader.Request(i,
"s3a://bucket/dir-" + i + "/file"));
+ }
+
+ Assertions.assertThrows(
+ java.io.IOException.class,
+ () -> UpdateFileMetadataLoader.load(requests, proxy, 4, 64));
+ Assertions.assertEquals(0, active.get());
+ }
+
+ private static FileStatus file(String path) {
+ return new FileStatus(1, false, 1, 1, 1, new Path(path));
+ }
+}
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 4c48a411fb..34eff76c1c 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
@@ -117,7 +117,9 @@ public class FtpFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.SYNC_MODE,
FileBaseSourceOptions.TARGET_HADOOP_CONF,
FileBaseSourceOptions.UPDATE_STRATEGY,
- FileBaseSourceOptions.COMPARE_MODE)
+ FileBaseSourceOptions.COMPARE_MODE,
+ FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM,
+ FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD)
.conditional(
FileBaseSourceOptions.SYNC_MODE,
FileSyncMode.UPDATE,
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystem.java
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystem.java
index dc312d8ff7..095d8f13b3 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystem.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystem.java
@@ -18,12 +18,15 @@
package org.apache.seatunnel.connectors.seatunnel.file.ftp.system;
import
org.apache.seatunnel.connectors.seatunnel.file.ftp.config.FtpFileBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileStatusListingSession;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.StreamingFileSystem;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
+import org.apache.commons.net.ftp.FTPListParseEngine;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
@@ -57,7 +60,7 @@ import java.net.URI;
@InterfaceAudience.Public
@InterfaceStability.Stable
@Slf4j
-public class SeaTunnelFTPFileSystem extends FileSystem {
+public class SeaTunnelFTPFileSystem extends FileSystem implements
StreamingFileSystem {
public static final Log LOG =
LogFactory.getLog(SeaTunnelFTPFileSystem.class);
public static final int DEFAULT_BUFFER_SIZE = 1024 * 1024;
@@ -490,6 +493,102 @@ public class SeaTunnelFTPFileSystem extends FileSystem {
}
}
+ @Override
+ public FileStatusListingSession openFileStatusListingSession() throws
IOException {
+ return new FtpListingSession(connect());
+ }
+
+ private final class FtpListingSession implements FileStatusListingSession {
+ private static final int PAGE_SIZE = 1_000;
+ private final FTPClient client;
+
+ private FtpListingSession(FTPClient client) {
+ this.client = client;
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ return SeaTunnelFTPFileSystem.this.getFileStatus(client, path);
+ }
+
+ @Override
+ public void list(Path directory, FileStatusConsumer consumer) throws
IOException {
+ Path workDir = new Path(client.printWorkingDirectory());
+ Path absolute = makeAbsolute(workDir, directory);
+ String previousDirectory = client.printWorkingDirectory();
+ if (!client.changeWorkingDirectory(absolute.toUri().getPath())) {
+ throw new FileNotFoundException("FTP directory does not exist:
" + mask(absolute));
+ }
+ Throwable failure = null;
+ try {
+ FTPListParseEngine engine = client.initiateListParsing();
+ if (!FTPReply.isPositiveCompletion(client.getReplyCode())) {
+ throw new IOException(
+ "FTP LIST failed for path="
+ + mask(absolute)
+ + ", replyCode="
+ + client.getReplyCode());
+ }
+ while (engine.hasNext()) {
+ FTPFile[] files = engine.getNext(PAGE_SIZE);
+ int skipped = emitFileStatuses(files, absolute, consumer);
+ if (skipped > 0) {
+ LOG.warn(
+ "Skipped "
+ + skipped
+ + " unparseable FTP listing entries
under "
+ + mask(absolute));
+ }
+ }
+ } catch (IOException | RuntimeException | Error e) {
+ failure = e;
+ throw e;
+ } finally {
+ if (!client.changeWorkingDirectory(previousDirectory)) {
+ IOException restoreFailure =
+ new IOException(
+ "Failed to restore FTP working directory,
replyCode="
+ + client.getReplyCode());
+ if (failure != null) {
+ failure.addSuppressed(restoreFailure);
+ } else {
+ throw restoreFailure;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ disconnect(client);
+ }
+ }
+
+ int emitFileStatuses(
+ FTPFile[] files, Path parent,
FileStatusListingSession.FileStatusConsumer consumer)
+ throws IOException {
+ int skipped = 0;
+ for (FTPFile file : files) {
+ if (file == null) {
+ skipped++;
+ continue;
+ }
+ String name = file.getName();
+ if (!".".equals(name) && !"..".equals(name)) {
+ consumer.accept(getFileStatus(file, parent));
+ }
+ }
+ return skipped;
+ }
+
+ private static String mask(Path path) {
+ URI uri = path.toUri();
+ if (uri.getUserInfo() == null || uri.getAuthority() == null) {
+ return path.toString();
+ }
+ return path.toString().replace(uri.getUserInfo() + "@", "***@");
+ }
+
/**
* Convenience method, so that we don't open a new connection when using
this method from within
* another method. Otherwise every API invocation incurs the overhead of
opening/closing a TCP
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 91c5ec816c..9fb0d9eb1e 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
@@ -42,6 +42,14 @@ class FtpFileFactoryTest {
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.UPDATE_STRATEGY));
Assertions.assertTrue(
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.COMPARE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD));
Assertions.assertTrue(
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.DISCOVERY_MODE));
Assertions.assertTrue(
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystemTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystemTest.java
index fd6a1a9d05..a29351bc8b 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystemTest.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-ftp/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystemTest.java
@@ -17,6 +17,9 @@
package org.apache.seatunnel.connectors.seatunnel.file.ftp.system;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileStatusListingSession;
+
+import org.apache.commons.net.ftp.FTPFile;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -37,10 +40,14 @@ import org.mockftpserver.fake.filesystem.UnixFakeFileSystem;
import java.io.IOException;
import java.net.URI;
import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Unit tests for SeaTunnelFTPFileSystem. */
@@ -162,6 +169,31 @@ public class SeaTunnelFTPFileSystemTest {
ftpFileSystem.delete(testDir, true);
}
+ @Test
+ public void testStreamingListingSkipsUnparseableEntries() throws
IOException {
+ FTPFile file = new FTPFile();
+ file.setName("valid.txt");
+ file.setType(FTPFile.FILE_TYPE);
+ file.setSize(1L);
+ file.setTimestamp(Calendar.getInstance());
+ List<FileStatus> statuses = new ArrayList<>();
+
+ ftpFileSystem.emitFileStatuses(
+ new FTPFile[] {null, file}, new Path(HOME_DIR), statuses::add);
+
+ assertEquals(1, statuses.size());
+ assertEquals("valid.txt", statuses.get(0).getPath().getName());
+ }
+
+ @Test
+ public void testStreamingListingFailsForMissingDirectory() throws
IOException {
+ try (FileStatusListingSession session =
ftpFileSystem.openFileStatusListingSession()) {
+ assertThrows(
+ IOException.class,
+ () -> session.list(new Path(HOME_DIR + "/missing"),
ignored -> {}));
+ }
+ }
+
@Test
public void testRename() throws IOException {
Path source = new Path(HOME_DIR + "/source.txt");
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 215a8b8b74..111941c248 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
@@ -121,7 +121,9 @@ public class HdfsFileSourceFactory implements
TableSourceFactory {
HdfsFileSourceOptions.SYNC_MODE,
HdfsFileSourceOptions.TARGET_HADOOP_CONF,
HdfsFileSourceOptions.UPDATE_STRATEGY,
- HdfsFileSourceOptions.COMPARE_MODE)
+ HdfsFileSourceOptions.COMPARE_MODE,
+ HdfsFileSourceOptions.UPDATE_COMPARE_PARALLELISM,
+ HdfsFileSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD)
.conditional(
HdfsFileSourceOptions.SYNC_MODE,
FileSyncMode.UPDATE,
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 e2ca58d910..5a57498627 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
@@ -37,6 +37,14 @@ class HdfsFileFactoryTest {
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.SCAN_INTERVAL));
Assertions.assertTrue(
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.START_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD));
Assertions.assertNotNull((new HdfsFileSinkFactory()).optionRule());
}
}
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 919eaecd31..fd7d613913 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
@@ -124,7 +124,9 @@ public class LocalFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.SYNC_MODE,
FileBaseSourceOptions.TARGET_HADOOP_CONF,
FileBaseSourceOptions.UPDATE_STRATEGY,
- FileBaseSourceOptions.COMPARE_MODE)
+ FileBaseSourceOptions.COMPARE_MODE,
+ FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM,
+ FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD)
.conditional(
FileBaseSourceOptions.SYNC_MODE,
FileSyncMode.UPDATE,
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 1fbbe1a12b..c9f63b686b 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
@@ -43,6 +43,14 @@ class LocalFileFactoryTest {
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.UPDATE_STRATEGY));
Assertions.assertTrue(
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.COMPARE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD));
Assertions.assertTrue(
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.DISCOVERY_MODE));
Assertions.assertTrue(
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 acf72f17e9..ca962549f6 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
@@ -105,7 +105,9 @@ public class SftpFileSourceFactory implements
TableSourceFactory {
FileBaseSourceOptions.SYNC_MODE,
FileBaseSourceOptions.TARGET_HADOOP_CONF,
FileBaseSourceOptions.UPDATE_STRATEGY,
- FileBaseSourceOptions.COMPARE_MODE)
+ FileBaseSourceOptions.COMPARE_MODE,
+ FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM,
+ FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD)
.conditional(
FileBaseSourceOptions.SYNC_MODE,
FileSyncMode.UPDATE,
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPFileSystem.java
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPFileSystem.java
index 0990726aa0..ec5093aed5 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPFileSystem.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-sftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sftp/system/SFTPFileSystem.java
@@ -17,6 +17,9 @@
package org.apache.seatunnel.connectors.seatunnel.file.sftp.system;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.FileStatusListingSession;
+import
org.apache.seatunnel.connectors.seatunnel.file.hadoop.StreamingFileSystem;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -31,6 +34,7 @@ import org.slf4j.LoggerFactory;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.ChannelSftp.LsEntry;
+import com.jcraft.jsch.ChannelSftp.LsEntrySelector;
import com.jcraft.jsch.SftpATTRS;
import com.jcraft.jsch.SftpException;
@@ -45,7 +49,7 @@ import java.util.ArrayList;
import java.util.Vector;
/** SFTP FileSystem. */
-public class SFTPFileSystem extends FileSystem {
+public class SFTPFileSystem extends FileSystem implements StreamingFileSystem {
public static final Logger LOG =
LoggerFactory.getLogger(SFTPFileSystem.class);
@@ -235,13 +239,17 @@ public class SFTPFileSystem extends FileSystem {
private FileStatus getFileStatus(ChannelSftp channel, LsEntry sftpFile,
Path parentPath)
throws IOException {
+ return getFileStatus(channel, sftpFile.getFilename(),
sftpFile.getAttrs(), parentPath);
+ }
- SftpATTRS attr = sftpFile.getAttrs();
+ private FileStatus getFileStatus(
+ ChannelSftp channel, String fileName, SftpATTRS attr, Path
parentPath)
+ throws IOException {
long length = attr.getSize();
boolean isDir = attr.isDir();
boolean isLink = attr.isLink();
if (isLink) {
- String link = parentPath.toUri().getPath() + "/" +
sftpFile.getFilename();
+ String link = parentPath.toUri().getPath() + "/" + fileName;
try {
link = channel.realpath(link);
@@ -260,12 +268,12 @@ public class SFTPFileSystem extends FileSystem {
long blockSize = DEFAULT_BLOCK_SIZE;
long modTime = attr.getMTime() * 1000L; // convert to milliseconds
long accessTime = attr.getATime() * 1000L;
- FsPermission permission = getPermissions(sftpFile);
+ FsPermission permission = new FsPermission((short)
attr.getPermissions());
// not be able to get the real user group name, just use the user and
group
// id
String user = Integer.toString(attr.getUId());
String group = Integer.toString(attr.getGId());
- Path filePath = new Path(parentPath, sftpFile.getFilename());
+ Path filePath = new Path(parentPath, fileName);
return new FileStatus(
length,
@@ -602,6 +610,82 @@ public class SFTPFileSystem extends FileSystem {
}
}
+ @Override
+ public FileStatusListingSession openFileStatusListingSession() throws
IOException {
+ return new SftpListingSession(connect());
+ }
+
+ private final class SftpListingSession implements FileStatusListingSession
{
+ private final ChannelSftp client;
+
+ private SftpListingSession(ChannelSftp client) {
+ this.client = client;
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path path) throws IOException {
+ Path absolute;
+ try {
+ absolute = makeAbsolute(new Path(client.pwd()), path);
+ if (absolute.getParent() == null) {
+ return SFTPFileSystem.this.getFileStatus(client, absolute);
+ }
+ SftpATTRS attrs = client.lstat(absolute.toUri().getPath());
+ return SFTPFileSystem.this.getFileStatus(
+ client, absolute.getName(), attrs,
absolute.getParent());
+ } catch (SftpException e) {
+ if (e.id == ChannelSftp.SSH_FX_NO_SUCH_FILE) {
+ throw new
FileNotFoundException(String.format(E_FILE_NOTFOUND, path));
+ }
+ throw new IOException("SFTP lstat failed for path=" + path, e);
+ }
+ }
+
+ @Override
+ public void list(Path directory, FileStatusConsumer consumer) throws
IOException {
+ Path absolute;
+ try {
+ absolute = makeAbsolute(new Path(client.pwd()), directory);
+ client.ls(
+ absolute.toUri().getPath(),
+ entry -> {
+ String name = entry.getFilename();
+ if (!".".equals(name) && !"..".equals(name)) {
+ try {
+ consumer.accept(
+ SFTPFileSystem.this.getFileStatus(
+ client, entry, absolute));
+ } catch (IOException e) {
+ throw new ListingRuntimeException(e);
+ }
+ }
+ return LsEntrySelector.CONTINUE;
+ });
+ } catch (ListingRuntimeException e) {
+ throw e.ioException;
+ } catch (SftpException e) {
+ if (e.id == ChannelSftp.SSH_FX_NO_SUCH_FILE) {
+ throw new
FileNotFoundException(String.format(E_FILE_NOTFOUND, directory));
+ }
+ throw new IOException("SFTP READDIR failed for path=" +
directory, e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ disconnect(client);
+ }
+ }
+
+ private static final class ListingRuntimeException extends
RuntimeException {
+ private final IOException ioException;
+
+ private ListingRuntimeException(IOException ioException) {
+ super(ioException);
+ this.ioException = ioException;
+ }
+ }
+
@Override
public void setWorkingDirectory(Path newDir) {
// we do not maintain the working directory state
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 b38f8c8794..cbf524a498 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
@@ -49,6 +49,14 @@ class SftpFileFactoryTest {
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.UPDATE_STRATEGY));
Assertions.assertTrue(
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.COMPARE_MODE));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.UPDATE_COMPARE_PARALLELISM));
+ Assertions.assertTrue(
+ optionRule
+ .getOptionalOptions()
+
.contains(FileBaseSourceOptions.UPDATE_COMPARE_BULK_THRESHOLD));
Assertions.assertTrue(
optionRule.getOptionalOptions().contains(FileBaseSourceOptions.DISCOVERY_MODE));
Assertions.assertTrue(
