This is an automated email from the ASF dual-hosted git repository.
dybyte 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 b4fe1ef25b [Test][SqlServer-CDC] Add multi-table e2e testcase (#11054)
b4fe1ef25b is described below
commit b4fe1ef25b67557db010ef6f3672e21feeb57b5e
Author: David Zollo <[email protected]>
AuthorDate: Sun Jun 14 01:45:35 2026 +0800
[Test][SqlServer-CDC] Add multi-table e2e testcase (#11054)
Co-authored-by: Daniel <[email protected]>
---
.../connector/cdc/sqlserver/SqlServerCDCIT.java | 69 +++++++++++++
.../src/test/resources/ddl/column_type_test.sql | 112 +++++++++++++++++++++
..._sqlserver_with_multi_table_mode_two_table.conf | 60 +++++++++++
3 files changed, 241 insertions(+)
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
index 8df8f19a8f..fc48f3f81e 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java
@@ -98,12 +98,20 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
+ "EXEC sys.sp_cdc_disable_db";
private static final String SOURCE_TABLE =
DATABASE_NAME + "." + SCHEMA_NAME + "." + "full_types";
+ // Additional source table used to verify multi-table CDC capture in one
job.
+ private static final String SOURCE_TABLE_2 =
+ DATABASE_NAME + "." + SCHEMA_NAME + "." + "full_types_2";
private static final String SOURCE_TABLE_NO_PRIMARY_KEY =
DATABASE_NAME + "." + SCHEMA_NAME + "." +
"full_types_no_primary_key";
private static final String SOURCE_TABLE_CUSTOM_PRIMARY_KEY =
DATABASE_NAME + "." + SCHEMA_NAME + "." +
"full_types_custom_primary_key";
private static final String SINK_TABLE =
DATABASE_NAME + "." + SCHEMA_NAME + "." + "full_types_sink";
+ // Sink tables are derived from the source names with the configured sink_
prefix.
+ private static final String MULTI_TABLE_SINK_1 =
+ DATABASE_NAME + "." + SCHEMA_NAME + "." + "sink_full_types";
+ private static final String MULTI_TABLE_SINK_2 =
+ DATABASE_NAME + "." + SCHEMA_NAME + "." + "sink_full_types_2";
private static final String SELECT_SOURCE_SQL =
"select\n"
@@ -259,6 +267,67 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
});
}
+ /**
+ * Verifies that a single SqlServer CDC source can capture multiple tables
and route them to
+ * different sink tables in the same database.
+ *
+ * <p>The sink tables are pre-created so this regression stays focused on
multi-table routing
+ * instead of SQL Server auto-create type derivation while still
exercising Jdbc table-mode
+ * writes.
+ */
+ @TestTemplate
+ public void testSqlServerCdcMultiTableE2e(TestContainer container) {
+ initializeSqlServerTable(DATABASE_NAME);
+
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ container.executeJob(
+
"/sqlservercdc_to_sqlserver_with_multi_table_mode_two_table.conf");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return null;
+ });
+
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertAll(
+ () ->
+
Assertions.assertIterableEquals(
+
querySql(SELECT_SOURCE_SQL, SOURCE_TABLE),
+ querySql(
+
SELECT_SINK_SQL,
+
MULTI_TABLE_SINK_1)),
+ () ->
+
Assertions.assertIterableEquals(
+
querySql(SELECT_SOURCE_SQL, SOURCE_TABLE_2),
+ querySql(
+
SELECT_SINK_SQL,
+
MULTI_TABLE_SINK_2))));
+
+ updateSourceTable(SOURCE_TABLE);
+ updateSourceTable(SOURCE_TABLE_2);
+
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertAll(
+ () ->
+
Assertions.assertIterableEquals(
+
querySql(SELECT_SOURCE_SQL, SOURCE_TABLE),
+ querySql(
+
SELECT_SINK_SQL,
+
MULTI_TABLE_SINK_1)),
+ () ->
+
Assertions.assertIterableEquals(
+
querySql(SELECT_SOURCE_SQL, SOURCE_TABLE_2),
+ querySql(
+
SELECT_SINK_SQL,
+
MULTI_TABLE_SINK_2))));
+ }
+
@TestTemplate
@DisabledOnContainer(
value = {},
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/column_type_test.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/column_type_test.sql
index 64ceebcaa8..72cd6a1ac5 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/column_type_test.sql
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/column_type_test.sql
@@ -75,6 +75,56 @@ INSERT INTO full_types VALUES (2,
'<a>b</a>',SYSDATETIMEOFFSET(),CAST('test_varbinary' AS varbinary(100)), 5.32);
EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name =
'full_types', @role_name = NULL, @supports_net_changes = 0;
+CREATE TABLE full_types_2 (
+ id int NOT NULL,
+ val_char char(3),
+ val_varchar varchar(1000),
+ val_text text,
+ val_nchar nchar(3),
+ val_nvarchar nvarchar(1000),
+ val_ntext ntext,
+ val_decimal decimal(6,3),
+ val_numeric numeric,
+ val_float float,
+ val_real real,
+ val_smallmoney smallmoney,
+ val_money money,
+ val_bit bit,
+ val_tinyint tinyint,
+ val_smallint smallint,
+ val_int int,
+ val_bigint bigint,
+ val_date date,
+ val_time time,
+ val_datetime2 datetime2,
+ val_datetime datetime,
+ val_smalldatetime smalldatetime,
+ val_xml xml,
+ val_datetimeoffset DATETIMEOFFSET(4),
+ val_varbinary varbinary(100),
+ val_udtdecimal UDTDECIMAL,
+ PRIMARY KEY (id)
+);
+INSERT INTO full_types_2 VALUES (0,
+ 'cč0', 'vcč', 'tč', N'cč', N'vcč', N'tč',
+ 1.123, 2, 3.323, 4.323, 5.323, 6.323,
+ 1, 22, 333, 4444, 55555,
+ '2018-07-13', '10:23:45', '2018-07-13
11:23:45.34', '2018-07-13 13:23:45.78', '2018-07-13 14:23:45',
+
'<a>b</a>',SYSDATETIMEOFFSET(),CAST('test_varbinary' AS varbinary(100)), 5.32);
+INSERT INTO full_types_2 VALUES (1,
+ 'cč1', 'vcč', 'tč', N'cč', N'vcč', N'tč',
+ 1.123, 2, 3.323, 4.323, 5.323, 6.323,
+ 1, 22, 333, 4444, 55555,
+ '2018-07-13', '10:23:45', '2018-07-13
11:23:45.34', '2018-07-13 13:23:45.78', '2018-07-13 14:23:45',
+
'<a>b</a>',SYSDATETIMEOFFSET(),CAST('test_varbinary' AS varbinary(100)), 5.32);
+INSERT INTO full_types_2 VALUES (2,
+ 'cč2', 'vcč', 'tč', N'cč', N'vcč', N'tč',
+ 1.123, 2, 3.323, 4.323, 5.323, 6.323,
+ 1, 22, 333, 4444, 55555,
+ '2018-07-13', '10:23:45', '2018-07-13
11:23:45.34', '2018-07-13 13:23:45.78', '2018-07-13 14:23:45',
+
'<a>b</a>',SYSDATETIMEOFFSET(),CAST('test_varbinary' AS varbinary(100)), 5.32);
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name =
'full_types_2', @role_name = NULL, @supports_net_changes = 0;
+
CREATE TABLE full_types_no_primary_key (
id int NOT NULL,
val_char char(3),
@@ -203,3 +253,65 @@ CREATE TABLE full_types_sink (
val_udtdecimal UDTDECIMAL,
PRIMARY KEY (id)
);
+
+CREATE TABLE sink_full_types (
+ id int NOT NULL,
+ val_char char(3),
+ val_varchar varchar(1000),
+ val_text text,
+ val_nchar nchar(3),
+ val_nvarchar nvarchar(1000),
+ val_ntext ntext,
+ val_decimal decimal(6,3),
+ val_numeric numeric,
+ val_float float,
+ val_real real,
+ val_smallmoney smallmoney,
+ val_money money,
+ val_bit bit,
+ val_tinyint tinyint,
+ val_smallint smallint,
+ val_int int,
+ val_bigint bigint,
+ val_date date,
+ val_time time,
+ val_datetime2 datetime2,
+ val_datetime datetime,
+ val_smalldatetime smalldatetime,
+ val_xml xml,
+ val_datetimeoffset DATETIMEOFFSET(4),
+ val_varbinary varbinary(100),
+ val_udtdecimal UDTDECIMAL,
+ PRIMARY KEY (id)
+);
+
+CREATE TABLE sink_full_types_2 (
+ id int NOT NULL,
+ val_char char(3),
+ val_varchar varchar(1000),
+ val_text text,
+ val_nchar nchar(3),
+ val_nvarchar nvarchar(1000),
+ val_ntext ntext,
+ val_decimal decimal(6,3),
+ val_numeric numeric,
+ val_float float,
+ val_real real,
+ val_smallmoney smallmoney,
+ val_money money,
+ val_bit bit,
+ val_tinyint tinyint,
+ val_smallint smallint,
+ val_int int,
+ val_bigint bigint,
+ val_date date,
+ val_time time,
+ val_datetime2 datetime2,
+ val_datetime datetime,
+ val_smalldatetime smalldatetime,
+ val_xml xml,
+ val_datetimeoffset DATETIMEOFFSET(4),
+ val_varbinary varbinary(100),
+ val_udtdecimal UDTDECIMAL,
+ PRIMARY KEY (id)
+);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_to_sqlserver_with_multi_table_mode_two_table.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_to_sqlserver_with_multi_table_mode_two_table.conf
new file mode 100644
index 0000000000..f367f29e4b
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_to_sqlserver_with_multi_table_mode_two_table.conf
@@ -0,0 +1,60 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in
seatunnel config
+######
+
+env {
+ # You can set engine configuration here
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the
feature source plugin**
+ 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"
+ }
+}
+
+transform {
+}
+
+sink {
+ Jdbc {
+ plugin_input = "customers"
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+ url = "jdbc:sqlserver://sqlserver-host:1433;encrypt=false"
+ user = "sa"
+ password = "Password!"
+ # Pre-create both sink tables so this E2E only validates multi-table CDC
routing.
+ # Keep table-mode writes enabled because Jdbc sink currently reserves
query mode for
+ # generate_sink_sql = false.
+ generate_sink_sql = true
+ database = "column_type_test"
+ schema = "dbo"
+ tablePrefix = "sink_"
+ batch_size = 1
+ primary_keys = ["id"]
+ }
+}
