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 3f391f6fe3 [Docs][Connector-V2] Improve SqlServer CDC FtpFile and
HdfsFile docs (#11325)
3f391f6fe3 is described below
commit 3f391f6fe3917fe8680e745c193d7c9aa5620b7d
Author: Daniel Carter <[email protected]>
AuthorDate: Tue Jul 7 18:04:28 2026 +0800
[Docs][Connector-V2] Improve SqlServer CDC FtpFile and HdfsFile docs
(#11325)
---
docs/en/connectors/sink/FtpFile.md | 69 +++++++++++++----------
docs/en/connectors/sink/HdfsFile.md | 56 +++++++++----------
docs/en/connectors/source/FtpFile.md | 11 ++++
docs/en/connectors/source/HdfsFile.md | 11 ++++
docs/en/connectors/source/SqlServer-CDC.md | 90 +++++++++++++++++++++++++++---
docs/zh/connectors/sink/FtpFile.md | 68 ++++++++++++----------
docs/zh/connectors/sink/HdfsFile.md | 56 +++++++++----------
docs/zh/connectors/source/FtpFile.md | 10 ++++
docs/zh/connectors/source/HdfsFile.md | 11 +++-
docs/zh/connectors/source/SqlServer-CDC.md | 89 ++++++++++++++++++++++++++---
10 files changed, 336 insertions(+), 135 deletions(-)
diff --git a/docs/en/connectors/sink/FtpFile.md
b/docs/en/connectors/sink/FtpFile.md
index 8098416911..75963ae937 100644
--- a/docs/en/connectors/sink/FtpFile.md
+++ b/docs/en/connectors/sink/FtpFile.md
@@ -23,6 +23,7 @@ If you use SeaTunnel Engine, It automatically integrated the
hadoop jar when you
Use binary file format to read and write files in any format, such as
videos, pictures, etc. In short, any files can be synchronized to the target
place.
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
By default, we use 2PC commit to ensure `exactly-once`
@@ -78,8 +79,10 @@ By default, we use 2PC commit to ensure `exactly-once`
| parquet_avro_write_fixed_as_int96 | array | no | -
| Only used when file_format is parquet.
|
| enable_header_write | boolean | no | false
| Only used when file_format_type is text,csv.<br/>
false:don't write header,true:write header.
|
| encoding | string | no | "UTF-8"
| Only used when file_format_type is
json,text,csv,xml.
|
+| schema_evolution_enabled | boolean | no | false
| Enable schema evolution support for CDC pipelines.
When true, ADD/DROP/RENAME/MODIFY column events from the source are applied to
the sink without a job restart. Not supported for binary format. |
| schema_save_mode | string | no |
CREATE_SCHEMA_WHEN_NOT_EXIST | Existing dir processing method
|
| data_save_mode | string | no | APPEND_DATA
| Existing data processing method
|
+| multi_table_sink_replica | int | no | 1
| The replica number of sink writers used for each
table in a multi-table sink job.
|
### host [string]
@@ -287,6 +290,40 @@ Existing data processing method.
- APPEND_DATA: preserve dir, preserve data files
- ERROR_WHEN_DATA_EXISTS: when there is data files, an error is reported
+### schema_evolution_enabled [boolean]
+
+When set to `true`, the file sink handles CDC schema change events (ADD
COLUMN, DROP COLUMN, RENAME COLUMN, MODIFY COLUMN type) at runtime without
requiring a job restart. On each schema change the current output file is
closed and a new file is opened with the updated schema.
+
+**Supported formats:** All file formats except `binary`. Enabling this option
with `file_format_type = binary` will fail at job startup with a config
validation error.
+
+**Partition constraint:** When `have_partition = true`, dropping a column
listed in `partition_by` is not allowed and will fail fast. Partition columns
must remain stable across schema changes.
+
+**When `schema_evolution_enabled = false` (default):** If the upstream CDC
source has `schema-changes.enabled = true` and an `AlterTableEvent` arrives at
the sink, the job will throw immediately with an actionable error:
+> `Received AlterTableEvent but schema_evolution_enabled=false at this sink.
Either set schema_evolution_enabled=true to handle schema changes, or set
schema-changes.enabled=false at the CDC source to suppress them.`
+
+Users on the default CDC source config (`schema-changes.enabled = false`) are
completely unaffected.
+
+**Known limitation:** Schema changes are not atomic with checkpointing. If the
job crashes in the narrow window between file rotation and schema metadata
update, rows written after restore may use the pre-change schema. This is a
known architectural gap shared across other SeaTunnel sinks. For full
restart-with-DDL correctness, a follow-up CDC source fix is required (tracked
separately).
+
+Example usage in a CDC pipeline:
+
+```hocon
+FtpFile {
+ host = "xxx.xxx.xxx.xxx"
+ port = 21
+ user = "username"
+ password = "password"
+ path = "/data/ftp/cdc/${table_name}"
+ file_format_type = "parquet"
+ schema_evolution_enabled = true
+}
+```
+
+### multi_table_sink_replica [int]
+
+The replica number of sink writers used for each table in a multi-table sink
job. The default value is `1`; increase it
+only when each table needs more sink writer parallelism.
+
## Example
For text file format simple config
@@ -333,7 +370,9 @@ FtpFile {
```
-When our source end is multiple tables, and wants different expressions to
different directory, we can configure this way
+When the upstream source has multiple tables and each table should be written
to its own FTP directory, include
+`${table_name}` in `path`. `schema_save_mode` and `data_save_mode` decide how
existing directories and files are handled
+before writing.
```hocon
@@ -362,34 +401,6 @@ FtpFile {
```
-### schema_evolution_enabled [boolean]
-
-When set to `true`, the file sink handles CDC schema change events (ADD
COLUMN, DROP COLUMN, RENAME COLUMN, MODIFY COLUMN type) at runtime without
requiring a job restart. On each schema change the current output file is
closed and a new file is opened with the updated schema.
-
-**Supported formats:** All file formats except `binary`. Enabling this option
with `file_format_type = binary` will fail at job startup with a config
validation error.
-
-**Partition constraint:** When `have_partition = true`, dropping a column
listed in `partition_by` is not allowed and will fail fast. Partition columns
must remain stable across schema changes.
-
-**When `schema_evolution_enabled = false` (default):** If the upstream CDC
source has `schema-changes.enabled = true` and an `AlterTableEvent` arrives at
the sink, the job will throw immediately with an actionable error:
-> `Received AlterTableEvent but schema_evolution_enabled=false at this sink.
Either set schema_evolution_enabled=true to handle schema changes, or set
schema-changes.enabled=false at the CDC source to suppress them.`
-
-Users on the default CDC source config (`schema-changes.enabled = false`) are
completely unaffected.
-
-**Known limitation:** Schema changes are not atomic with checkpointing. If the
job crashes in the narrow window between file rotation and schema metadata
update, rows written after restore may use the pre-change schema. This is a
known architectural gap shared across other SeaTunnel sinks. For full
restart-with-DDL correctness, a follow-up CDC source fix is required (tracked
separately).
-
-Example usage in a CDC pipeline:
-
-```hocon
-LocalFile {
- path = "/tmp/cdc/${table_name}"
- file_format_type = "parquet"
- schema_evolution_enabled = true
- have_partition = true
- partition_by = ["updated_at_month"]
-}
-```
-
-
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/sink/HdfsFile.md
b/docs/en/connectors/sink/HdfsFile.md
index e40e3b7c33..c47d7d2f2a 100644
--- a/docs/en/connectors/sink/HdfsFile.md
+++ b/docs/en/connectors/sink/HdfsFile.md
@@ -89,11 +89,11 @@ Output data to hdfs file
| enable_header_write | boolean | no | false
| Only used when file_format_type is text,csv.<br/>
false:don't write header,true:write header.
[...]
| encoding | string | no | "UTF-8"
| Only used when file_format_type is
json,text,csv,xml.
[...]
| remote_user | string | no | -
| The remote user name of hdfs.
[...]
+| schema_evolution_enabled | boolean | no | false
| Enable schema evolution support for CDC pipelines.
When true, ADD/DROP/RENAME/MODIFY column events from the source are applied to
the sink without a job restart. Not supported for binary format.
[...]
| schema_save_mode | string | no |
CREATE_SCHEMA_WHEN_NOT_EXIST | Existing dir processing method
[...]
| data_save_mode | string | no | APPEND_DATA
| Existing data processing method
[...]
| multi_table_sink_replica | int | no | 1
| The replica number of sink writers used for each
table in a multi-table sink job.
[...]
| merge_update_event | boolean | no | false
| Only used when file_format_type is
canal_json,debezium_json or maxwell_json. When value is true, the UPDATE_AFTER
and UPDATE_BEFORE event will be merged into UPDATE event data
[...]
-| schema_evolution_enabled | boolean | no | false
| Enable schema evolution support for CDC pipelines.
When true, ADD/DROP/RENAME/MODIFY column events from the source are applied to
the sink without a job restart. Not supported for binary format. |
### Tips
@@ -114,6 +114,32 @@ Existing data processing method.
- APPEND_DATA: preserve dir, preserve data files
- ERROR_WHEN_DATA_EXISTS: when there is data files, an error is reported
+### schema_evolution_enabled [boolean]
+
+When set to `true`, the file sink handles CDC schema change events (ADD
COLUMN, DROP COLUMN, RENAME COLUMN, MODIFY COLUMN type) at runtime without
requiring a job restart. On each schema change the current output file is
closed and a new file is opened with the updated schema.
+
+**Supported formats:** All file formats except `binary`. Enabling this option
with `file_format_type = binary` will fail at job startup with a config
validation error.
+
+**Partition constraint:** When `have_partition = true`, dropping a column
listed in `partition_by` is not allowed and will fail fast. Partition columns
must remain stable across schema changes.
+
+**When `schema_evolution_enabled = false` (default):** If the upstream CDC
source has `schema-changes.enabled = true` and an `AlterTableEvent` arrives at
the sink, the job will throw immediately with an actionable error:
+> `Received AlterTableEvent but schema_evolution_enabled=false at this sink.
Either set schema_evolution_enabled=true to handle schema changes, or set
schema-changes.enabled=false at the CDC source to suppress them.`
+
+Users on the default CDC source config (`schema-changes.enabled = false`) are
completely unaffected.
+
+**Known limitation:** Schema changes are not atomic with checkpointing. If the
job crashes in the narrow window between file rotation and schema metadata
update, rows written after restore may use the pre-change schema. This is a
known architectural gap shared across other SeaTunnel sinks. For full
restart-with-DDL correctness, a follow-up CDC source fix is required (tracked
separately).
+
+Example usage in a CDC pipeline:
+
+```hocon
+HdfsFile {
+ fs.defaultFS = "hdfs://hadoopcluster"
+ path = "/tmp/seatunnel/cdc/${table_name}"
+ file_format_type = "parquet"
+ schema_evolution_enabled = true
+}
+```
+
### multi_table_sink_replica [int]
The replica number of sink writers used for each table in a multi-table sink
job. The default value is `1`; increase it only when each table needs more sink
writer parallelism.
@@ -299,34 +325,6 @@ Configure mount table in `core-site.xml`:
```
-### schema_evolution_enabled [boolean]
-
-When set to `true`, the file sink handles CDC schema change events (ADD
COLUMN, DROP COLUMN, RENAME COLUMN, MODIFY COLUMN type) at runtime without
requiring a job restart. On each schema change the current output file is
closed and a new file is opened with the updated schema.
-
-**Supported formats:** All file formats except `binary`. Enabling this option
with `file_format_type = binary` will fail at job startup with a config
validation error.
-
-**Partition constraint:** When `have_partition = true`, dropping a column
listed in `partition_by` is not allowed and will fail fast. Partition columns
must remain stable across schema changes.
-
-**When `schema_evolution_enabled = false` (default):** If the upstream CDC
source has `schema-changes.enabled = true` and an `AlterTableEvent` arrives at
the sink, the job will throw immediately with an actionable error:
-> `Received AlterTableEvent but schema_evolution_enabled=false at this sink.
Either set schema_evolution_enabled=true to handle schema changes, or set
schema-changes.enabled=false at the CDC source to suppress them.`
-
-Users on the default CDC source config (`schema-changes.enabled = false`) are
completely unaffected.
-
-**Known limitation:** Schema changes are not atomic with checkpointing. If the
job crashes in the narrow window between file rotation and schema metadata
update, rows written after restore may use the pre-change schema. This is a
known architectural gap shared across other SeaTunnel sinks. For full
restart-with-DDL correctness, a follow-up CDC source fix is required (tracked
separately).
-
-Example usage in a CDC pipeline:
-
-```hocon
-LocalFile {
- path = "/tmp/cdc/${table_name}"
- file_format_type = "parquet"
- schema_evolution_enabled = true
- have_partition = true
- partition_by = ["updated_at_month"]
-}
-```
-
-
## Changelog
<ChangeLog />
diff --git a/docs/en/connectors/source/FtpFile.md
b/docs/en/connectors/source/FtpFile.md
index faa9517a3b..18e22d344b 100644
--- a/docs/en/connectors/source/FtpFile.md
+++ b/docs/en/connectors/source/FtpFile.md
@@ -22,6 +22,7 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
- [x] [column projection](../../introduction/concepts/connector-v2-features.md)
- [x] [parallelism](../../introduction/concepts/connector-v2-features.md)
- [ ] [support user-defined
split](../../introduction/concepts/connector-v2-features.md)
+- [x] [support multiple table
read](../../introduction/concepts/connector-v2-features.md)
- [x] file format type
- [x] text
- [x] csv
@@ -52,6 +53,7 @@ If you use SeaTunnel Engine, It automatically integrated the
hadoop jar when you
| user | string | yes | -
|
| password | string | yes | -
|
| path | string | yes | -
|
+| tables_configs | list | no | -
|
| file_format_type | string | yes | -
|
| connection_mode | string | no | active_local
|
| remote_verification_enabled | boolean | no | true
|
@@ -114,6 +116,12 @@ The target ftp password is required
The source file path.
+### tables_configs [list]
+
+Configure multiple FTP source tables in one source block. Each item in
`tables_configs` has the same options as a
+single-table `FtpFile` source, such as `host`, `port`, `user`, `password`,
`path`, `file_format_type`, and `schema`.
+Use `schema.table` to set the table name sent downstream.
+
### remote_verification_enabled [boolean]
Whether to enable remote host verification for FTP data channels, default is
`true`.
@@ -580,6 +588,9 @@ Source plugin common parameters, please refer to [Source
Common Options](../comm
### Multiple Table
+Each entry in `tables_configs` is read as an independent table. Keep
connection and format options inside every entry
+unless all tables use the same values through an outer shared configuration.
+
```hocon
FtpFile {
diff --git a/docs/en/connectors/source/HdfsFile.md
b/docs/en/connectors/source/HdfsFile.md
index 514f1485de..c71079a696 100644
--- a/docs/en/connectors/source/HdfsFile.md
+++ b/docs/en/connectors/source/HdfsFile.md
@@ -52,6 +52,7 @@ Read data from hdfs file system.
| Name | Type | Required | Default
| Description
|
|----------------------------|---------|----------|-----------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| path | string | yes | -
| The source file path.
|
+| tables_configs | list | no | -
| Configure multiple HDFS source tables in one source block. Each item has the
same options as a single-table `HdfsFile` source, and can set its downstream
table name through `schema.table`.
|
| file_format_type | string | yes | -
| We supported as the following file types:`text` `csv` `parquet` `orc` `json`
`excel` `xml` `binary` `markdown`.Please note that, The final file name will
end with the file_format's suffix, the suffix of the text file is `txt`.
|
| fs.defaultFS | string | yes | -
| The hadoop cluster address that start with `hdfs://`, for example:
`hdfs://hadoopcluster`
|
| read_columns | list | no | -
| The read column list of the data source, user can use it to implement field
projection.The file type supported column projection as the following
shown:[text,json,csv,orc,parquet,excel,xml].Tips: If the user wants to use this
feature when reading `text` `json` `csv` files, the schema option must be
configured. |
@@ -128,6 +129,12 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+### tables_configs [list]
+
+Use `tables_configs` when one HDFS source needs to read multiple tables or
directories. Each item can define its own
+`path`, `file_format_type`, `schema`, and HDFS options. Set `schema.table`
when the downstream sink needs table-aware
+routing.
+
### delimiter/field_delimiter [string]
**delimiter** parameter will deprecate after version 2.3.5, please use
**field_delimiter** instead.
@@ -483,6 +490,10 @@ sink {
```
### Multiple Table
+
+Each item in `tables_configs` is read as a separate table. This is useful when
different HDFS directories should keep
+their own table name downstream.
+
```hocon
env {
parallelism = 1
diff --git a/docs/en/connectors/source/SqlServer-CDC.md
b/docs/en/connectors/source/SqlServer-CDC.md
index 56cb05c390..43063426d1 100644
--- a/docs/en/connectors/source/SqlServer-CDC.md
+++ b/docs/en/connectors/source/SqlServer-CDC.md
@@ -75,20 +75,16 @@ case-sensitive databases, make sure the configured
identifier case matches the d
| Name | Type | Required | Default |
Description
[...]
|-------------------------------------------|----------|----------|---------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
-| username | String | Yes | - |
Name of the database to use when connecting to the database server.
[...]
+| username | String | Yes | - |
Username to use when connecting to the SQL Server instance.
[...]
| password | String | Yes | - |
Password to use when connecting to the database server.
[...]
| database-names | List | Yes | - |
Database name of the database to monitor.
[...]
-| table-names | List | Yes | - |
Table name is a combination of schema name and table name
(databaseName.schemaName.tableName).
[...]
+| table-names | List | Yes | - |
Tables to monitor. Use the full table identifier:
`databaseName.schemaName.tableName`. This option is mutually exclusive with
`table-pattern`.
[...]
+| table-pattern | String | Yes | - |
Regular expression for tables to monitor. Use the full table identifier in the
pattern, for example `column_type_test\\.dbo\\..*`. This option is mutually
exclusive with `table-names`.
[...]
| table-names-config | List | No | - |
Table config list. for example: [{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]. The snapshotSplitColumn option must
be configured with a unique key. If a non-unique column is provided, the
configuration is ignored and SeaTunnel automatically selects an appropriate
split column internally.
[...]
| url | String | Yes | - |
URL has to be with database, like
"jdbc:sqlserver://localhost:1433;databaseName=test".
[...]
-| startup.mode | Enum | No | INITIAL |
Optional startup mode for SqlServer CDC consumer, valid enumerations are
"initial", "earliest", "latest", "timestamp" and "specific".
[...]
+| startup.mode | Enum | No | INITIAL |
Optional startup mode for SqlServer CDC consumer. Valid values are `initial`,
`earliest`, `latest`, and `timestamp`.<br/>`initial`: read the table snapshot
first and then continue with incremental changes.<br/>`earliest`: start from
the earliest available CDC LSN.<br/>`latest`: only read changes after the job
starts.<br/>`timestamp`: start from the LSN resolved from `startup.timestamp`.
[...]
| startup.timestamp | Long | No | - |
Start from the specified epoch timestamp (in milliseconds). This timestamp is
converted with `server-time-zone` when `startup.mode = timestamp`.<br/> **Note,
This option is required when** the **"startup.mode" option used
`'timestamp'`.**
[...]
-| startup.specific-offset.file | String | No | - |
Start from the specified binlog file name. <br/>**Note, This option is required
when the "startup.mode" option used `'specific'`.**
[...]
-| startup.specific-offset.pos | Long | No | - |
Start from the specified binlog file position.<br/>**Note, This option is
required when the "startup.mode" option used `'specific'`.**
[...]
| stop.mode | Enum | No | NEVER |
Optional stop mode for SqlServer CDC consumer, valid enumerations are "never".
[...]
-| stop.timestamp | Long | No | - |
Stop from the specified epoch timestamp (in milliseconds). <br/>**Note, This
option is required when the "stop.mode" option used `'timestamp'`.**
[...]
-| stop.specific-offset.file | String | No | - |
Stop from the specified binlog file name.<br/>**Note, This option is required
when the "stop.mode" option used `'specific'`.**
[...]
-| stop.specific-offset.pos | Long | No | - |
Stop from the specified binlog file position.<br/>**Note, This option is
required when the "stop.mode" option used `'specific'`.**
[...]
| incremental.parallelism | Integer | No | 1 |
The number of parallel readers in the incremental phase.
[...]
| snapshot.split.size | Integer | No | 8096 |
The split size (number of rows) of table snapshot, captured tables are split
into multiple splits when read the snapshotof table.
[...]
| snapshot.fetch.size | Integer | No | 1024 |
The maximum fetch size for per poll when read table snapshot.
[...]
@@ -138,6 +134,12 @@ case-sensitive databases, make sure the configured
identifier case matches the d
> EXEC sys.sp_cdc_enable_db;<br/>
> SELECT name, is_tracked_by_cdc FROM sys.tables WHERE name = 'table'; --
> table Replace with the name of the table you want to check
+5. Enable CDC for each table that SeaTunnel should read
+
+> USE TestDB; -- Replace with the actual database name <br/>
+> EXEC sys.sp_cdc_enable_table @source_schema = N'dbo', @source_name =
N'full_types', @role_name = NULL, @supports_net_changes = 0;<br/>
+> SELECT name, is_tracked_by_cdc FROM sys.tables WHERE name = 'full_types';
+
## Task Example
### initiali read Simple
@@ -244,6 +246,76 @@ sink {
}
```
+### Read multiple tables
+
+Use full table identifiers in `table-names`. Downstream sinks that support
multi-table routing can use the source table
+metadata to write each table to its own target table.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ SqlServer-CDC {
+ plugin_output = "customers"
+ username = "sa"
+ password = "Password!"
+ database-names = ["column_type_test"]
+ table-names = [
+ "column_type_test.dbo.full_types",
+ "column_type_test.dbo.full_types_2"
+ ]
+ url = "jdbc:sqlserver://sqlserver-host:1433;databaseName=column_type_test"
+ }
+}
+
+sink {
+ Jdbc {
+ plugin_input = "customers"
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+ url =
"jdbc:sqlserver://sqlserver-host:1433;databaseName=column_type_test;encrypt=false"
+ user = "sa"
+ password = "Password!"
+ generate_sink_sql = true
+ database = "column_type_test"
+ schema = "dbo"
+ tablePrefix = "sink_"
+ primary_keys = ["id"]
+ }
+}
+```
+
+### Start from a timestamp
+
+When `startup.mode = "timestamp"`, `startup.timestamp` is an epoch millisecond
value. Set `server-time-zone`
+explicitly if the database server time zone differs from the JVM time zone.
+
+```hocon
+source {
+ SqlServer-CDC {
+ plugin_output = "customers"
+ username = "sa"
+ password = "Password!"
+ database-names = ["column_type_test"]
+ table-names = ["column_type_test.dbo.full_types_custom_primary_key"]
+ url = "jdbc:sqlserver://sqlserver-host:1433;databaseName=column_type_test"
+ startup.mode = "timestamp"
+ startup.timestamp = 1719820800000
+ server-time-zone = "UTC"
+ exactly_once = true
+ table-names-config = [
+ {
+ table = "column_type_test.dbo.full_types_custom_primary_key"
+ primaryKeys = ["id"]
+ }
+ ]
+ }
+}
+```
+
### Schema change event filtering
When `schema-changes.enabled = true`, you can further control which schema
change event types are
@@ -276,7 +348,7 @@ source {
}
```
-**Data handling when `drop.column` is excluded. For a retained **NOT NULL**
column the `NULL` write is rejected
+**Data handling when `drop.column` is excluded.** For a retained **NOT NULL**
column the `NULL` write is rejected
by the sink, so excluding `drop.column` for a NOT NULL column that the source
has stopped supplying
will fail at the sink.
diff --git a/docs/zh/connectors/sink/FtpFile.md
b/docs/zh/connectors/sink/FtpFile.md
index 571fcb5843..89ed55299e 100644
--- a/docs/zh/connectors/sink/FtpFile.md
+++ b/docs/zh/connectors/sink/FtpFile.md
@@ -26,6 +26,8 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
默认情况下，我们使用两阶段提交（2PC）来确保`精确一次`
+- [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md)
+
- [x] 文件格式
- [x] text
- [x] csv
@@ -78,8 +80,10 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
| parquet_avro_write_fixed_as_int96 | array | 否 | -
| 仅在 `file_format` 为 `parquet` 时使用。
|
| enable_header_write | boolean | 否 | false
| 仅在 `file_format_type` 为 `text`、`csv` 时使用。<br/>
`false`：不写入表头，`true`：写入表头。 |
| encoding | string | 否 | "UTF-8"
| 仅在 `file_format_type` 为 `json`、`text`、`csv`、`xml` 时使用。
|
+| schema_evolution_enabled | boolean | 否 | false
| 开启 Schema 演变支持，适用于 CDC 管道。为 true 时，来自上游的
ADD/DROP/RENAME/MODIFY 列事件无需重启作业即可应用到 Sink。不支持 binary 格式。 |
| schema_save_mode | string | 否 |
CREATE_SCHEMA_WHEN_NOT_EXIST | 现有目录处理方法
|
| data_save_mode | string | 否 | APPEND_DATA
| 现有数据处理方法
|
+| multi_table_sink_replica | int | 否 | 1
| 多表写入时，每张表对应的 Sink Writer 副本数。
|
### host [string]
@@ -291,6 +295,39 @@ Sink 插件的通用参数，请参考[Sink通用选项](../common-options/sink-
- APPEND_DATA（追加数据）：保留目录和数据文件。
- ERROR_WHEN_DATA_EXISTS（数据存在时报错）：当存在数据文件时，报告错误。
+### schema_evolution_enabled [boolean]
+
+设置为 `true` 时，文件 Sink 可在运行时处理 CDC Schema 变更事件（ADD COLUMN、DROP COLUMN、RENAME
COLUMN、MODIFY COLUMN 类型），无需重启作业。每次 Schema 变更时，当前输出文件会被关闭，并以新 Schema 打开一个新文件。
+
+**支持的格式：** 除 `binary` 外的所有文件格式。将此选项与 `file_format_type = binary`
一起使用时，作业启动时会抛出配置校验错误。
+
+**分区约束：** 当 `have_partition = true` 时，不允许删除 `partition_by`
中列出的列，违反时会立即抛出异常。分区列在 Schema 变更过程中必须保持稳定。
+
+**当 `schema_evolution_enabled = false`（默认值）时：** 若上游 CDC Source 配置了
`schema-changes.enabled = true` 且 Sink 收到 `AlterTableEvent`，作业会立即抛出如下错误：
+> `Received AlterTableEvent but schema_evolution_enabled=false at this sink.
Either set schema_evolution_enabled=true to handle schema changes, or set
schema-changes.enabled=false at the CDC source to suppress them.`
+
+使用默认 CDC Source 配置（`schema-changes.enabled = false`）的用户不受影响。
+
+**已知限制：** Schema 变更与 Checkpoint 不是原子操作。若作业在文件轮转与 Schema
元数据更新之间的窗口期崩溃，恢复后写入的数据行可能使用变更前的 Schema。这是与其他 SeaTunnel Sink 共同存在的已知架构限制。完整的重启后
DDL 正确性支持需要配套的 CDC Source 修复（另行跟踪）。
+
+CDC 管道中的使用示例：
+
+```hocon
+FtpFile {
+ host = "xxx.xxx.xxx.xxx"
+ port = 21
+ user = "username"
+ password = "password"
+ path = "/data/ftp/cdc/${table_name}"
+ file_format_type = "parquet"
+ schema_evolution_enabled = true
+}
+```
+
+### multi_table_sink_replica [int]
+
+多表写入时，每张表对应的 Sink Writer 副本数。默认值为 `1`；只有单表写入压力较大、需要更多写入并行度时再调大。
+
## 示例
对于文本文件格式的简易配置
@@ -337,7 +374,8 @@ FtpFile {
```
-当我们的数据源端是多个表，并且希望将不同的数据按照不同的表达式存储到不同的目录时，我们可以按照这种方式进行配置。
+当上游是多张表，并且希望每张表写入不同的 FTP 目录时，可以在 `path` 中使用 `${table_name}`。
+`schema_save_mode` 和 `data_save_mode` 用于控制写入前如何处理已存在的目录和文件。
```hocon
@@ -366,34 +404,6 @@ FtpFile {
```
-### schema_evolution_enabled [boolean]
-
-设置为 `true` 时，文件 Sink 可在运行时处理 CDC Schema 变更事件（ADD COLUMN、DROP COLUMN、RENAME
COLUMN、MODIFY COLUMN 类型），无需重启作业。每次 Schema 变更时，当前输出文件会被关闭，并以新 Schema 打开一个新文件。
-
-**支持的格式：** 除 `binary` 外的所有文件格式。将此选项与 `file_format_type = binary`
一起使用时，作业启动时会抛出配置校验错误。
-
-**分区约束：** 当 `have_partition = true` 时，不允许删除 `partition_by`
中列出的列，违反时会立即抛出异常。分区列在 Schema 变更过程中必须保持稳定。
-
-**当 `schema_evolution_enabled = false`（默认值）时：** 若上游 CDC Source 配置了
`schema-changes.enabled = true` 且 Sink 收到 `AlterTableEvent`，作业会立即抛出如下错误：
-> `Received AlterTableEvent but schema_evolution_enabled=false at this sink.
Either set schema_evolution_enabled=true to handle schema changes, or set
schema-changes.enabled=false at the CDC source to suppress them.`
-
-使用默认 CDC Source 配置（`schema-changes.enabled = false`）的用户不受影响。
-
-**已知限制：** Schema 变更与 Checkpoint 不是原子操作。若作业在文件轮转与 Schema
元数据更新之间的窗口期崩溃，恢复后写入的数据行可能使用变更前的 Schema。这是与其他 SeaTunnel Sink 共同存在的已知架构限制。完整的重启后
DDL 正确性支持需要配套的 CDC Source 修复（另行跟踪）。
-
-CDC 管道中的使用示例：
-
-```hocon
-LocalFile {
- path = "/tmp/cdc/${table_name}"
- file_format_type = "parquet"
- schema_evolution_enabled = true
- have_partition = true
- partition_by = ["updated_at_month"]
-}
-```
-
-
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/sink/HdfsFile.md
b/docs/zh/connectors/sink/HdfsFile.md
index 477a51df81..becfc8eae7 100644
--- a/docs/zh/connectors/sink/HdfsFile.md
+++ b/docs/zh/connectors/sink/HdfsFile.md
@@ -80,11 +80,11 @@ import ChangeLog from
'../changelog/connector-file-hadoop.md';
| max_rows_in_memory | int | 否 | -
| 仅当 file_format 为 excel 时使用。当文件格式为 Excel 时，可以缓存在内存中的最大数据项数。
|
| sheet_name | string | 否 | Sheet${Random number}
| 仅当 file_format 为 excel 时使用。将工作簿的表写入指定的表名
|
| remote_user | string | 否 | -
| Hdfs的远端用户名。
|
+| schema_evolution_enabled | boolean | 否 | false
| 开启 Schema 演变支持，适用于 CDC 管道。为 true 时，来自上游的
ADD/DROP/RENAME/MODIFY 列事件无需重启作业即可应用到 Sink。不支持 binary 格式。
|
| schema_save_mode | string | 否 |
CREATE_SCHEMA_WHEN_NOT_EXIST | 现有目录处理方式
|
| data_save_mode | string | 否 | APPEND_DATA
| 现有数据处理方式
|
| multi_table_sink_replica | int | 否 | 1
| 多表写入时，每张表对应的 Sink Writer 副本数。
|
| merge_update_event | boolean | 否 | false
| 仅当file_format_type为canal_json、debezium_json、maxwell_json.
|
-| schema_evolution_enabled | boolean | 否 | false
| 开启 Schema 演变支持，适用于 CDC 管道。为 true 时，来自上游的
ADD/DROP/RENAME/MODIFY 列事件无需重启作业即可应用到 Sink。不支持 binary 格式。 |
### 提示
@@ -107,6 +107,32 @@ import ChangeLog from
'../changelog/connector-file-hadoop.md';
- APPEND_DATA：保留目录，保留数据文件
- ERROR_WHEN_DATA_EXISTS：当有数据文件时，会报告错误
+### schema_evolution_enabled [boolean]
+
+设置为 `true` 时，文件 Sink 可在运行时处理 CDC Schema 变更事件（ADD COLUMN、DROP COLUMN、RENAME
COLUMN、MODIFY COLUMN 类型），无需重启作业。每次 Schema 变更时，当前输出文件会被关闭，并以新 Schema 打开一个新文件。
+
+**支持的格式：** 除 `binary` 外的所有文件格式。将此选项与 `file_format_type = binary`
一起使用时，作业启动时会抛出配置校验错误。
+
+**分区约束：** 当 `have_partition = true` 时，不允许删除 `partition_by`
中列出的列，违反时会立即抛出异常。分区列在 Schema 变更过程中必须保持稳定。
+
+**当 `schema_evolution_enabled = false`（默认值）时：** 若上游 CDC Source 配置了
`schema-changes.enabled = true` 且 Sink 收到 `AlterTableEvent`，作业会立即抛出如下错误：
+> `Received AlterTableEvent but schema_evolution_enabled=false at this sink.
Either set schema_evolution_enabled=true to handle schema changes, or set
schema-changes.enabled=false at the CDC source to suppress them.`
+
+使用默认 CDC Source 配置（`schema-changes.enabled = false`）的用户不受影响。
+
+**已知限制：** Schema 变更与 Checkpoint 不是原子操作。若作业在文件轮转与 Schema
元数据更新之间的窗口期崩溃，恢复后写入的数据行可能使用变更前的 Schema。这是与其他 SeaTunnel Sink 共同存在的已知架构限制。完整的重启后
DDL 正确性支持需要配套的 CDC Source 修复（另行跟踪）。
+
+CDC 管道中的使用示例：
+
+```hocon
+HdfsFile {
+ fs.defaultFS = "hdfs://hadoopcluster"
+ path = "/tmp/seatunnel/cdc/${table_name}"
+ file_format_type = "parquet"
+ schema_evolution_enabled = true
+}
+```
+
### multi_table_sink_replica [int]
多表写入时，每张表对应的 Sink Writer 副本数。默认值为 `1`；只有单表写入压力较大、需要更多写入并行度时再调大。
@@ -295,34 +321,6 @@ HdfsFile {
```
-### schema_evolution_enabled [boolean]
-
-设置为 `true` 时，文件 Sink 可在运行时处理 CDC Schema 变更事件（ADD COLUMN、DROP COLUMN、RENAME
COLUMN、MODIFY COLUMN 类型），无需重启作业。每次 Schema 变更时，当前输出文件会被关闭，并以新 Schema 打开一个新文件。
-
-**支持的格式：** 除 `binary` 外的所有文件格式。将此选项与 `file_format_type = binary`
一起使用时，作业启动时会抛出配置校验错误。
-
-**分区约束：** 当 `have_partition = true` 时，不允许删除 `partition_by`
中列出的列，违反时会立即抛出异常。分区列在 Schema 变更过程中必须保持稳定。
-
-**当 `schema_evolution_enabled = false`（默认值）时：** 若上游 CDC Source 配置了
`schema-changes.enabled = true` 且 Sink 收到 `AlterTableEvent`，作业会立即抛出如下错误：
-> `Received AlterTableEvent but schema_evolution_enabled=false at this sink.
Either set schema_evolution_enabled=true to handle schema changes, or set
schema-changes.enabled=false at the CDC source to suppress them.`
-
-使用默认 CDC Source 配置（`schema-changes.enabled = false`）的用户不受影响。
-
-**已知限制：** Schema 变更与 Checkpoint 不是原子操作。若作业在文件轮转与 Schema
元数据更新之间的窗口期崩溃，恢复后写入的数据行可能使用变更前的 Schema。这是与其他 SeaTunnel Sink 共同存在的已知架构限制。完整的重启后
DDL 正确性支持需要配套的 CDC Source 修复（另行跟踪）。
-
-CDC 管道中的使用示例：
-
-```hocon
-LocalFile {
- path = "/tmp/cdc/${table_name}"
- file_format_type = "parquet"
- schema_evolution_enabled = true
- have_partition = true
- partition_by = ["updated_at_month"]
-}
-```
-
-
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/FtpFile.md
b/docs/zh/connectors/source/FtpFile.md
index 9cd145cad0..eaa0925cb0 100644
--- a/docs/zh/connectors/source/FtpFile.md
+++ b/docs/zh/connectors/source/FtpFile.md
@@ -22,6 +22,7 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
- [x] [列投影](../../introduction/concepts/connector-v2-features.md)
- [x] [并行度](../../introduction/concepts/connector-v2-features.md)
- [ ] [支持用户自定义分片](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持多表读取](../../introduction/concepts/connector-v2-features.md)
- [x] 文件格式类型
- [x] 文本
- [x] CSV
@@ -50,6 +51,7 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
| user | string | 是 | - |
| password | string | 是 | - |
| path | string | 是 | - |
+| tables_configs | list | 否 | - |
| file_format_type | string | 是 | - |
| connection_mode | string | 否 | active_local |
| remote_verification_enabled | boolean | 否 | true |
@@ -110,6 +112,12 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
源文件路径。
+### tables_configs [list]
+
+在一个 Source 块中配置多张 FTP 源表。`tables_configs` 中每一项都使用与单表 `FtpFile` Source 相同的参数，
+例如 `host`、`port`、`user`、`password`、`path`、`file_format_type` 和 `schema`。可通过
`schema.table`
+指定传递给下游的表名。
+
### remote_verification_enabled [boolean]
是否启用FTP数据通道的远程主机验证。默认值为 `true`。
@@ -549,6 +557,8 @@ compare_mode = "len_mtime"
### 多表配置
+`tables_configs` 中的每一项都会作为一张独立表读取。除非通过外层共享配置复用相同参数，否则建议在每一项中明确写出连接和格式参数。
+
```hocon
FtpFile {
diff --git a/docs/zh/connectors/source/HdfsFile.md
b/docs/zh/connectors/source/HdfsFile.md
index 48814806db..04da70a457 100644
--- a/docs/zh/connectors/source/HdfsFile.md
+++ b/docs/zh/connectors/source/HdfsFile.md
@@ -25,7 +25,7 @@ import ChangeLog from '../changelog/connector-file-hadoop.md';
- [x] [列投影](../../introduction/concepts/connector-v2-features.md)
- [x] [并行度](../../introduction/concepts/connector-v2-features.md)
- [ ] [支持用户定义分片](../../introduction/concepts/connector-v2-features.md)
-- [x] [支持多表读](../../introduction/concepts/connector-v2-features.md)
+- [x] [支持多表读取](../../introduction/concepts/connector-v2-features.md)
- [x] 文件格式类型
- [x] text
- [x] csv
@@ -52,6 +52,7 @@ import ChangeLog from '../changelog/connector-file-hadoop.md';
| 名称 | 类型 | 是否必须 | 默认值 | 描述
|
|----------------------------|---------|------|---------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| path | string | 是 | - | 源文件路径。
|
+| tables_configs | list | 否 | - | 在一个
Source 块中配置多张 HDFS 源表。每一项都使用与单表 `HdfsFile` Source 相同的参数，并可通过 `schema.table`
设置传递给下游的表名。 |
| file_format_type | string | 是 | - |
我们支持以下文件类型：`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary`
`markdown`。请注意，最终文件名将以文件格式的后缀结束，文本文件的后缀是 `txt`。
|
| fs.defaultFS | string | 是 | - | 以
`hdfs://` 开头的 hadoop 集群地址，例如：`hdfs://hadoopcluster`
|
| read_columns | list | 否 | - |
数据源的读取列列表，用户可以使用它来实现字段投影。支持列投影的文件类型如下所示：[text,json,csv,orc,parquet,excel,xml]。提示：如果用户想在读取
`text` `json` `csv` 文件时使用此功能，必须配置 schema 选项。
|
@@ -128,6 +129,11 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
注意：Markdown 格式仅支持读取，不支持写入。
+### tables_configs [list]
+
+当一个 HDFS Source 需要读取多张表或多个目录时，可以使用 `tables_configs`。每一项都可以单独配置
+`path`、`file_format_type`、`schema` 以及 HDFS 相关参数。下游 Sink 需要按表路由时，请设置
`schema.table`。
+
### delimiter/field_delimiter [string]
**delimiter** 参数将在 2.3.5 版本后弃用，请使用 **field_delimiter** 代替。
@@ -500,6 +506,9 @@ sink {
```
### 多表配置
+
+`tables_configs` 中的每一项都会作为一张独立表读取。不同 HDFS 目录需要在下游保留各自表名时，可以使用这种配置。
+
```hocon
env {
parallelism = 1
diff --git a/docs/zh/connectors/source/SqlServer-CDC.md
b/docs/zh/connectors/source/SqlServer-CDC.md
index c7453e8d14..db47d86f90 100644
--- a/docs/zh/connectors/source/SqlServer-CDC.md
+++ b/docs/zh/connectors/source/SqlServer-CDC.md
@@ -73,20 +73,16 @@ Sql Server CDC 连接器允许从 SqlServer 数据库读取快照数据和增量
| 名称 | 类型 | 是否必填 | 默认值 | 描述
|
| ---------------------------------------------- | -------- | -------- |
------- |
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
-| username | String | 是 | -
| 连接数据库服务器时使用的数据库名称。
|
+| username | String | 是 | -
| 连接 SQL Server 实例时使用的用户名。
|
| password | String | 是 | -
| 连接数据库服务器时使用的密码。
|
| database-names | List | 是 | -
| 要监控的数据库名称。
|
-| table-names | List | 是 | -
| 表名是模式名和表名的组合 (databaseName.schemaName.tableName)。
|
+| table-names | List | 是 | -
| 要监控的表，使用完整表标识：`databaseName.schemaName.tableName`。该参数与 `table-pattern` 互斥。
|
+| table-pattern | String | 是 | -
| 要监控的表名正则表达式，需要匹配完整表标识，例如 `column_type_test\\.dbo\\..*`。该参数与 `table-names`
互斥。
|
| table-names-config | List | 否 | -
| 表配置列表。例如：[{"table": "db1.schema1.table1","primaryKeys":
["key1"],"snapshotSplitColumn": "key2"}]。 snapshotSplitColumn
选项必须配置为唯一键(主键或唯一索引)。 如果指定了非唯一列，该配置将被忽略，SeaTunnel 会在内部自动选择合适的拆分列。
|
| url | String | 是 | -
| URL 必须包含数据库，如 "jdbc:sqlserver://localhost:1433;databaseName=test"。
|
-| startup.mode | Enum | 否 |
INITIAL | SqlServer CDC 消费者的可选启动模式，有效枚举为
"initial"、"earliest"、"latest"、"timestamp" 和 "specific"。
|
+| startup.mode | Enum | 否 |
INITIAL | SqlServer CDC 消费者的可选启动模式，有效值为 `initial`、`earliest`、`latest` 和
`timestamp`。<br/>`initial`：先读取表快照，再继续读取增量变更。<br/>`earliest`：从最早可用的 CDC LSN
开始读取。<br/>`latest`：只读取作业启动后的变更。<br/>`timestamp`：从 `startup.timestamp` 解析出的 LSN
开始读取。
|
| startup.timestamp | Long | 否 | -
| 从指定的纪元时间戳（以毫秒为单位）开始。当 `startup.mode = timestamp` 时，该时间戳会按 `server-time-zone`
转换。<br/> **注意，当 "startup.mode" 选项使用 `'timestamp'` 时，此选项是必需的。**
|
-| startup.specific-offset.file | String | 否 | -
| 从指定的 binlog 文件名开始。<br/>**注意，当 "startup.mode" 选项使用 `'specific'` 时，此选项是必需的。**
|
-| startup.specific-offset.pos | Long | 否 | -
| 从指定的 binlog 文件位置开始。<br/>**注意，当 "startup.mode" 选项使用 `'specific'` 时，此选项是必需的。**
|
| stop.mode | Enum | 否 | NEVER
| SqlServer CDC 消费者的可选停止模式，有效枚举为 "never"。
|
-| stop.timestamp | Long | 否 | -
| 在指定的纪元时间戳（以毫秒为单位）停止。<br/>**注意，当 "stop.mode" 选项使用 `'timestamp'` 时，此选项是必需的。**
|
-| stop.specific-offset.file | String | 否 | -
| 在指定的 binlog 文件名停止。<br/>**注意，当 "stop.mode" 选项使用 `'specific'` 时，此选项是必需的。**
|
-| stop.specific-offset.pos | Long | 否 | -
| 在指定的 binlog 文件位置停止。<br/>**注意，当 "stop.mode" 选项使用 `'specific'` 时，此选项是必需的。**
|
| incremental.parallelism | Integer | 否 | 1
| 增量阶段中并行读取器的数量。
|
| snapshot.split.size | Integer | 否 | 8096
| 表快照的分割大小（行数），读取表快照时，捕获的表会被分割为多个分割。
|
| snapshot.fetch.size | Integer | 否 | 1024
| 读取表快照时每次轮询的最大获取大小。
|
@@ -136,6 +132,12 @@ Sql Server CDC 连接器允许从 SqlServer 数据库读取快照数据和增量
> EXEC sys.sp_cdc_enable_db;<br/>
> SELECT name, is_tracked_by_cdc FROM sys.tables WHERE name = 'table'; --
> table 替换为您要检查的表名
+5. 为 SeaTunnel 需要读取的每张表启用 CDC
+
+> USE TestDB; -- 替换为实际的数据库名称 <br/>
+> EXEC sys.sp_cdc_enable_table @source_schema = N'dbo', @source_name =
N'full_types', @role_name = NULL, @supports_net_changes = 0;<br/>
+> SELECT name, is_tracked_by_cdc FROM sys.tables WHERE name = 'full_types';
+
## 任务示例
### 初始读取简单示例
@@ -242,6 +244,75 @@ sink {
}
```
+### 读取多张表
+
+在 `table-names` 中使用完整表标识。下游支持多表路由的 Sink 可以根据源表元数据，将不同源表写入不同目标表。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ SqlServer-CDC {
+ plugin_output = "customers"
+ username = "sa"
+ password = "Password!"
+ database-names = ["column_type_test"]
+ table-names = [
+ "column_type_test.dbo.full_types",
+ "column_type_test.dbo.full_types_2"
+ ]
+ url = "jdbc:sqlserver://sqlserver-host:1433;databaseName=column_type_test"
+ }
+}
+
+sink {
+ Jdbc {
+ plugin_input = "customers"
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+ url =
"jdbc:sqlserver://sqlserver-host:1433;databaseName=column_type_test;encrypt=false"
+ user = "sa"
+ password = "Password!"
+ generate_sink_sql = true
+ database = "column_type_test"
+ schema = "dbo"
+ tablePrefix = "sink_"
+ primary_keys = ["id"]
+ }
+}
+```
+
+### 从指定时间戳开始读取
+
+当 `startup.mode = "timestamp"` 时，`startup.timestamp` 是毫秒级 Unix 时间戳。如果数据库服务器时区与
JVM
+时区不同，请显式设置 `server-time-zone`。
+
+```hocon
+source {
+ SqlServer-CDC {
+ plugin_output = "customers"
+ username = "sa"
+ password = "Password!"
+ database-names = ["column_type_test"]
+ table-names = ["column_type_test.dbo.full_types_custom_primary_key"]
+ url = "jdbc:sqlserver://sqlserver-host:1433;databaseName=column_type_test"
+ startup.mode = "timestamp"
+ startup.timestamp = 1719820800000
+ server-time-zone = "UTC"
+ exactly_once = true
+ table-names-config = [
+ {
+ table = "column_type_test.dbo.full_types_custom_primary_key"
+ primaryKeys = ["id"]
+ }
+ ]
+ }
+}
+```
+
### Schema change 事件过滤
当 `schema-changes.enabled = true` 时，可通过 `schema-changes.include` /
`schema-changes.exclude` 进一步
@@ -274,7 +345,7 @@ source {
}
```
-**排除 `drop.column` 时的数据处理方式。对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
+**排除 `drop.column` 时的数据处理方式。** 对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
NOT NULL 列排除 `drop.column` 会在 sink 端失败。
## 变更日志
