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 34940bf66c [Feature][Connector-V2][CDC] Add include/exclude filtering
for schema change event types #11044 (#11108)
34940bf66c is described below
commit 34940bf66cd4fd6b07f28fe4b5eb3e6211587b71
Author: Shuai Liu <[email protected]>
AuthorDate: Thu Jun 18 08:41:31 2026 +0800
[Feature][Connector-V2][CDC] Add include/exclude filtering for schema
change event types #11044 (#11108)
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
docs/en/connectors/source/MySQL-CDC.md | 39 +++
docs/en/connectors/source/Oracle-CDC.md | 38 +++
docs/en/connectors/source/SqlServer-CDC.md | 39 +++
docs/zh/connectors/source/MySQL-CDC.md | 38 +++
docs/zh/connectors/source/Oracle-CDC.md | 37 +++
docs/zh/connectors/source/SqlServer-CDC.md | 38 +++
.../connectors/cdc/base/option/SourceOptions.java | 25 ++
.../cdc/base/schema/SchemaChangeEventFilter.java | 174 ++++++++++++
.../cdc/base/schema/SchemaChangeEventType.java | 95 +++++++
.../source/BaseChangeStreamTableSourceFactory.java | 3 +
.../row/SeaTunnelRowDebeziumDeserializeSchema.java | 18 ++
.../base/schema/SchemaChangeEventFilterTest.java | 308 +++++++++++++++++++++
.../cdc/mysql/source/MySqlIncrementalSource.java | 2 +
.../source/MySqlIncrementalSourceFactory.java | 2 +
.../cdc/oracle/source/OracleIncrementalSource.java | 2 +
.../source/OracleIncrementalSourceFactory.java | 4 +-
.../source/SqlServerIncrementalSource.java | 2 +
.../source/SqlServerIncrementalSourceFactory.java | 4 +-
.../cdc/mysql/MysqlCDCWithSchemaChangeIT.java | 121 ++++++++
.../src/test/resources/ddl/add_columns_filter.sql | 24 ++
.../src/test/resources/ddl/drop_columns_filter.sql | 29 ++
.../src/test/resources/ddl/shop.sql | 8 +
...ysqlcdc_to_mysql_with_schema_change_filter.conf | 53 ++++
23 files changed, 1101 insertions(+), 2 deletions(-)
diff --git a/docs/en/connectors/source/MySQL-CDC.md
b/docs/en/connectors/source/MySQL-CDC.md
index 2210231123..5ee7a0c647 100644
--- a/docs/en/connectors/source/MySQL-CDC.md
+++ b/docs/en/connectors/source/MySQL-CDC.md
@@ -215,6 +215,8 @@ When an initial consistent snapshot is made for large
databases, your establishe
| exactly_once | Boolean | No | false |
Enable exactly once semantic.
[...]
| format | Enum | No | DEFAULT |
Optional output format for MySQL CDC, valid enumerations are
`DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`.
[...]
| schema-changes.enabled | Boolean | No | false |
Schema evolution is disabled by default. Now we only support `add column`、`drop
column`、`rename column` and `modify column`.
[...]
+| schema-changes.include | List | No | - |
Only the listed schema change event types are sent downstream (when
`schema-changes.enabled = true`). Empty means all are eligible. See [Schema
change event filtering](#schema-change-event-filtering).
[...]
+| schema-changes.exclude | List | No | - |
Schema change event types listed here are NOT sent downstream. Applied after
`schema-changes.include`; exclude wins on conflict. See [Schema change event
filtering](#schema-change-event-filtering).
[...]
| debezium | Config | No | - |
Pass-through [Debezium's
properties](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/mysql.adoc#connector-properties)
to Debezium Embedded Engine which is used to capture data changes from MySQL
server.
[...]
| int_type_narrowing | Boolean | No | true |
Int type narrowing, if true, the tinyint(1) type will be narrowed to the
boolean type if without loss of precision. Support for MySQL at now. Please
refer to `int_type_narrowing` below
[...]
| common-options | | no | - |
Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
[...]
@@ -340,6 +342,43 @@ sink {
}
```
+
+### Schema change event filtering
+
+When `schema-changes.enabled = true`, you can further control which schema
change event types are
+propagated downstream using `schema-changes.include` /
`schema-changes.exclude`.
+
+Use these SeaTunnel-owned canonical names:
+
+| Canonical name | Operation
|
+|-----------------|---------------------------------------------------------------------------|
+| `add.column` | add a column
|
+| `drop.column` | drop a column
|
+| `modify.column` | change a column's type/attributes, name unchanged
|
+| `change.column` | rename a column, optionally re-type
|
+| `update.columns`| group alias for all four column-level changes above
|
+
+Precedence is deterministic:
+
+1. if `schema-changes.include` is set, only included event types are eligible;
+2. `schema-changes.exclude` is then applied;
+3. **exclude wins** when a type appears in both lists.
+
+```hocon
+source {
+ MySQL-CDC {
+ # ...
+ schema-changes.enabled = true
+ schema-changes.include = ["add.column", "drop.column"]
+ schema-changes.exclude = ["change.column"]
+ }
+}
+```
+
+**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.
+
### Support table-pattern for multi-table reading
> `table-pattern` and `table-names` are mutually exclusive
diff --git a/docs/en/connectors/source/Oracle-CDC.md
b/docs/en/connectors/source/Oracle-CDC.md
index c67fec3a1a..5393335b44 100644
--- a/docs/en/connectors/source/Oracle-CDC.md
+++ b/docs/en/connectors/source/Oracle-CDC.md
@@ -254,6 +254,8 @@ exit;
| skip_analyze | Boolean | No | false |
Skip the analysis of table count in full stage.In this scenario, you schedule
analysis table sql to update related table statistics periodically or your
table data does not change frequently
[...]
| format | Enum | No | DEFAULT |
Optional output format for Oracle CDC, valid enumerations are
`DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`.
[...]
| schema-changes.enabled | Boolean | No | false |
Schema evolution is disabled by default. Now we only support `add column`、`drop
column`、`rename column` and `modify column`.
[...]
+| schema-changes.include | List | No | -
| Only the listed schema change event types are sent downstream (when
`schema-changes.enabled = true`). Empty means all are eligible. See [Schema
change event filtering](#schema-change-event-filtering).
[...]
+| schema-changes.exclude | List | No | -
| Schema change event types listed here are NOT sent downstream. Applied after
`schema-changes.include`; exclude wins on conflict. See [Schema change event
filtering](#schema-change-event-filtering).
[...]
| debezium | Config | No | - |
Pass-through [Debezium's
properties](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/oracle.adoc#connector-properties)
to Debezium Embedded Engine which is used to capture data changes from Oracle
server.
[...]
| common-options | | no | - |
Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details
[...]
| decimal_type_narrowing | Boolean | No | true
| Decimal type narrowing, if true, the decimal type will be narrowed to
the int or long type if without loss of precision. Only support for Oracle at
now. Please refer to `decimal_type_narrowing` below
[...]
@@ -363,6 +365,42 @@ source {
}
```
+### Schema change event filtering
+
+When `schema-changes.enabled = true`, you can further control which schema
change event types are
+propagated downstream using `schema-changes.include` /
`schema-changes.exclude`.
+
+Use these SeaTunnel-owned canonical names:
+
+| Canonical name | Operation |
+|------------------|------------------------------------------------------|
+| `add.column` | add a column |
+| `drop.column` | drop a column |
+| `modify.column` | change a column's type/attributes, name unchanged |
+| `change.column` | rename a column, optionally re-type |
+| `update.columns` | group alias for all four column-level changes above |
+
+Precedence is deterministic:
+
+1. if `schema-changes.include` is set, only included event types are eligible;
+2. `schema-changes.exclude` is then applied;
+3. **exclude wins** when a type appears in both lists.
+
+```hocon
+source {
+ Oracle-CDC {
+ # ...
+ schema-changes.enabled = true
+ schema-changes.include = ["add.column", "drop.column"]
+ schema-changes.exclude = ["change.column"]
+ }
+}
+```
+
+**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.
+
### Support debezium-compatible format send to kafka
> Must be used with kafka connector sink, see [compatible debezium
> format](../formats/cdc-compatible-debezium-json.md) for details
diff --git a/docs/en/connectors/source/SqlServer-CDC.md
b/docs/en/connectors/source/SqlServer-CDC.md
index 5ca2cfb2ce..56cb05c390 100644
--- a/docs/en/connectors/source/SqlServer-CDC.md
+++ b/docs/en/connectors/source/SqlServer-CDC.md
@@ -104,6 +104,9 @@ case-sensitive databases, make sure the configured
identifier case matches the d
| exactly_once | Boolean | No | false |
Enable exactly once semantic.
[...]
| debezium.* | config | No | - |
Pass-through Debezium's properties to Debezium Embedded Engine which is used to
capture data changes from SqlServer server.<br/>See more about<br/>the
[Debezium's SqlServer Connector
properties](https://github.com/debezium/debezium/blob/1.6/documentation/modules/ROOT/pages/connectors/sqlserver.adoc#connector-properties)
[...]
| format | Enum | No | DEFAULT |
Optional output format for SqlServer CDC, valid enumerations are
"DEFAULT"、"COMPATIBLE_DEBEZIUM_JSON".
[...]
+| schema-changes.enabled | Boolean | No | false |
Schema evolution is disabled by default. Now we only support `add column`、`drop
column`、`rename column` and `modify column`.
[...]
+| schema-changes.include | List | No | - |
Only the listed schema change event types are sent downstream (when
`schema-changes.enabled = true`). Empty means all are eligible. See [Schema
change event filtering](#schema-change-event-filtering).
[...]
+| schema-changes.exclude | List | No | - |
Schema change event types listed here are NOT sent downstream. Applied after
`schema-changes.include`; exclude wins on conflict. See [Schema change event
filtering](#schema-change-event-filtering).
[...]
| common-options | | no | - |
Source plugin common parameters, please refer to [Source Common
Options](../common-options/source-common-options.md) for details.
[...]
### Enable Sql Server CDC
@@ -241,6 +244,42 @@ sink {
}
```
+### Schema change event filtering
+
+When `schema-changes.enabled = true`, you can further control which schema
change event types are
+propagated downstream using `schema-changes.include` /
`schema-changes.exclude`.
+
+Use these SeaTunnel-owned canonical names:
+
+| Canonical name | Operation |
+|------------------|------------------------------------------------------|
+| `add.column` | add a column |
+| `drop.column` | drop a column |
+| `modify.column` | change a column's type/attributes, name unchanged |
+| `change.column` | rename a column, optionally re-type |
+| `update.columns` | group alias for all four column-level changes above |
+
+Precedence is deterministic:
+
+1. if `schema-changes.include` is set, only included event types are eligible;
+2. `schema-changes.exclude` is then applied;
+3. **exclude wins** when a type appears in both lists.
+
+```hocon
+source {
+ SqlServer-CDC {
+ # ...
+ schema-changes.enabled = true
+ schema-changes.include = ["add.column", "drop.column"]
+ schema-changes.exclude = ["change.column"]
+ }
+}
+```
+
+**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.
+
## Changelog
<ChangeLog />
diff --git a/docs/zh/connectors/source/MySQL-CDC.md
b/docs/zh/connectors/source/MySQL-CDC.md
index c08e6596ce..b8c27ecc0f 100644
--- a/docs/zh/connectors/source/MySQL-CDC.md
+++ b/docs/zh/connectors/source/MySQL-CDC.md
@@ -214,6 +214,8 @@ show variables where variable_name in ('log_bin',
'binlog_format', 'binlog_row_i
| exactly_once | Boolean | 否 | false |
启用精确一次语义.
|
| format | Enum | 否 | DEFAULT |
MySQL CDC 的可选输出格式, 有效的枚举值为 `DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`.
|
| schema-changes.enabled | Boolean | 否 | false |
模式演进默认是禁用的. 当前我们只支持 `add column`、`drop column`、`rename column` 和 `modify
column`.
|
+| schema-changes.include | List | 否 | - |
仅向下游发送列出的 schema change 事件类型（需 `schema-changes.enabled = true`）。为空表示全部允许。详见
[Schema change 事件过滤](#schema-change-事件过滤)。
|
+| schema-changes.exclude | List | 否 | - |
此处列出的 schema change 事件类型不会发送到下游。在 `schema-changes.include` 之后应用；冲突时 exclude
优先。详见 [Schema change 事件过滤](#schema-change-事件过滤)。
|
| debezium | Config | 否 | - | 传递
[Debezium的属性](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/mysql.adoc#connector-properties)
给Debezium嵌入式引擎, 该引擎用于捕获 MySQL 服务的数据变更.
|
| int_type_narrowing | Boolean | 否 | true |
Int类型收窄，如果为 true，则 tinyint(1) 类型将被收窄为 boolean 类型（如果没有精度损失）。目前仅支持 MySQL。
|
| common-options | | 否 | - |
Source插件通用参数, 详见 [Source Common
Options](../common-options/source-common-options.md)
|
@@ -339,6 +341,42 @@ sink {
}
```
+
+### Schema change 事件过滤
+
+当 `schema-changes.enabled = true` 时，可通过 `schema-changes.include` /
`schema-changes.exclude` 进一步
+控制哪些 schema change 事件类型会被发送到下游。
+
+使用以下 SeaTunnel 统一的规范名称
+
+| 规范名称 | 操作 |
+|-----------------|-------------------------------------------------------------|
+| `add.column` | 新增列 |
+| `drop.column` | 删除列 |
+| `modify.column` | 修改列的类型/属性，列名不变 |
+| `change.column` | 列重命名，可同时改类型 |
+| `update.columns` | 上述四种列级变更的分组别名 |
+
+优先级规则（确定性）：
+
+1. 若设置了 `schema-changes.include`，则只有被包含的事件类型才有资格；
+2. 然后应用 `schema-changes.exclude`；
+3. 当某类型同时出现在两个列表中时，**exclude 优先**。
+
+```hocon
+source {
+ MySQL-CDC {
+ # ...
+ schema-changes.enabled = true
+ schema-changes.include = ["add.column", "drop.column"]
+ schema-changes.exclude = ["change.column"]
+ }
+}
+```
+
+**排除 `drop.column` 时的数据处理方式。对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
+NOT NULL 列排除 `drop.column` 会在 sink 端失败。
+
### 表名支持正则以读取多个表
> `table-pattern` 和 `table-names` 只能选择一个
diff --git a/docs/zh/connectors/source/Oracle-CDC.md
b/docs/zh/connectors/source/Oracle-CDC.md
index d542a5057f..534144ce30 100644
--- a/docs/zh/connectors/source/Oracle-CDC.md
+++ b/docs/zh/connectors/source/Oracle-CDC.md
@@ -253,6 +253,8 @@ exit;
| skip_analyze | Boolean | 否 | false |
在全量阶段跳过表行数的分析。在这种情况下，您需要定期调度分析表 SQL 以更新相关表统计信息，或者您的表数据更改不频繁。
[...]
| format | Enum | 否 | DEFAULT |
Oracle CDC 的可选输出格式，有效枚举值为 `DEFAULT`、`COMPATIBLE_DEBEZIUM_JSON`。
[...]
| schema-changes.enabled | Boolean | 否 | false |
Schema 演进默认禁用。目前我们仅支持 `add column`、`drop column`、`rename column` 和 `modify
column`。
[...]
+| schema-changes.include | List | 否 | - |
仅向下游发送列出的 schema change 事件类型（需 `schema-changes.enabled = true`）。为空表示全部允许。详见
[Schema change 事件过滤](#schema-change-事件过滤)。
[...]
+| schema-changes.exclude | List | 否 | - |
此处列出的 schema change 事件类型不会发送到下游。在 `schema-changes.include` 之后应用；冲突时 exclude
优先。详见 [Schema change 事件过滤](#schema-change-事件过滤)。
[...]
| debezium | Config | 否 | - | 透传
[Debezium
属性](https://github.com/debezium/debezium/blob/v1.9.8.Final/documentation/modules/ROOT/pages/connectors/oracle.adoc#connector-properties)
给 Debezium Embedded Engine，该引擎用于捕获 Oracle 服务器的数据更改。
[...]
| common-options | | 否 | - |
源端插件常用参数，详情请参阅 [源端常用选项](../common-options/source-common-options.md)。
[...]
| decimal_type_narrowing | Boolean | 否 | true
| 数值类型收缩，如果为 true，则在不损失精度的情况下，将 decimal 类型收缩为 int 或 long 类型。目前仅支持
Oracle。请参阅下文的 `decimal_type_narrowing`。
[...]
@@ -362,6 +364,41 @@ source {
}
```
+### Schema change 事件过滤
+
+当 `schema-changes.enabled = true` 时，可通过 `schema-changes.include` /
`schema-changes.exclude` 进一步
+控制哪些 schema change 事件类型会被发送到下游。过滤只影响“发往下游”的部分。
+
+使用以下 SeaTunnel 统一的规范名称：
+
+| 规范名称 | 操作 |
+|------------------|---------------------------------------------|
+| `add.column` | 新增列 |
+| `drop.column` | 删除列 |
+| `modify.column` | 修改列的类型/属性，列名不变 |
+| `change.column` | 列重命名，可同时改类型 |
+| `update.columns` | 上述四种列级变更的分组别名 |
+
+优先级规则（确定性）：
+
+1. 若设置了 `schema-changes.include`，则只有被包含的事件类型才有资格；
+2. 然后应用 `schema-changes.exclude`；
+3. 当某类型同时出现在两个列表中时，**exclude 优先**。
+
+```hocon
+source {
+ Oracle-CDC {
+ # ...
+ schema-changes.enabled = true
+ schema-changes.include = ["add.column", "drop.column"]
+ schema-changes.exclude = ["change.column"]
+ }
+}
+```
+
+**排除 `drop.column` 时的数据处理方式。对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
+NOT NULL 列排除 `drop.column` 会在 sink 端失败。
+
### 支持以兼容 debezium 的格式发送到 kafka
> 必须与 kafka 连接器 sink 配合使用，详情请参阅 [兼容 debezium
> 格式](../formats/cdc-compatible-debezium-json.md)
diff --git a/docs/zh/connectors/source/SqlServer-CDC.md
b/docs/zh/connectors/source/SqlServer-CDC.md
index ee849ea7f4..c7453e8d14 100644
--- a/docs/zh/connectors/source/SqlServer-CDC.md
+++ b/docs/zh/connectors/source/SqlServer-CDC.md
@@ -102,6 +102,9 @@ Sql Server CDC 连接器允许从 SqlServer 数据库读取快照数据和增量
| exactly_once | Boolean | 否 | false
| 启用精确一次语义。
|
| debezium.* | config | 否 | -
| 将 Debezium 的属性传递给 Debezium Embedded Engine，用于捕获来自 SqlServer
服务器的数据变更。<br/>了解更多关于<br/>[Debezium 的 SqlServer
连接器属性](https://github.com/debezium/debezium/blob/1.6/documentation/modules/ROOT/pages/connectors/sqlserver.adoc#connector-properties)
|
| format | Enum | 否 |
DEFAULT | SqlServer CDC 的可选输出格式，有效枚举为 "DEFAULT"、"COMPATIBLE_DEBEZIUM_JSON"。
|
+| schema-changes.enabled | Boolean | 否 | false
| 模式演进默认是禁用的。当前我们只支持 `add column`、`drop column`、`rename column` 和 `modify
column`。
|
+| schema-changes.include | List | 否 | -
| 仅向下游发送列出的 schema change 事件类型（需 `schema-changes.enabled = true`）。为空表示全部允许。详见
[Schema change 事件过滤](#schema-change-事件过滤)。
|
+| schema-changes.exclude | List | 否 | -
| 此处列出的 schema change 事件类型不会发送到下游。在 `schema-changes.include` 之后应用；冲突时 exclude
优先。详见 [Schema change 事件过滤](#schema-change-事件过滤)。
|
| common-options | | 否 | -
| 源插件通用参数，请参考 [源通用选项](../common-options/source-common-options.md) 获取详细信息。
|
### 启用 Sql Server CDC
@@ -239,6 +242,41 @@ sink {
}
```
+### Schema change 事件过滤
+
+当 `schema-changes.enabled = true` 时，可通过 `schema-changes.include` /
`schema-changes.exclude` 进一步
+控制哪些 schema change 事件类型会被发送到下游。
+
+使用以下 SeaTunnel 统一的规范名称：
+
+| 规范名称 | 操作 |
+|------------------|---------------------------------------------|
+| `add.column` | 新增列 |
+| `drop.column` | 删除列 |
+| `modify.column` | 修改列的类型/属性，列名不变 |
+| `change.column` | 列重命名，可同时改类型 |
+| `update.columns` | 上述四种列级变更的分组别名 |
+
+优先级规则（确定性）：
+
+1. 若设置了 `schema-changes.include`，则只有被包含的事件类型才有资格；
+2. 然后应用 `schema-changes.exclude`；
+3. 当某类型同时出现在两个列表中时，**exclude 优先**。
+
+```hocon
+source {
+ SqlServer-CDC {
+ # ...
+ schema-changes.enabled = true
+ schema-changes.include = ["add.column", "drop.column"]
+ schema-changes.exclude = ["change.column"]
+ }
+}
+```
+
+**排除 `drop.column` 时的数据处理方式。对于被保留的 **NOT NULL** 列，写入 `NULL` 会被 sink
拒绝，因此对一个源端已不再供数的
+NOT NULL 列排除 `drop.column` 会在 sink 端失败。
+
## 变更日志
<ChangeLog />
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/SourceOptions.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/SourceOptions.java
index 013d798642..97b229aedf 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/SourceOptions.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/option/SourceOptions.java
@@ -20,8 +20,11 @@ package org.apache.seatunnel.connectors.cdc.base.option;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.Options;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeEventType;
import org.apache.seatunnel.connectors.cdc.debezium.DeserializeFormat;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
@SuppressWarnings("MagicNumber")
@@ -121,6 +124,28 @@ public class SourceOptions {
.withDescription(
"Enable send schema change events, by default is
false. If set to true, the schema changes will be sent to downstream.");
+ public static final Option<List<String>> SCHEMA_CHANGES_INCLUDE =
+ Options.key("schema-changes.include")
+ .listType()
+ .defaultValue(Collections.emptyList())
+ .withDescription(
+ "Only schema change event types listed here are
sent downstream when schema-changes.enabled is true. "
+ + "Empty means all event types are
eligible. "
+ + "Valid values: "
+ + SchemaChangeEventType.validNames()
+ + " (update.columns is a group alias for
all column-level changes). ");
+
+ public static final Option<List<String>> SCHEMA_CHANGES_EXCLUDE =
+ Options.key("schema-changes.exclude")
+ .listType()
+ .defaultValue(Collections.emptyList())
+ .withDescription(
+ "Schema change event types listed here are NOT
sent downstream. Applied after schema-changes.include; "
+ + "exclude wins when a type appears in
both lists. "
+ + "Valid values: "
+ + SchemaChangeEventType.validNames()
+ + " (update.columns is a group alias for
all column-level changes). ");
+
public static OptionRule.Builder getBaseRule() {
return OptionRule.builder()
.optional(FORMAT)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventFilter.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventFilter.java
new file mode 100644
index 0000000000..3ebfb4b551
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventFilter.java
@@ -0,0 +1,174 @@
+/*
+ * 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.cdc.base.schema;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.event.EventType;
+import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
+
+import java.io.Serializable;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Include/exclude filter for CDC schema change event types.
+ *
+ * <p>Applied after schema change events are normalized to the SeaTunnel event
model, and before
+ * they are emitted to downstream schema-change coordination. A filtered-out
event is neither
+ * forwarded to downstream coordination nor applied to the produced schema, so
the produced row
+ * shape stays in lockstep with the (filtered) sink schema — this is what
keeps a column whose
+ * {@code drop.column} was suppressed by {@code exclude} present on both sides.
+ *
+ * <p>Precedence (deterministic):
+ *
+ * <ol>
+ * <li>if {@code include} is non-empty, only included event types are
eligible;
+ * <li>{@code exclude} is then applied;
+ * <li>{@code exclude} wins when the same type appears in both lists.
+ * </ol>
+ *
+ * <p>{@code update.columns} acts as a group alias for all column-level
changes ({@code add.column},
+ * {@code drop.column}, {@code modify.column}, {@code change.column}):
including it admits the whole
+ * group, excluding it suppresses the whole group.
+ */
+public final class SchemaChangeEventFilter implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Set<EventType> includeTypes;
+ private final Set<EventType> excludeTypes;
+
+ public SchemaChangeEventFilter(Set<EventType> includeTypes, Set<EventType>
excludeTypes) {
+ this.includeTypes = new HashSet<>(includeTypes);
+ this.excludeTypes = new HashSet<>(excludeTypes);
+ }
+
+ /** Builds a filter from the {@code schema-changes.include} / {@code
exclude} options. */
+ public static SchemaChangeEventFilter fromConfig(ReadonlyConfig config) {
+ List<String> include =
config.get(SourceOptions.SCHEMA_CHANGES_INCLUDE);
+ List<String> exclude =
config.get(SourceOptions.SCHEMA_CHANGES_EXCLUDE);
+ return new SchemaChangeEventFilter(
+ SchemaChangeEventType.fromCanonicalNames(include),
+ SchemaChangeEventType.fromCanonicalNames(exclude));
+ }
+
+ /**
+ * Validates the {@code schema-changes.include} / {@code
schema-changes.exclude} option values.
+ *
+ * <p>Invoked at job submission time (from the source factory) so an
unknown canonical name —
+ * e.g. a typo such as {@code rename.tabble} — fails fast during
submission with a message
+ * listing the valid names, instead of bypassing submission-time option
validation and failing
+ * later during source initialization.
+ */
+ public static void validateOptions(ReadonlyConfig config) {
+ validateNames(
+ SourceOptions.SCHEMA_CHANGES_INCLUDE.key(),
+ config.get(SourceOptions.SCHEMA_CHANGES_INCLUDE));
+ validateNames(
+ SourceOptions.SCHEMA_CHANGES_EXCLUDE.key(),
+ config.get(SourceOptions.SCHEMA_CHANGES_EXCLUDE));
+ }
+
+ private static void validateNames(String optionKey, List<String> names) {
+ try {
+ SchemaChangeEventType.fromCanonicalNames(names);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Invalid value for option '" + optionKey + "'. " +
e.getMessage(), e);
+ }
+ }
+
+ public boolean isNoOp() {
+ return includeTypes.isEmpty() && excludeTypes.isEmpty();
+ }
+
+ /**
+ * Applies the filter to a normalized schema change event.
+ *
+ * @return the original event when fully eligible, a reduced {@link
AlterTableColumnsEvent} when
+ * only some of its column sub-events are eligible, or {@code null}
when the whole event is
+ * filtered out.
+ */
+ public SchemaChangeEvent filter(SchemaChangeEvent event) {
+ if (event == null || isNoOp()) {
+ return event;
+ }
+
+ if (event instanceof AlterTableColumnsEvent) {
+ AlterTableColumnsEvent composite = (AlterTableColumnsEvent) event;
+ List<AlterTableColumnEvent> survivors =
+ composite.getEvents().stream()
+ .filter(sub ->
isColumnLevelEligible(sub.getEventType()))
+ .collect(Collectors.toList());
+ if (survivors.isEmpty()) {
+ return null;
+ }
+ if (survivors.size() == composite.getEvents().size()) {
+ return composite;
+ }
+ return rebuildComposite(composite, survivors);
+ }
+
+ if (event instanceof AlterTableColumnEvent) {
+ // A standalone column-level event (not wrapped in a composite).
+ return isColumnLevelEligible(event.getEventType()) ? event : null;
+ }
+
+ // Table-level events. No canonical name currently maps to a
table-level type (rename.table
+ // is not exposed yet), so this is defensive: such events are not
produced today, but if one
+ // arrives it is filtered consistently with the include/exclude
precedence.
+ return isEligible(event.getEventType()) ? event : null;
+ }
+
+ /** Eligibility for table-level event types (no group alias). */
+ private boolean isEligible(EventType type) {
+ boolean included = includeTypes.isEmpty() ||
includeTypes.contains(type);
+ return included && !excludeTypes.contains(type);
+ }
+
+ /**
+ * Eligibility for column-level event types, honoring {@code
update.columns} as the group alias
+ * for the whole column-change family.
+ */
+ private boolean isColumnLevelEligible(EventType type) {
+ boolean included =
+ includeTypes.isEmpty()
+ || includeTypes.contains(type)
+ ||
includeTypes.contains(EventType.SCHEMA_CHANGE_UPDATE_COLUMNS);
+ boolean excluded =
+ excludeTypes.contains(type)
+ ||
excludeTypes.contains(EventType.SCHEMA_CHANGE_UPDATE_COLUMNS);
+ return included && !excluded;
+ }
+
+ private static AlterTableColumnsEvent rebuildComposite(
+ AlterTableColumnsEvent original, List<AlterTableColumnEvent>
survivors) {
+ AlterTableColumnsEvent reduced =
+ new AlterTableColumnsEvent(original.tableIdentifier(),
survivors);
+ reduced.setStatement(original.getStatement());
+ reduced.setSourceDialectName(original.getSourceDialectName());
+ reduced.setChangeAfter(original.getChangeAfter());
+ reduced.setJobId(original.getJobId());
+ return reduced;
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventType.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventType.java
new file mode 100644
index 0000000000..dec081d9bc
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventType.java
@@ -0,0 +1,95 @@
+/*
+ * 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.cdc.base.schema;
+
+import org.apache.seatunnel.api.event.EventType;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Maps the user-facing canonical names used in {@code schema-changes.include}
/ {@code
+ * schema-changes.exclude} to the internal {@link EventType} enum, so users
never need to know
+ * internal class or enum names. {@code update.columns} is a group alias for
all column-level
+ * changes.
+ */
+public final class SchemaChangeEventType {
+
+ private static final Map<String, EventType> CANONICAL_NAME_TO_EVENT_TYPE;
+
+ static {
+ Map<String, EventType> map = new LinkedHashMap<>();
+ map.put("add.column", EventType.SCHEMA_CHANGE_ADD_COLUMN);
+ map.put("drop.column", EventType.SCHEMA_CHANGE_DROP_COLUMN);
+ map.put("modify.column", EventType.SCHEMA_CHANGE_MODIFY_COLUMN);
+ map.put("change.column", EventType.SCHEMA_CHANGE_CHANGE_COLUMN);
+ map.put("update.columns", EventType.SCHEMA_CHANGE_UPDATE_COLUMNS);
+ // NOTE: rename.table (SCHEMA_CHANGE_RENAME_TABLE) is intentionally
NOT exposed yet. CDC has
+ // no end-to-end handling for table renames: the DDL is never parsed
into an
+ // AlterTableNameEvent, the schema handlers treat that event as a
no-op, and no sink applies
+ // it. Exposing it as a filterable name would advertise a capability
that does not exist.
+ // It should be added back only once table-rename is implemented
end-to-end (see the
+ // rename-table design follow-up).
+ CANONICAL_NAME_TO_EVENT_TYPE = Collections.unmodifiableMap(map);
+ }
+
+ private SchemaChangeEventType() {}
+
+ public static String validNames() {
+ return String.join(", ", CANONICAL_NAME_TO_EVENT_TYPE.keySet());
+ }
+
+ public static EventType fromCanonicalName(String canonicalName) {
+ if (canonicalName == null) {
+ throw new IllegalArgumentException(
+ "Schema change event type name must not be null. Valid
names are: "
+ + validNames());
+ }
+ String normalized = canonicalName.trim().toLowerCase();
+ EventType eventType = CANONICAL_NAME_TO_EVENT_TYPE.get(normalized);
+ if (eventType == null) {
+ throw new IllegalArgumentException(
+ "Unknown schema change event type '"
+ + canonicalName
+ + "'. Valid names are: "
+ + validNames());
+ }
+ return eventType;
+ }
+
+ public static Set<EventType> fromCanonicalNames(Collection<String>
canonicalNames) {
+ if (canonicalNames == null || canonicalNames.isEmpty()) {
+ return Collections.emptySet();
+ }
+ return canonicalNames.stream()
+ .map(SchemaChangeEventType::fromCanonicalName)
+ .collect(Collectors.toCollection(LinkedHashSet::new));
+ }
+
+ /** Visible for testing: the supported canonical names. */
+ static List<String> canonicalNames() {
+ return new ArrayList<>(CANONICAL_NAME_TO_EVENT_TYPE.keySet());
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/BaseChangeStreamTableSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/BaseChangeStreamTableSourceFactory.java
index a801a626e1..bc2c4961cd 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/BaseChangeStreamTableSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/BaseChangeStreamTableSourceFactory.java
@@ -25,6 +25,7 @@ import org.apache.seatunnel.api.table.connector.TableSource;
import org.apache.seatunnel.api.table.factory.ChangeStreamTableSourceFactory;
import org.apache.seatunnel.api.table.factory.ChangeStreamTableSourceState;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
+import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeEventFilter;
import org.apache.seatunnel.connectors.cdc.base.source.split.IncrementalSplit;
import org.apache.seatunnel.connectors.cdc.base.source.split.SourceSplitBase;
@@ -47,6 +48,7 @@ public abstract class BaseChangeStreamTableSourceFactory
implements ChangeStream
@Override
public <T, SplitT extends SourceSplit, StateT extends Serializable>
TableSource<T, SplitT, StateT>
createSource(TableSourceFactoryContext context) {
+ SchemaChangeEventFilter.validateOptions(context.getOptions());
return restoreSource(context, Collections.emptyList());
}
@@ -55,6 +57,7 @@ public abstract class BaseChangeStreamTableSourceFactory
implements ChangeStream
TableSource<T, SplitT, StateT> restoreSource(
TableSourceFactoryContext context,
ChangeStreamTableSourceState<StateT, SplitT> state) {
+ SchemaChangeEventFilter.validateOptions(context.getOptions());
return restoreSource(context, getRestoreTableStruct(state));
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
index 4dcfb77078..4c237b1bcd 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializeSchema.java
@@ -30,6 +30,7 @@ import
org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventHandl
import org.apache.seatunnel.api.table.type.MetadataUtil;
import org.apache.seatunnel.api.table.type.RowKind;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeEventFilter;
import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeResolver;
import org.apache.seatunnel.connectors.cdc.base.utils.SourceRecordUtils;
import
org.apache.seatunnel.connectors.cdc.debezium.AbstractDebeziumDeserializationSchema;
@@ -72,6 +73,7 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
private final ZoneId serverTimeZone;
private final DebeziumDeserializationConverterFactory
userDefinedConverterFactory;
private final SchemaChangeResolver schemaChangeResolver;
+ private final SchemaChangeEventFilter schemaChangeEventFilter;
private final TableSchemaChangeEventHandler tableSchemaChangeHandler;
private List<CatalogTable> tables;
private Map<String, SeaTunnelRowDebeziumDeserializationConverters>
tableRowConverters;
@@ -82,6 +84,7 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
ZoneId serverTimeZone,
DebeziumDeserializationConverterFactory
userDefinedConverterFactory,
SchemaChangeResolver schemaChangeResolver,
+ SchemaChangeEventFilter schemaChangeEventFilter,
Map<TableId, Struct> tableIdTableChangeMap) {
super(tableIdTableChangeMap);
this.metadataConverters = metadataConverters;
@@ -89,6 +92,7 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
this.userDefinedConverterFactory = userDefinedConverterFactory;
this.tables = checkNotNull(tables);
this.schemaChangeResolver = schemaChangeResolver;
+ this.schemaChangeEventFilter = schemaChangeEventFilter;
this.tableSchemaChangeHandler = new TableSchemaChangeEventDispatcher();
this.tableRowConverters =
createTableRowConverters(
@@ -136,6 +140,18 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
log.warn("Unsupported resolve schemaChangeEvent {}, just skip.",
record);
return;
}
+
+ // Filter before updating the produced schema, so the produced row
shape stays in lockstep
+ // with the (filtered) sink schema. Only surviving events are applied
below.
+ if (schemaChangeEventFilter != null) {
+ schemaChangeEvent =
schemaChangeEventFilter.filter(schemaChangeEvent);
+ }
+ if (schemaChangeEvent == null) {
+ log.debug(
+ "Schema change event is fully filtered out by
schema-changes.include/exclude, not applied to schema and not sent
downstream.");
+ return;
+ }
+
boolean tableExist = false;
for (int i = 0; i < tables.size(); i++) {
CatalogTable changeBefore = tables.get(i);
@@ -394,6 +410,7 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
DebeziumDeserializationConverterFactory.DEFAULT;
private Map<TableId, Struct> tableIdTableChangeMap = new HashMap<>();
private SchemaChangeResolver schemaChangeResolver;
+ private SchemaChangeEventFilter schemaChangeEventFilter;
public SeaTunnelRowDebeziumDeserializeSchema build() {
return new SeaTunnelRowDebeziumDeserializeSchema(
@@ -402,6 +419,7 @@ public final class SeaTunnelRowDebeziumDeserializeSchema
serverTimeZone,
userDefinedConverterFactory,
schemaChangeResolver,
+ schemaChangeEventFilter,
tableIdTableChangeMap);
}
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventFilterTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventFilterTest.java
new file mode 100644
index 0000000000..edd42553f0
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/base/schema/SchemaChangeEventFilterTest.java
@@ -0,0 +1,308 @@
+/*
+ * 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.cdc.base.schema;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.event.EventType;
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableModifyColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableNameEvent;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.BasicType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+class SchemaChangeEventFilterTest {
+
+ private static final TableIdentifier TABLE = TableIdentifier.of("", "db",
"tbl");
+
+ private static Column column(String name) {
+ return PhysicalColumn.of(name, BasicType.STRING_TYPE, 10L, true, null,
"");
+ }
+
+ private static SchemaChangeEventFilter filter(List<String> include,
List<String> exclude) {
+ Map<String, Object> map = new HashMap<>();
+ map.put("schema-changes.include", include);
+ map.put("schema-changes.exclude", exclude);
+ return SchemaChangeEventFilter.fromConfig(ReadonlyConfig.fromMap(map));
+ }
+
+ private static AlterTableColumnsEvent composite(AlterTableColumnEvent...
events) {
+ return new AlterTableColumnsEvent(TABLE, new
ArrayList<>(Arrays.asList(events)));
+ }
+
+ private static List<EventType> subTypes(SchemaChangeEvent event) {
+ return ((AlterTableColumnsEvent) event)
+ .getEvents().stream()
+ .map(AlterTableColumnEvent::getEventType)
+ .collect(Collectors.toList());
+ }
+
+ @Test
+ void noConfigIsAllowAll() {
+ SchemaChangeEventFilter f = filter(Collections.emptyList(),
Collections.emptyList());
+ Assertions.assertTrue(f.isNoOp());
+ AlterTableColumnsEvent event =
+ composite(
+ AlterTableAddColumnEvent.add(TABLE, column("a")),
+ new AlterTableDropColumnEvent(TABLE, "b"));
+ Assertions.assertSame(event, f.filter(event));
+ }
+
+ @Test
+ void includeOnlyKeepsOnlyListedLeafTypes() {
+ SchemaChangeEventFilter f = filter(Arrays.asList("add.column"),
Collections.emptyList());
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableAddColumnEvent.add(TABLE,
column("a")),
+ new AlterTableDropColumnEvent(TABLE, "b"),
+ AlterTableModifyColumnEvent.modify(TABLE,
column("c"))));
+ Assertions.assertEquals(
+ Collections.singletonList(EventType.SCHEMA_CHANGE_ADD_COLUMN),
subTypes(result));
+ }
+
+ @Test
+ void excludeOnlyDropsListedLeafTypes() {
+ SchemaChangeEventFilter f = filter(Collections.emptyList(),
Arrays.asList("drop.column"));
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableAddColumnEvent.add(TABLE,
column("a")),
+ new AlterTableDropColumnEvent(TABLE, "b")));
+ Assertions.assertEquals(
+ Collections.singletonList(EventType.SCHEMA_CHANGE_ADD_COLUMN),
subTypes(result));
+ }
+
+ @Test
+ void excludeWinsOverIncludeForSameType() {
+ SchemaChangeEventFilter f =
+ filter(Arrays.asList("add.column", "drop.column"),
Arrays.asList("add.column"));
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableAddColumnEvent.add(TABLE,
column("a")),
+ new AlterTableDropColumnEvent(TABLE, "b")));
+ Assertions.assertEquals(
+
Collections.singletonList(EventType.SCHEMA_CHANGE_DROP_COLUMN),
subTypes(result));
+ }
+
+ @Test
+ void wholeEventDroppedWhenNoSubEventSurvives() {
+ SchemaChangeEventFilter f =
+ filter(Collections.emptyList(), Arrays.asList("add.column",
"drop.column"));
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableAddColumnEvent.add(TABLE,
column("a")),
+ new AlterTableDropColumnEvent(TABLE, "b")));
+ Assertions.assertNull(result);
+ }
+
+ @Test
+ void changeColumnIsFilteredIndependentlyOfModifyColumn() {
+ // exclude modify.column must NOT drop a change.column (rename)
sub-event
+ SchemaChangeEventFilter f = filter(Collections.emptyList(),
Arrays.asList("modify.column"));
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableModifyColumnEvent.modify(TABLE,
column("c")),
+ AlterTableChangeColumnEvent.change(TABLE,
"old", column("new"))));
+ Assertions.assertEquals(
+
Collections.singletonList(EventType.SCHEMA_CHANGE_CHANGE_COLUMN),
subTypes(result));
+ }
+
+ @Test
+ void updateColumnsIncludeIsGroupAliasForAllColumnChanges() {
+ SchemaChangeEventFilter f =
+ filter(Arrays.asList("update.columns"),
Collections.emptyList());
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableAddColumnEvent.add(TABLE,
column("a")),
+ new AlterTableDropColumnEvent(TABLE, "b"),
+ AlterTableModifyColumnEvent.modify(TABLE,
column("c"))));
+ Assertions.assertEquals(3, ((AlterTableColumnsEvent)
result).getEvents().size());
+ }
+
+ @Test
+ void updateColumnsExcludeSuppressesAllColumnChanges() {
+ SchemaChangeEventFilter f =
+ filter(Collections.emptyList(),
Arrays.asList("update.columns"));
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableAddColumnEvent.add(TABLE,
column("a")),
+ AlterTableModifyColumnEvent.modify(TABLE,
column("c"))));
+ Assertions.assertNull(result);
+ }
+
+ @Test
+ void renameTableIsNotAnExposedCanonicalName() {
+ // rename.table is intentionally not exposed: CDC has no end-to-end
handling for table
+ // renames, so it must be rejected as an unknown name rather than
advertised as filterable.
+ IllegalArgumentException ex =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
SchemaChangeEventType.fromCanonicalName("rename.table"));
+ Assertions.assertTrue(ex.getMessage().contains("rename.table"));
+ }
+
+ @Test
+ void strayTableLevelEventIsHandledDefensively() {
+ // No canonical name maps to a table-level type, so such an event is
not produced by CDC
+ // today. The filter still handles it defensively: an active
column-level include list
+ // drops it, and a no-op filter passes it through unchanged.
+ AlterTableNameEvent rename =
+ new AlterTableNameEvent(TABLE, TableIdentifier.of("", "db",
"tbl2"));
+ Assertions.assertNull(
+ filter(Arrays.asList("add.column"),
Collections.emptyList()).filter(rename));
+ Assertions.assertSame(
+ rename, filter(Collections.emptyList(),
Collections.emptyList()).filter(rename));
+ }
+
+ @Test
+ void unknownNameFailsFast() {
+ IllegalArgumentException ex =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> filter(Arrays.asList("create.table"),
Collections.emptyList()));
+ Assertions.assertTrue(ex.getMessage().contains("create.table"));
+ Assertions.assertTrue(ex.getMessage().contains("add.column"));
+ }
+
+ @Test
+ void validateOptionsAcceptsValidNames() {
+ Map<String, Object> map = new HashMap<>();
+ map.put("schema-changes.include", Arrays.asList("add.column"));
+ map.put("schema-changes.exclude", Arrays.asList("drop.column"));
+ Assertions.assertDoesNotThrow(
+ () ->
SchemaChangeEventFilter.validateOptions(ReadonlyConfig.fromMap(map)));
+ }
+
+ @Test
+ void validateOptionsAcceptsEmptyConfig() {
+ Assertions.assertDoesNotThrow(
+ () ->
+ SchemaChangeEventFilter.validateOptions(
+ ReadonlyConfig.fromMap(new HashMap<>())));
+ }
+
+ @Test
+ void validateOptionsFailsFastOnUnknownIncludeName() {
+ Map<String, Object> map = new HashMap<>();
+ map.put("schema-changes.include", Arrays.asList("rename.tabble"));
+ IllegalArgumentException ex =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
SchemaChangeEventFilter.validateOptions(ReadonlyConfig.fromMap(map)));
+
Assertions.assertTrue(ex.getMessage().contains("schema-changes.include"));
+ Assertions.assertTrue(ex.getMessage().contains("rename.tabble"));
+ Assertions.assertTrue(ex.getMessage().contains("add.column"));
+ }
+
+ @Test
+ void validateOptionsFailsFastOnUnknownExcludeName() {
+ Map<String, Object> map = new HashMap<>();
+ map.put("schema-changes.exclude", Arrays.asList("drop.colum"));
+ IllegalArgumentException ex =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
SchemaChangeEventFilter.validateOptions(ReadonlyConfig.fromMap(map)));
+
Assertions.assertTrue(ex.getMessage().contains("schema-changes.exclude"));
+ Assertions.assertTrue(ex.getMessage().contains("drop.colum"));
+ }
+
+ @Test
+ void namesAreNormalizedAndDeduplicated() {
+ SchemaChangeEventFilter f =
+ filter(Arrays.asList(" ADD.COLUMN ", "add.column"),
Collections.emptyList());
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableAddColumnEvent.add(TABLE,
column("a")),
+ new AlterTableDropColumnEvent(TABLE, "b")));
+ Assertions.assertEquals(
+ Collections.singletonList(EventType.SCHEMA_CHANGE_ADD_COLUMN),
subTypes(result));
+ }
+
+ @Test
+ void canonicalNameMappingIsExhaustiveAndStable() {
+ Assertions.assertEquals(
+ EventType.SCHEMA_CHANGE_ADD_COLUMN,
+ SchemaChangeEventType.fromCanonicalName("add.column"));
+ Assertions.assertEquals(
+ EventType.SCHEMA_CHANGE_DROP_COLUMN,
+ SchemaChangeEventType.fromCanonicalName("drop.column"));
+ Assertions.assertEquals(
+ EventType.SCHEMA_CHANGE_MODIFY_COLUMN,
+ SchemaChangeEventType.fromCanonicalName("modify.column"));
+ Assertions.assertEquals(
+ EventType.SCHEMA_CHANGE_CHANGE_COLUMN,
+ SchemaChangeEventType.fromCanonicalName("change.column"));
+ Assertions.assertEquals(
+ EventType.SCHEMA_CHANGE_UPDATE_COLUMNS,
+ SchemaChangeEventType.fromCanonicalName("update.columns"));
+ Assertions.assertEquals(5,
SchemaChangeEventType.canonicalNames().size());
+ }
+
+ /**
+ * Per-type coverage: for each column-level canonical name, {@code
include=[name]} must keep
+ * exactly that sub-event out of a composite carrying all four
column-level changes.
+ */
+ @Test
+ void includeEachColumnLevelTypeKeepsOnlyThatType() {
+ Map<String, EventType> cases = new HashMap<>();
+ cases.put("add.column", EventType.SCHEMA_CHANGE_ADD_COLUMN);
+ cases.put("drop.column", EventType.SCHEMA_CHANGE_DROP_COLUMN);
+ cases.put("modify.column", EventType.SCHEMA_CHANGE_MODIFY_COLUMN);
+ cases.put("change.column", EventType.SCHEMA_CHANGE_CHANGE_COLUMN);
+
+ for (Map.Entry<String, EventType> c : cases.entrySet()) {
+ SchemaChangeEventFilter f =
+ filter(Collections.singletonList(c.getKey()),
Collections.emptyList());
+ SchemaChangeEvent result =
+ f.filter(
+ composite(
+ AlterTableAddColumnEvent.add(TABLE,
column("a")),
+ new AlterTableDropColumnEvent(TABLE, "b"),
+ AlterTableModifyColumnEvent.modify(TABLE,
column("c")),
+ AlterTableChangeColumnEvent.change(TABLE,
"old", column("d"))));
+ Assertions.assertEquals(
+ Collections.singletonList(c.getValue()),
+ subTypes(result),
+ "include=[" + c.getKey() + "] should keep only " +
c.getValue());
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
index 4a7cb09c9b..abb39f5897 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSource.java
@@ -31,6 +31,7 @@ import
org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
+import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeEventFilter;
import org.apache.seatunnel.connectors.cdc.base.source.IncrementalSource;
import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
import
org.apache.seatunnel.connectors.cdc.debezium.ConnectTableChangeSerializer;
@@ -129,6 +130,7 @@ public class MySqlIncrementalSource<T> extends
IncrementalSource<T, JdbcSourceCo
.setTableIdTableChangeMap(tableIdTableChangeMap)
.setSchemaChangeResolver(
new
MySqlSchemaChangeResolver(createSourceConfigFactory(config)))
+
.setSchemaChangeEventFilter(SchemaChangeEventFilter.fromConfig(config))
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
index 667f7b0654..04858dcb29 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/MySqlIncrementalSourceFactory.java
@@ -78,6 +78,8 @@ public class MySqlIncrementalSourceFactory extends
BaseChangeStreamTableSourceFa
MySqlIncrementalSourceOptions.SPLIT_ALLOW_SAMPLING,
MySqlIncrementalSourceOptions.TABLE_NAMES_CONFIG,
MySqlIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED,
+ MySqlIncrementalSourceOptions.SCHEMA_CHANGES_INCLUDE,
+ MySqlIncrementalSourceOptions.SCHEMA_CHANGES_EXCLUDE,
MySqlIncrementalSourceOptions.INT_TYPE_NARROWING)
.optional(
MySqlIncrementalSourceOptions.STARTUP_MODE,
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
index fb28bda574..c2c9e860fa 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSource.java
@@ -31,6 +31,7 @@ import
org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
+import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeEventFilter;
import org.apache.seatunnel.connectors.cdc.base.source.IncrementalSource;
import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
import
org.apache.seatunnel.connectors.cdc.debezium.ConnectTableChangeSerializer;
@@ -112,6 +113,7 @@ public class OracleIncrementalSource<T> extends
IncrementalSource<T, JdbcSourceC
.setServerTimeZone(ZoneId.of(zoneId))
.setSchemaChangeResolver(
new
OracleSchemaChangeResolver(createSourceConfigFactory(config)))
+
.setSchemaChangeEventFilter(SchemaChangeEventFilter.fromConfig(config))
.setTableIdTableChangeMap(tableIdStructMap)
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
index 7c0371478c..93e2e7f5b4 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-oracle/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/oracle/source/OracleIncrementalSourceFactory.java
@@ -78,7 +78,9 @@ public class OracleIncrementalSourceFactory extends
BaseChangeStreamTableSourceF
OracleIncrementalSourceOptions.INVERSE_SAMPLING_RATE,
OracleIncrementalSourceOptions.SPLIT_ALLOW_SAMPLING,
OracleIncrementalSourceOptions.TABLE_NAMES_CONFIG,
- OracleIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED)
+ OracleIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED,
+ OracleIncrementalSourceOptions.SCHEMA_CHANGES_INCLUDE,
+ OracleIncrementalSourceOptions.SCHEMA_CHANGES_EXCLUDE)
.optional(
OracleIncrementalSourceOptions.STARTUP_MODE,
OracleIncrementalSourceOptions.STOP_MODE)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSource.java
index c80340d99b..2945d26777 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSource.java
@@ -31,6 +31,7 @@ import
org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
import org.apache.seatunnel.connectors.cdc.base.option.JdbcSourceOptions;
import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
+import org.apache.seatunnel.connectors.cdc.base.schema.SchemaChangeEventFilter;
import org.apache.seatunnel.connectors.cdc.base.source.IncrementalSource;
import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
import
org.apache.seatunnel.connectors.cdc.debezium.ConnectTableChangeSerializer;
@@ -119,6 +120,7 @@ public class SqlServerIncrementalSource<T> extends
IncrementalSource<T, JdbcSour
.setTableIdTableChangeMap(tableIdTableChangeMap)
.setSchemaChangeResolver(
schemaChangesEnabled ? new
SqlServerSchemaChangeResolver() : null)
+
.setSchemaChangeEventFilter(SchemaChangeEventFilter.fromConfig(config))
.build();
}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactory.java
index 9436bc885f..02e40917b7 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactory.java
@@ -72,7 +72,9 @@ public class SqlServerIncrementalSourceFactory implements
TableSourceFactory {
SqlServerIncrementalSourceOptions.INVERSE_SAMPLING_RATE,
SqlServerIncrementalSourceOptions.SPLIT_ALLOW_SAMPLING,
SqlServerIncrementalSourceOptions.TABLE_NAMES_CONFIG,
-
SqlServerIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED)
+
SqlServerIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED,
+
SqlServerIncrementalSourceOptions.SCHEMA_CHANGES_INCLUDE,
+
SqlServerIncrementalSourceOptions.SCHEMA_CHANGES_EXCLUDE)
.optional(
SqlServerIncrementalSourceOptions.STARTUP_MODE,
SqlServerIncrementalSourceOptions.STOP_MODE)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
index b865fd8847..d2150a0103 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/MysqlCDCWithSchemaChangeIT.java
@@ -51,6 +51,7 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
@@ -70,6 +71,9 @@ public class MysqlCDCWithSchemaChangeIT extends TestSuiteBase
implements TestRes
private static final String SINK_TABLE =
"mysql_cdc_e2e_sink_table_with_schema_change";
private static final String SINK_TABLE2 =
"mysql_cdc_e2e_sink_table_with_schema_change_exactly_once";
+ private static final String SINK_TABLE_FILTER =
"mysql_cdc_e2e_sink_table_schema_change_filter";
+ private static final String STABLE_QUERY =
+ "select id,name,description,weight from %s.%s order by id";
private static final String MYSQL_HOST = "mysql_cdc_e2e";
private static final String MYSQL_USER_NAME = "mysqluser";
private static final String MYSQL_USER_PASSWORD = "mysqlpw";
@@ -198,6 +202,123 @@ public class MysqlCDCWithSchemaChangeIT extends
TestSuiteBase implements TestRes
assertSchemaEvolution(MYSQL_DATABASE, SOURCE_TABLE, SINK_TABLE2);
}
+ /**
+ * Regression for issue #11044. With {@code schema-changes.exclude =
["drop.column"]}: a dropped
+ * column must NOT propagate to the sink (it stays in the sink schema),
and the data changes
+ * happening at the same time must still reach the sink.
+ *
+ * <p>The dropped column here is intentionally <b>NULLABLE</b>. #11044 is
event-type filtering
+ * only and, per its non-goals, does not define a schema-change
data-handling policy, so the
+ * source simply writes {@code null} for a retained-but-no-longer-supplied
column — valid for a
+ * nullable column. Excluding {@code drop.column} for a NOT NULL column is
a known limitation
+ * that fails at the sink and is deferred to a future behavior-policy
feature; see MySQL-CDC.md.
+ * Using a nullable column is exactly why this test needs its own {@code
*_filter} DDL templates
+ * instead of the shared {@code add_columns}/{@code drop_columns} (which
add/drop NOT NULL
+ * columns).
+ */
+ @Order(3)
+ @TestTemplate
+ public void testMysqlCdcSchemaChangeEventTypeFilter(TestContainer
container) {
+ shopDatabase.setTemplateName("shop").createAndInitialize();
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+
container.executeJob("/mysqlcdc_to_mysql_with_schema_change_filter.conf");
+ } catch (Exception e) {
+ log.error("Commit task exception :" + e.getMessage());
+ throw new RuntimeException(e);
+ }
+ });
+
+ // initial snapshot synced
+ await().atMost(30000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertIterableEquals(
+ query(
+ String.format(
+ STABLE_QUERY,
+ MYSQL_DATABASE,
+ SOURCE_TABLE)),
+ query(
+ String.format(
+ STABLE_QUERY,
+ MYSQL_DATABASE,
+ SINK_TABLE_FILTER))));
+
+ // add.column is NOT excluded
+
shopDatabase.setTemplateName("add_columns_filter").createAndInitialize();
+ await().atMost(30000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertTrue(
+ columnExists(
+ MYSQL_DATABASE,
SINK_TABLE_FILTER, "add_column1"),
+ "add.column should propagate to the
sink"));
+
+ // drop.column IS excluded; this template also inserts/updates/deletes
rows at the same time
+
shopDatabase.setTemplateName("drop_columns_filter").createAndInitialize();
+
+ // regression: the concurrent data changes must still reach the sink
(job did not crash)
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertIterableEquals(
+ query(
+ String.format(
+ STABLE_QUERY,
+ MYSQL_DATABASE,
+ SOURCE_TABLE)),
+ query(
+ String.format(
+ STABLE_QUERY,
+ MYSQL_DATABASE,
+ SINK_TABLE_FILTER))));
+
+ // Row-level hardening: drop_columns_filter.sql performs INSERT +
UPDATE + DELETE alongside
+ // the filtered drop.column DDL. Every one of those data changes must
be reflected in the
+ // sink (the earlier iterable-equals already converged, so these reads
are stable).
+ List<List<Object>> sourceRows =
+ query(String.format(STABLE_QUERY, MYSQL_DATABASE,
SOURCE_TABLE));
+ List<List<Object>> sinkRows =
+ query(String.format(STABLE_QUERY, MYSQL_DATABASE,
SINK_TABLE_FILTER));
+ Assertions.assertEquals(
+ sourceRows.size(),
+ sinkRows.size(),
+ "sink row count must match source after the concurrent
INSERT/UPDATE/DELETE");
+ // DELETE propagated: drop_columns_filter.sql runs `delete from
products where id = 102`.
+ Assertions.assertTrue(
+ sinkRows.stream().noneMatch(row -> ((Number)
row.get(0)).intValue() == 102),
+ "rows deleted at the source must also be deleted in the sink");
+ // INSERT propagated: id 110 is inserted by drop_columns_filter.sql.
+ Assertions.assertTrue(
+ sinkRows.stream().anyMatch(row -> ((Number)
row.get(0)).intValue() == 110),
+ "rows inserted at the source must appear in the sink");
+ // UPDATE propagated: `set name='dailai' where id = 101`.
+ assertSinkNameEquals(sinkRows, 101, "dailai");
+
+ // the excluded drop.column must NOT have been applied to the sink
schema
+ Assertions.assertTrue(
+ columnExists(MYSQL_DATABASE, SINK_TABLE_FILTER, "add_column1"),
+ "drop.column was excluded, so the sink must keep the column
the source dropped");
+ }
+
+ /** Asserts the sink row with the given id exists and its {@code name}
matches expectedName. */
+ private void assertSinkNameEquals(List<List<Object>> sinkRows, int id,
Object expectedName) {
+ Optional<List<Object>> row =
+ sinkRows.stream().filter(r -> ((Number) r.get(0)).intValue()
== id).findFirst();
+ Assertions.assertTrue(row.isPresent(), "expected sink row with id=" +
id);
+ Assertions.assertEquals(
+ expectedName,
+ row.get().get(1),
+ "updated value must propagate to the sink for id=" + id);
+ }
+
+ private boolean columnExists(String database, String table, String column)
{
+ return query(String.format(DESC, database, table)).stream()
+ .anyMatch(row ->
column.equalsIgnoreCase(String.valueOf(row.get(0))));
+ }
+
private void assertSchemaEvolution(String database, String sourceTable,
String sinkTable) {
await().atMost(30000, TimeUnit.MILLISECONDS)
.untilAsserted(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/add_columns_filter.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/add_columns_filter.sql
new file mode 100644
index 0000000000..814bcbf7af
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/add_columns_filter.sql
@@ -0,0 +1,24 @@
+--
+-- 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.
+--
+
+--
----------------------------------------------------------------------------------------------------------------
+-- DATABASE: shop
+--
----------------------------------------------------------------------------------------------------------------
+CREATE DATABASE IF NOT EXISTS `shop`;
+use shop;
+
+alter table products add column add_column1 varchar(64) null;
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/drop_columns_filter.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/drop_columns_filter.sql
new file mode 100644
index 0000000000..1376bedd1f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/drop_columns_filter.sql
@@ -0,0 +1,29 @@
+--
+-- 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.
+--
+
+--
----------------------------------------------------------------------------------------------------------------
+-- DATABASE: shop
+--
----------------------------------------------------------------------------------------------------------------
+CREATE DATABASE IF NOT EXISTS `shop`;
+use shop;
+
+alter table products drop column add_column1;
+insert into products
+values (110,"spare tire","24 inch spare tire",22.2),
+ (111,"new battery","12V battery",8.1);
+update products set name = 'dailai' where id = 101;
+delete from products where id = 102;
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/shop.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/shop.sql
index 9887b4e687..bddc222fb5 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/shop.sql
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/ddl/shop.sql
@@ -46,6 +46,14 @@ CREATE TABLE if not exists
mysql_cdc_e2e_sink_table_with_schema_change_exactly_o
weight FLOAT
);
+drop table if exists mysql_cdc_e2e_sink_table_schema_change_filter;
+CREATE TABLE if not exists mysql_cdc_e2e_sink_table_schema_change_filter (
+ id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+ name VARCHAR(255) NOT NULL DEFAULT 'SeaTunnel',
+ description VARCHAR(512),
+ weight FLOAT
+);
+
ALTER TABLE products AUTO_INCREMENT = 101;
INSERT INTO products
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_schema_change_filter.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_schema_change_filter.conf
new file mode 100644
index 0000000000..828ef81cc2
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/resources/mysqlcdc_to_mysql_with_schema_change_filter.conf
@@ -0,0 +1,53 @@
+#
+# 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.
+#
+
+# Schema change event-type filtering (issue #11044):
+# add.column propagates, drop.column is excluded. Regression guard: a filtered
+# drop.column must NOT desync the produced row shape from the sink schema, so
+# subsequent data changes keep flowing instead of crashing the sink.
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MySQL-CDC {
+ server-id = 5680-5685
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["shop.products"]
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+
+ schema-changes.enabled = true
+ schema-changes.exclude = ["drop.column"]
+ }
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/shop"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "st_user_sink"
+ password = "mysqlpw"
+ generate_sink_sql = true
+ database = shop
+ table = mysql_cdc_e2e_sink_table_schema_change_filter
+ primary_keys = ["id"]
+ }
+}
