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 32bef0e8a2 [Feature][Connector-V2][SqlServer CDC] support sqlserver
schema evolution (#10890)
32bef0e8a2 is described below
commit 32bef0e8a262b2ff46808f117628b3591dbd5590
Author: cloverdew <[email protected]>
AuthorDate: Wed May 27 21:47:35 2026 +0800
[Feature][Connector-V2][SqlServer CDC] support sqlserver schema evolution
(#10890)
---
.../cdc/base/utils/SourceRecordUtils.java | 3 +-
.../connector/sqlserver/SqlServerConnection.java | 148 +++++-
.../SqlServerStreamingChangeEventSource.java | 182 +++++++
.../config/SqlServerSourceConfigFactory.java | 5 +-
.../cdc/sqlserver/source/SqlServerDialect.java | 2 +-
.../source/SqlServerIncrementalSource.java | 62 ++-
.../source/SqlServerIncrementalSourceFactory.java | 3 +-
.../source/SqlServerSchemaChangeResolver.java | 576 +++++++++++++++++++++
.../sqlserver/SqlServerConnectionTest.java | 17 +
.../config/SqlServerSourceConfigFactoryTest.java | 47 ++
.../source/SqlServerSchemaChangeResolverTest.java | 299 +++++++++++
.../dialect/sqlserver/SqlServerDialect.java | 98 +++-
.../dialect/sqlserver/SqlServerDialectTest.java | 60 +++
.../connector/cdc/sqlserver/SqlServerCDCIT.java | 183 ++++++-
.../src/test/resources/ddl/schema_change_test.sql | 47 ++
.../ddl/sqlserver_schema_change_add_columns.sql | 55 ++
.../ddl/sqlserver_schema_change_drop_columns.sql | 47 ++
.../ddl/sqlserver_schema_change_modify_columns.sql | 36 ++
.../ddl/sqlserver_schema_change_rename_columns.sql | 48 ++
...lservercdc_to_sqlserver_with_schema_change.conf | 52 ++
20 files changed, 1909 insertions(+), 61 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/utils/SourceRecordUtils.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/utils/SourceRecordUtils.java
index 370fd93b36..7237d99dbc 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/utils/SourceRecordUtils.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/utils/SourceRecordUtils.java
@@ -51,7 +51,8 @@ public class SourceRecordUtils {
public static final List<String> SUPPORT_SCHEMA_CHANGE_EVENT_KEY_NAME =
Arrays.asList(
"io.debezium.connector.mysql.SchemaChangeKey",
- "io.debezium.connector.oracle.SchemaChangeKey");
+ "io.debezium.connector.oracle.SchemaChangeKey",
+ "io.debezium.connector.sqlserver.SchemaChangeKey");
public static final String HEARTBEAT_VALUE_SCHEMA_KEY_NAME =
"io.debezium.connector.common.Heartbeat";
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java
index 3a2bb3c729..0c790b954a 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerConnection.java
@@ -129,6 +129,15 @@ public class SqlServerConnection extends JdbcConnection {
private static final String GET_NEW_CHANGE_TABLES =
"SELECT * FROM [#db].cdc.change_tables WHERE start_lsn BETWEEN ?
AND ?";
+ private static final String GET_DDL_HISTORY =
+ "SELECT OBJECT_SCHEMA_NAME(source_object_id, DB_ID(?)),"
+ + " OBJECT_NAME(source_object_id, DB_ID(?)),"
+ + " ddl_command,"
+ + " ddl_lsn,"
+ + " ddl_time"
+ + " FROM [#db].cdc.ddl_history"
+ + " WHERE ddl_lsn > ? AND ddl_lsn <= ?"
+ + " ORDER BY ddl_lsn ASC";
private static final String OPENING_QUOTING_CHARACTER = "[";
private static final String CLOSING_QUOTING_CHARACTER = "]";
@@ -377,36 +386,29 @@ public class SqlServerConnection extends JdbcConnection {
protected Optional<ColumnEditor> readTableColumn(
ResultSet columnMetadata, TableId tableId, Tables.ColumnNameFilter
columnFilter)
throws SQLException {
- return doReadTableColumn(columnMetadata, tableId, columnFilter);
+ return doReadTableColumn(columnMetadata, tableId, columnFilter, null);
}
+ /**
+ * Reads a single column from the JDBC metadata ResultSet.
+ *
+ * @param columnTypeMapping pre-fetched UDT name map (column name → type
name) for the table, or
+ * {@code null} to fall back to per-column query
+ */
private Optional<ColumnEditor> doReadTableColumn(
- ResultSet columnMetadata, TableId tableId, Tables.ColumnNameFilter
columnFilter)
+ ResultSet columnMetadata,
+ TableId tableId,
+ Tables.ColumnNameFilter columnFilter,
+ Map<String, String> columnTypeMapping)
throws SQLException {
// Oracle drivers require this for LONG/LONGRAW to be fetched first.
final String defaultValue = columnMetadata.getString(13);
- String tableSql =
- StringUtils.isNotEmpty(tableId.table())
- ? "AND tbl.name = '" + tableId.table() + "'"
- : "";
-
- Map<String, String> columnTypeMapping = new HashMap<>();
- // Support user-defined types (UDTs)
- try (PreparedStatement ps =
- connection()
- .prepareStatement(
- String.format(
- SELECT_COLUMNS_SQL_TEMPLATE,
- tableId.schema(),
- tableSql));
- ResultSet resultSet = ps.executeQuery()) {
- while (resultSet.next()) {
- String columnName = resultSet.getString("column_name");
- String dataType = resultSet.getString("type");
- columnTypeMapping.put(columnName, dataType);
- }
+ if (columnTypeMapping == null) {
+ // Fallback: fetch UDT mapping for this single call
+ columnTypeMapping = fetchColumnTypeMapping(tableId);
}
+
final String columnName = columnMetadata.getString(4);
if (columnFilter == null
|| columnFilter.matches(
@@ -446,6 +448,32 @@ public class SqlServerConnection extends JdbcConnection {
return Optional.empty();
}
+ /**
+ * Fetches UDT type names for all columns in {@code tableId} in a single
query.
+ *
+ * @return map of column name -> resolved type name (empty map if the
query returns no rows)
+ */
+ private Map<String, String> fetchColumnTypeMapping(TableId tableId) throws
SQLException {
+ String tableSql =
+ StringUtils.isNotEmpty(tableId.table())
+ ? "AND tbl.name = '" + tableId.table() + "'"
+ : "";
+ Map<String, String> mapping = new HashMap<>();
+ try (PreparedStatement ps =
+ connection()
+ .prepareStatement(
+ String.format(
+ SELECT_COLUMNS_SQL_TEMPLATE,
+ tableId.schema(),
+ tableSql));
+ ResultSet resultSet = ps.executeQuery()) {
+ while (resultSet.next()) {
+ mapping.put(resultSet.getString("column_name"),
resultSet.getString("type"));
+ }
+ }
+ return mapping;
+ }
+
/**
* Provides all changes recorder by the SQL Server CDC capture process for
a set of tables.
*
@@ -695,12 +723,42 @@ public class SqlServerConnection extends JdbcConnection {
});
}
+ public List<SqlServerDdlEntry> getDdlHistory(String databaseName, Lsn
fromLsn, Lsn toLsn)
+ throws SQLException {
+ final String query = replaceDatabaseNamePlaceholder(GET_DDL_HISTORY,
databaseName);
+
+ return prepareQueryAndMap(
+ query,
+ ps -> {
+ ps.setString(1, databaseName);
+ ps.setString(2, databaseName);
+ ps.setBytes(3, fromLsn.getBinary());
+ ps.setBytes(4, toLsn.getBinary());
+ },
+ rs -> {
+ final List<SqlServerDdlEntry> ddlEntries = new
ArrayList<>();
+ while (rs.next()) {
+ ddlEntries.add(
+ new SqlServerDdlEntry(
+ new TableId(databaseName,
rs.getString(1), rs.getString(2)),
+ rs.getString(3),
+ Lsn.valueOf(rs.getBytes(4)),
+ rs.getTimestamp(5)));
+ }
+ return ddlEntries;
+ });
+ }
+
public Table getTableSchemaFromTable(String databaseName,
SqlServerChangeTable changeTable)
throws SQLException {
final DatabaseMetaData metadata = connection().getMetaData();
JdbcIdentifierUtils.IdentifierCaseStrategy identifierCaseStrategy =
JdbcIdentifierUtils.identifierCaseStrategy(metadata);
+ // Fetch UDT type mapping once for the whole table to avoid N queries
inside the column
+ final Map<String, String> columnTypeMapping =
+ fetchColumnTypeMapping(changeTable.getSourceTableId());
+
List<Column> columns = new ArrayList<>();
int filteredRows = 0;
try (ResultSet rs =
@@ -728,7 +786,7 @@ public class SqlServerConnection extends JdbcConnection {
filteredRows++;
continue;
}
- readTableColumn(rs, changeTable.getSourceTableId(), null)
+ doReadTableColumn(rs, changeTable.getSourceTableId(), null,
columnTypeMapping)
.ifPresent(
ce -> {
// Filter out columns not included in the
change table.
@@ -763,6 +821,50 @@ public class SqlServerConnection extends JdbcConnection {
return captureName + "_CT";
}
+ public static class SqlServerDdlEntry {
+ private final TableId sourceTableId;
+ private final String ddl;
+ private final Lsn ddlLsn;
+ private final java.sql.Timestamp ddlTime;
+
+ public SqlServerDdlEntry(
+ TableId sourceTableId, String ddl, Lsn ddlLsn,
java.sql.Timestamp ddlTime) {
+ this.sourceTableId = sourceTableId;
+ this.ddl = ddl;
+ this.ddlLsn = ddlLsn;
+ this.ddlTime = ddlTime;
+ }
+
+ public TableId getSourceTableId() {
+ return sourceTableId;
+ }
+
+ public String getDdl() {
+ return ddl;
+ }
+
+ public Lsn getDdlLsn() {
+ return ddlLsn;
+ }
+
+ public java.sql.Timestamp getDdlTime() {
+ return ddlTime;
+ }
+
+ @Override
+ public String toString() {
+ return "SqlServerDdlEntry{"
+ + "sourceTableId="
+ + sourceTableId
+ + ", ddlLsn="
+ + ddlLsn
+ + ", ddl='"
+ + ddl
+ + "'"
+ + '}';
+ }
+ }
+
/**
* Retrieve the name of the database in the original case as it's defined
on the server.
*
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerStreamingChangeEventSource.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerStreamingChangeEventSource.java
index 68438f4ad2..57d1a40276 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerStreamingChangeEventSource.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/io/debezium/connector/sqlserver/SqlServerStreamingChangeEventSource.java
@@ -37,6 +37,7 @@ import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -240,6 +241,8 @@ public class SqlServerStreamingChangeEventSource
databaseName,
lastProcessedPosition.getCommitLsn())
: lastProcessedPosition.getCommitLsn();
streamingExecutionContext.setShouldIncreaseFromLsn(true);
+ final Queue<SqlServerConnection.SqlServerDdlEntry> ddlEntries =
+ getDdlEntriesToQuery(databaseName, fromLsn, toLsn);
while (!schemaChangeCheckpoints.isEmpty()) {
migrateTable(partition, schemaChangeCheckpoints,
offsetContext);
@@ -396,6 +399,14 @@ public class SqlServerStreamingChangeEventSource
offsetContext);
}
}
+ dispatchPendingDdlHistory(
+ partition,
+ offsetContext,
+ tablesSlot.get(),
+ ddlEntries,
+ tableWithSmallestLsn
+ .getChangePosition()
+ .getCommitLsn());
final TableId tableId =
tableWithSmallestLsn
.getChangeTable()
@@ -455,6 +466,12 @@ public class SqlServerStreamingChangeEventSource
clock));
tableWithSmallestLsn.next();
}
+ dispatchPendingDdlHistory(
+ partition,
+ offsetContext,
+ tablesSlot.get(),
+ ddlEntries,
+ null);
});
streamingExecutionContext.setLastProcessedPosition(
TxLogPosition.valueOf(toLsn));
@@ -512,6 +529,171 @@ public class SqlServerStreamingChangeEventSource
newTable.setSourceTable(tableSchema);
}
+ private Queue<SqlServerConnection.SqlServerDdlEntry> getDdlEntriesToQuery(
+ String databaseName, Lsn fromLsn, Lsn toLsn) throws SQLException {
+ Queue<SqlServerConnection.SqlServerDdlEntry> ddlEntries =
+ new PriorityQueue<>(
+
Comparator.comparing(SqlServerConnection.SqlServerDdlEntry::getDdlLsn));
+ dataConnection.getDdlHistory(databaseName, fromLsn, toLsn).stream()
+ .filter(
+ ddlEntry ->
+ connectorConfig
+ .getTableFilters()
+ .dataCollectionFilter()
+
.isIncluded(ddlEntry.getSourceTableId()))
+ .forEach(ddlEntries::add);
+ return ddlEntries;
+ }
+
+ private void dispatchPendingDdlHistory(
+ SqlServerPartition partition,
+ SqlServerOffsetContext offsetContext,
+ SqlServerChangeTable[] currentTables,
+ Queue<SqlServerConnection.SqlServerDdlEntry> ddlEntries,
+ Lsn currentCommitLsn)
+ throws InterruptedException, SQLException {
+ while (!ddlEntries.isEmpty()
+ && (currentCommitLsn == null
+ ||
ddlEntries.peek().getDdlLsn().compareTo(currentCommitLsn) <= 0)) {
+ SqlServerConnection.SqlServerDdlEntry ddlEntry = ddlEntries.poll();
+ dispatchDdlHistorySchemaChange(partition, offsetContext,
currentTables, ddlEntry);
+ }
+ }
+
+ private void dispatchDdlHistorySchemaChange(
+ SqlServerPartition partition,
+ SqlServerOffsetContext offsetContext,
+ SqlServerChangeTable[] currentTables,
+ SqlServerConnection.SqlServerDdlEntry ddlEntry)
+ throws InterruptedException, SQLException {
+ SqlServerChangeTable currentTable =
+ findOldestActiveChangeTable(currentTables,
ddlEntry.getSourceTableId());
+ if (currentTable == null) {
+ LOGGER.warn("Ignoring SQL Server DDL history for unknown captured
table {}", ddlEntry);
+ return;
+ }
+
+ // When a DDL event is detected via the DDL history, a newer capture
instance may exist
+ // (or be pending) for the same source table. Dispatching the schema
change here would
+ // update the Debezium internal schema registry before the old capture
instance finishes
+ // streaming its records. Subsequent records from the old capture
table (which still
+ // use the pre-DDL column layout) would then be validated against the
wider schema and
+ // crash with "Data row is smaller than a column index".
+ //
+ // The schema transition is already handled at the correct LSN
boundary by migrateTable()
+ // when the commit LSN reaches the new capture instance's startLsn.
Therefore, defer
+ // whenever a newer capture instance is known to exist.
+ if (currentTable.getStopLsn().isAvailable()) {
+ // currentTable is the old capture (a newer one is already
in-flight).
+ // migrateTable() will handle the transition at the right boundary.
+ LOGGER.info(
+ "Skipping DDL history dispatch for {} - newer capture
instance pending; "
+ + "schema transition deferred to migrateTable()",
+ ddlEntry.getSourceTableId());
+ return;
+ }
+
+ // No sibling capture in currentTables; check whether a newer capture
instance exists
+ // outside the current streaming window (its startLsn > current toLsn).
+ SqlServerChangeTable newerTable =
+ findNewerCaptureInstance(
+ partition.getDatabaseName(),
ddlEntry.getSourceTableId(), currentTable);
+ if (newerTable != null) {
+ // A newer capture instance exists but has not entered the window
yet.
+ // migrateTable() will handle the transition once its startLsn is
reached.
+ LOGGER.info(
+ "Skipping DDL history dispatch for {} - newer capture
instance exists "
+ + "outside current window; schema transition
deferred to migrateTable()",
+ ddlEntry.getSourceTableId());
+ return;
+ }
+
+ Table oldTableSchema = schema.tableFor(ddlEntry.getSourceTableId());
+ Table tableSchema =
+ metadataConnection.getTableSchemaFromTable(
+ partition.getDatabaseName(), currentTable);
+ if (oldTableSchema != null && oldTableSchema.equals(tableSchema)) {
+ LOGGER.info("Migration skipped, no table schema changes detected
for {}", ddlEntry);
+ return;
+ }
+
offsetContext.setChangePosition(TxLogPosition.valueOf(ddlEntry.getDdlLsn()), 1);
+ offsetContext.event(
+ ddlEntry.getSourceTableId(),
+ ddlEntry.getDdlTime() == null ? Instant.now() :
ddlEntry.getDdlTime().toInstant());
+ LOGGER.info(
+ "Dispatching SQL Server DDL history schema change for {} at
{}: {}",
+ ddlEntry.getSourceTableId(),
+ ddlEntry.getDdlLsn(),
+ ddlEntry.getDdl());
+ dispatcher.dispatchSchemaChangeEvent(
+ partition,
+ ddlEntry.getSourceTableId(),
+ new SqlServerSchemaChangeEventEmitter(
+ partition,
+ offsetContext,
+ currentTable,
+ tableSchema,
+ SchemaChangeEventType.ALTER));
+ currentTable.setSourceTable(tableSchema);
+ }
+
+ /**
+ * Returns the oldest (currently streaming) capture instance for {@code
sourceTableId} in the
+ * active tables array, i.e. the one whose {@code stopLsn} is set (meaning
a newer capture is
+ * already in-flight), or the sole capture instance if only one exists for
the table.
+ *
+ * <p>This is used in {@link #dispatchDdlHistorySchemaChange} to make sure
we operate on the
+ * capture instance that is actually feeding records at the current LSN,
not a future one.
+ */
+ private SqlServerChangeTable findOldestActiveChangeTable(
+ SqlServerChangeTable[] currentTables, TableId sourceTableId) {
+ if (currentTables == null) {
+ return null;
+ }
+ SqlServerChangeTable oldest = null;
+ for (SqlServerChangeTable table : currentTables) {
+ if (!table.getSourceTableId().equals(sourceTableId)) {
+ continue;
+ }
+ if (oldest == null) {
+ oldest = table;
+ } else if (table.getStopLsn().isAvailable() &&
!oldest.getStopLsn().isAvailable()) {
+ // prefer the one with a stopLsn (the old capture) over the
unbounded one
+ oldest = table;
+ } else if (table.getStartLsn().compareTo(oldest.getStartLsn()) <
0) {
+ oldest = table;
+ }
+ }
+ return oldest;
+ }
+
+ /**
+ * Queries all active change tables (without an LSN upper-bound) to find a
capture instance for
+ * {@code sourceTableId} whose start LSN is strictly greater than {@code
knownTable}'s start
+ * LSN. This is used to detect a newer capture instance that was created
as part of a
+ * schema-change operation but has not yet entered the active streaming
window.
+ *
+ * @return the newest capture instance beyond {@code knownTable}, or
{@code null} if none
+ */
+ private SqlServerChangeTable findNewerCaptureInstance(
+ String databaseName, TableId sourceTableId, SqlServerChangeTable
knownTable)
+ throws SQLException {
+ List<SqlServerChangeTable> allTables =
dataConnection.getChangeTables(databaseName);
+ SqlServerChangeTable newest = null;
+ for (SqlServerChangeTable table : allTables) {
+ if (!table.getSourceTableId().equals(sourceTableId)) {
+ continue;
+ }
+ if (table.getStartLsn().compareTo(knownTable.getStartLsn()) <= 0) {
+ continue;
+ }
+ if (newest == null ||
table.getStartLsn().compareTo(newest.getStartLsn()) > 0) {
+ newest = table;
+ }
+ }
+ return newest;
+ }
+
private SqlServerChangeTable[] processErrorFromChangeTableQuery(
String databaseName, SQLException exception,
SqlServerChangeTable[] currentChangeTables)
throws Exception {
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactory.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactory.java
index c0d96932d8..9de44488cc 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactory.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactory.java
@@ -33,6 +33,7 @@ public class SqlServerSourceConfigFactory extends
JdbcSourceConfigFactory {
private static final String DATABASE_SERVER_NAME =
"sqlserver_transaction_log_source";
private static final String DRIVER_CLASS_NAME =
"com.microsoft.sqlserver.jdbc.SQLServerDriver";
+ public static final String SCHEMA_CHANGE_KEY = "include.schema.changes";
@Override
public SqlServerSourceConfig create(int subtask) {
@@ -58,8 +59,8 @@ public class SqlServerSourceConfigFactory extends
JdbcSourceConfigFactory {
props.setProperty("database.history.skip.unparseable.ddl",
String.valueOf(true));
props.setProperty("database.history.refer.ddl", String.valueOf(true));
- // TODO Not yet supported
- props.setProperty("include.schema.changes", String.valueOf(false));
+ // Emit Debezium schema change records only when schema evolution is
enabled.
+ props.setProperty(SCHEMA_CHANGE_KEY,
String.valueOf(schemaChangeEnabled));
if (databaseList != null) {
props.setProperty("database.include.list", String.join(",",
databaseList));
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerDialect.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerDialect.java
index 55838d16b1..5c191ad507 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerDialect.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerDialect.java
@@ -51,7 +51,7 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
-/** The {@link JdbcDataSourceDialect} implementation for MySQL datasource. */
+/** The {@link JdbcDataSourceDialect} implementation for SQL Server
datasource. */
public class SqlServerDialect implements JdbcDataSourceDialect {
private static final long serialVersionUID = 1L;
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 4b046a7a7e..c80340d99b 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
@@ -20,8 +20,11 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source;
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.source.SupportParallelism;
+import org.apache.seatunnel.api.source.SupportSchemaEvolution;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.schema.SchemaChangeType;
import org.apache.seatunnel.common.utils.JdbcUrlUtil;
+import org.apache.seatunnel.common.utils.SeaTunnelException;
import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
import org.apache.seatunnel.connectors.cdc.base.config.SourceConfig;
import org.apache.seatunnel.connectors.cdc.base.dialect.DataSourceDialect;
@@ -30,6 +33,7 @@ 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.source.IncrementalSource;
import org.apache.seatunnel.connectors.cdc.base.source.offset.OffsetFactory;
+import
org.apache.seatunnel.connectors.cdc.debezium.ConnectTableChangeSerializer;
import
org.apache.seatunnel.connectors.cdc.debezium.DebeziumDeserializationSchema;
import org.apache.seatunnel.connectors.cdc.debezium.DeserializeFormat;
import
org.apache.seatunnel.connectors.cdc.debezium.row.DebeziumJsonDeserializeSchema;
@@ -39,12 +43,23 @@ import
org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source.offset.Lsn
import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.sqlserver.SqlServerURLParser;
import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
+import org.apache.kafka.connect.data.Struct;
+
+import io.debezium.jdbc.JdbcConnection;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+
import java.time.ZoneId;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
public class SqlServerIncrementalSource<T> extends IncrementalSource<T,
JdbcSourceConfig>
- implements SupportParallelism {
+ implements SupportParallelism, SupportSchemaEvolution {
static final String IDENTIFIER = "SqlServer-CDC";
@@ -84,11 +99,16 @@ public class SqlServerIncrementalSource<T> extends
IncrementalSource<T, JdbcSour
@Override
public DebeziumDeserializationSchema<T>
createDebeziumDeserializationSchema(
ReadonlyConfig config) {
+ boolean schemaChangesEnabled =
+
config.get(SqlServerIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED);
+ Map<TableId, Struct> tableIdTableChangeMap =
+ schemaChangesEnabled ? tableChanges() : Collections.emptyMap();
if (DeserializeFormat.COMPATIBLE_DEBEZIUM_JSON.equals(
config.get(JdbcSourceOptions.FORMAT))) {
return (DebeziumDeserializationSchema<T>)
new DebeziumJsonDeserializeSchema(
- config.get(JdbcSourceOptions.DEBEZIUM_PROPERTIES));
+ config.get(JdbcSourceOptions.DEBEZIUM_PROPERTIES),
+ tableIdTableChangeMap);
}
String zoneId = config.get(JdbcSourceOptions.SERVER_TIME_ZONE);
@@ -96,6 +116,9 @@ public class SqlServerIncrementalSource<T> extends
IncrementalSource<T, JdbcSour
SeaTunnelRowDebeziumDeserializeSchema.builder()
.setTables(catalogTables)
.setServerTimeZone(ZoneId.of(zoneId))
+ .setTableIdTableChangeMap(tableIdTableChangeMap)
+ .setSchemaChangeResolver(
+ schemaChangesEnabled ? new
SqlServerSchemaChangeResolver() : null)
.build();
}
@@ -114,4 +137,39 @@ public class SqlServerIncrementalSource<T> extends
IncrementalSource<T, JdbcSour
public Optional<String> driverName() {
return Optional.of("com.microsoft.sqlserver.jdbc.SQLServerDriver");
}
+
+ @Override
+ public List<SchemaChangeType> supports() {
+ return Arrays.asList(
+ SchemaChangeType.ADD_COLUMN,
+ SchemaChangeType.DROP_COLUMN,
+ SchemaChangeType.UPDATE_COLUMN,
+ SchemaChangeType.RENAME_COLUMN);
+ }
+
+ private Map<TableId, Struct> tableChanges() {
+ // Reuse the already-initialized dialect and config rather than
constructing a second
+ SqlServerDialect dialect = (SqlServerDialect) dataSourceDialect;
+ JdbcSourceConfig jdbcSourceConfig = configFactory.create(0);
+ List<TableId> discoverTables =
dialect.discoverDataCollections(jdbcSourceConfig);
+ ConnectTableChangeSerializer connectTableChangeSerializer =
+ new ConnectTableChangeSerializer();
+ try (JdbcConnection jdbcConnection =
dialect.openJdbcConnection(jdbcSourceConfig)) {
+ return discoverTables.stream()
+ .collect(
+ Collectors.toMap(
+ Function.identity(),
+ tableId -> {
+ TableChanges tableChanges = new
TableChanges();
+ tableChanges.create(
+
dialect.queryTableSchema(jdbcConnection, tableId)
+ .getTable());
+ return connectTableChangeSerializer
+ .serialize(tableChanges)
+ .get(0);
+ }));
+ } catch (Exception e) {
+ throw new SeaTunnelException(e);
+ }
+ }
}
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 2046353a20..9436bc885f 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
@@ -71,7 +71,8 @@ public class SqlServerIncrementalSourceFactory implements
TableSourceFactory {
SqlServerIncrementalSourceOptions.SAMPLE_SHARDING_THRESHOLD,
SqlServerIncrementalSourceOptions.INVERSE_SAMPLING_RATE,
SqlServerIncrementalSourceOptions.SPLIT_ALLOW_SAMPLING,
- SqlServerIncrementalSourceOptions.TABLE_NAMES_CONFIG)
+ SqlServerIncrementalSourceOptions.TABLE_NAMES_CONFIG,
+
SqlServerIncrementalSourceOptions.SCHEMA_CHANGES_ENABLED)
.optional(
SqlServerIncrementalSourceOptions.STARTUP_MODE,
SqlServerIncrementalSourceOptions.STOP_MODE)
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerSchemaChangeResolver.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerSchemaChangeResolver.java
new file mode 100644
index 0000000000..b296a04278
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerSchemaChangeResolver.java
@@ -0,0 +1,576 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source;
+
+import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+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.catalog.TablePath;
+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.SchemaChangeEvent;
+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.ConnectTableChangeSerializer;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.utils.SqlServerTypeUtils;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
+
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+import io.debezium.relational.Table;
+import io.debezium.relational.history.HistoryRecord;
+import io.debezium.relational.history.TableChanges;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class SqlServerSchemaChangeResolver implements SchemaChangeResolver {
+
+ private static final String SOURCE_DIALECT = DatabaseIdentifier.SQLSERVER;
+ private static final String SQLSERVER_SCHEMA_CHANGE_KEY =
+ "io.debezium.connector.sqlserver.SchemaChangeKey";
+
+ // This value must stay in sync with the entry in
+ // SourceRecordUtils.SUPPORT_SCHEMA_CHANGE_EVENT_KEY_NAME.
+ // TODO: expose it as a named constant in SourceRecordUtils and reference
it here.
+ private static final Pattern ALTER_TABLE_PATTERN =
+ Pattern.compile(
+
"(?i)ALTER\\s+TABLE\\s+(.+?)\\s+(ADD|DROP|ALTER|RENAME|WITH|SWITCH)\\b");
+ private static final Pattern SP_RENAME_COLUMN_PATTERN =
+ Pattern.compile(
+
"(?i)(?:EXEC(?:UTE)?\\s+)?(?:(?:\\[[^\\]]+\\]|[^.\\s]+)\\.)?(?:sys\\.)?sp_rename\\s+N?'([^']+)'\\s*,\\s*N?'([^']+)'\\s*,\\s*N?'COLUMN'");
+ private static final Pattern TABLE_IDENTIFIER_PART_PATTERN =
+
Pattern.compile("\\[([^\\]]+)]|\"([^\"]+)\"|`([^`]+)`|([^\\.\\s]+)");
+
+ private final ConnectTableChangeSerializer tableChangeSerializer =
+ new ConnectTableChangeSerializer();
+
+ @Override
+ public boolean support(SourceRecord record) {
+ if (!isSqlServerSchemaChangeEvent(record)) {
+ return false;
+ }
+ Struct value = (Struct) record.value();
+ List<Struct> tableChanges =
value.getArray(HistoryRecord.Fields.TABLE_CHANGES);
+ return tableChanges != null && !tableChanges.isEmpty();
+ }
+
+ @Override
+ public SchemaChangeEvent resolve(SourceRecord record, List<CatalogTable>
catalogTables) {
+ TablePath tablePath = resolveTablePath(record);
+ CatalogTable currentCatalogTable = findCatalogTable(catalogTables,
tablePath);
+ if (currentCatalogTable == null) {
+ log.warn("Ignoring SQL Server schema change for unknown table {}",
tablePath);
+ return null;
+ }
+
+ Table currentTable = getCurrentTable(record, tablePath);
+ if (currentTable == null) {
+ log.warn(
+ "Ignoring SQL Server schema change with missing table
change payload {}",
+ tablePath);
+ return null;
+ }
+
+ String ddl = SourceRecordUtils.getDdl(record);
+ List<AlterTableColumnEvent> events =
+ diffColumns(currentCatalogTable, currentTable,
parseSpRenameColumns(ddl));
+ if (events.isEmpty()) {
+ log.info(
+ "Ignoring SQL Server schema change without column diff for
table {}",
+ tablePath);
+ return null;
+ }
+
+ TableIdentifier tableIdentifier = currentCatalogTable.getTableId();
+ AlterTableColumnsEvent event = new
AlterTableColumnsEvent(tableIdentifier, events);
+ event.setStatement(ddl);
+ event.setSourceDialectName(SOURCE_DIALECT);
+ return event;
+ }
+
+ private CatalogTable findCatalogTable(List<CatalogTable> catalogTables,
TablePath tablePath) {
+ if (catalogTables == null) {
+ return null;
+ }
+ return catalogTables.stream()
+ .filter(table -> isSameTablePath(table.getTablePath(),
tablePath))
+ .findFirst()
+ .orElse(null);
+ }
+
+ private TablePath resolveTablePath(SourceRecord record) {
+ TablePath pathFromSource = SourceRecordUtils.getTablePath(record);
+ TablePath pathFromDdl =
parseTablePathFromDdl(SourceRecordUtils.getDdl(record));
+ if (pathFromDdl == null) {
+ return pathFromSource;
+ }
+ return TablePath.of(
+ StringUtils.defaultIfBlank(
+ pathFromSource.getDatabaseName(),
pathFromDdl.getDatabaseName()),
+ StringUtils.defaultIfBlank(
+ pathFromSource.getSchemaName(),
pathFromDdl.getSchemaName()),
+ StringUtils.defaultIfBlank(
+ pathFromSource.getTableName(),
pathFromDdl.getTableName()));
+ }
+
+ private Table getCurrentTable(SourceRecord record, TablePath tablePath) {
+ Struct value = (Struct) record.value();
+ List<Struct> tableChangesStruct =
value.getArray(HistoryRecord.Fields.TABLE_CHANGES);
+ TableChanges tableChanges =
tableChangeSerializer.deserialize(tableChangesStruct, false);
+ for (TableChanges.TableChange tableChange : tableChanges) {
+ Table table = tableChange.getTable();
+ if (table == null) {
+ continue;
+ }
+ if (isSameIdentifier(table.id().catalog(),
tablePath.getDatabaseName())
+ && isSameIdentifier(table.id().schema(),
tablePath.getSchemaName())
+ && isSameIdentifier(table.id().table(),
tablePath.getTableName())) {
+ return table;
+ }
+ }
+ return null;
+ }
+
+ private List<AlterTableColumnEvent> diffColumns(
+ CatalogTable currentCatalogTable,
+ Table currentTable,
+ Map<String, String> explicitRenames) {
+ List<Column> previousColumns =
currentCatalogTable.getTableSchema().getColumns();
+ List<io.debezium.relational.Column> newColumns =
currentTable.columns();
+ TableIdentifier tableIdentifier = currentCatalogTable.getTableId();
+
+ Map<String, Column> previousByName =
+ previousColumns.stream()
+ .collect(
+ Collectors.toMap(
+ column ->
normalizeIdentifier(column.getName()),
+ column -> column,
+ (left, right) -> left,
+ LinkedHashMap::new));
+ Map<String, io.debezium.relational.Column> currentByName =
+ newColumns.stream()
+ .collect(
+ Collectors.toMap(
+ column ->
normalizeIdentifier(column.name()),
+ column -> column,
+ (left, right) -> left,
+ LinkedHashMap::new));
+
+ Set<String> matchedNames = new HashSet<>(previousByName.keySet());
+ matchedNames.retainAll(currentByName.keySet());
+
+ List<AlterTableColumnEvent> events = new ArrayList<>();
+ for (int index = 0; index < newColumns.size(); index++) {
+ io.debezium.relational.Column newColumn = newColumns.get(index);
+ String normalizedName = normalizeIdentifier(newColumn.name());
+ if (!matchedNames.contains(normalizedName)) {
+ continue;
+ }
+ Column oldColumn = previousByName.get(normalizedName);
+ Column convertedColumn = convertColumn(newColumn);
+ boolean changed = hasColumnChanged(oldColumn, convertedColumn,
index, previousColumns);
+ if (!changed) {
+ continue;
+ }
+ AlterTableModifyColumnEvent modifyEvent =
+ buildModifyEvent(tableIdentifier, convertedColumn, index,
newColumns);
+ modifyEvent.setTypeChanged(hasTypeChanged(oldColumn,
convertedColumn));
+ modifyEvent.setSourceDialectName(SOURCE_DIALECT);
+ events.add(modifyEvent);
+ }
+
+ List<ColumnWithIndex<Column>> removedColumns = new ArrayList<>();
+ for (int index = 0; index < previousColumns.size(); index++) {
+ Column column = previousColumns.get(index);
+ if
(!currentByName.containsKey(normalizeIdentifier(column.getName()))) {
+ removedColumns.add(new ColumnWithIndex<>(column, index));
+ }
+ }
+
+ List<ColumnWithIndex<io.debezium.relational.Column>> addedColumns =
new ArrayList<>();
+ for (int index = 0; index < newColumns.size(); index++) {
+ io.debezium.relational.Column column = newColumns.get(index);
+ if
(!previousByName.containsKey(normalizeIdentifier(column.name()))) {
+ addedColumns.add(new ColumnWithIndex<>(column, index));
+ }
+ }
+
+ pairRenameColumns(
+ events, tableIdentifier, newColumns, removedColumns,
addedColumns, explicitRenames);
+
+ for (ColumnWithIndex<io.debezium.relational.Column> added :
addedColumns) {
+ Column convertedColumn = convertColumn(added.value);
+ AlterTableAddColumnEvent addEvent =
+ buildAddEvent(tableIdentifier, convertedColumn,
added.index, newColumns);
+ addEvent.setSourceDialectName(SOURCE_DIALECT);
+ events.add(addEvent);
+ }
+
+ for (ColumnWithIndex<Column> removed : removedColumns) {
+ AlterTableDropColumnEvent dropEvent =
+ new AlterTableDropColumnEvent(tableIdentifier,
removed.value.getName());
+ dropEvent.setSourceDialectName(SOURCE_DIALECT);
+ events.add(dropEvent);
+ }
+
+ return events;
+ }
+
+ /** Extracts explicit SQL Server column renames declared through {@code
sp_rename}. */
+ private Map<String, String> parseSpRenameColumns(String ddl) {
+ if (StringUtils.isBlank(ddl)) {
+ return Collections.emptyMap();
+ }
+ Map<String, String> renames = new LinkedHashMap<>();
+ Matcher matcher = SP_RENAME_COLUMN_PATTERN.matcher(ddl);
+ while (matcher.find()) {
+ String oldColumnName = extractLastIdentifierPart(matcher.group(1));
+ String newColumnName = extractLastIdentifierPart(matcher.group(2));
+ if (StringUtils.isBlank(oldColumnName) ||
StringUtils.isBlank(newColumnName)) {
+ continue;
+ }
+ renames.put(normalizeIdentifier(oldColumnName),
normalizeIdentifier(newColumnName));
+ }
+ return renames;
+ }
+
+ private void pairRenameColumns(
+ List<AlterTableColumnEvent> events,
+ TableIdentifier tableIdentifier,
+ List<io.debezium.relational.Column> newColumns,
+ List<ColumnWithIndex<Column>> removedColumns,
+ List<ColumnWithIndex<io.debezium.relational.Column>> addedColumns,
+ Map<String, String> explicitRenames) {
+ Set<ColumnWithIndex<Column>> matchedRemoved = new HashSet<>();
+ Set<ColumnWithIndex<io.debezium.relational.Column>> matchedAdded = new
HashSet<>();
+
+ for (ColumnWithIndex<io.debezium.relational.Column> added :
addedColumns) {
+ String addedName = normalizeIdentifier(added.value.name());
+ Column convertedAdded = convertColumn(added.value);
+ ColumnWithIndex<Column> renameCandidate = null;
+ for (ColumnWithIndex<Column> removed : removedColumns) {
+ if (matchedRemoved.contains(removed)) {
+ continue;
+ }
+ String removedName =
normalizeIdentifier(removed.value.getName());
+ if (!StringUtils.equalsIgnoreCase(addedName,
explicitRenames.get(removedName))) {
+ continue;
+ }
+ if (!sameDefinitionExceptName(removed.value, convertedAdded)) {
+ continue;
+ }
+ if (renameCandidate != null) {
+ renameCandidate = null;
+ break;
+ }
+ renameCandidate = removed;
+ }
+ if (renameCandidate == null) {
+ continue;
+ }
+ AlterTableChangeColumnEvent changeEvent =
+ buildChangeEvent(
+ tableIdentifier,
+ renameCandidate.value.getName(),
+ convertedAdded,
+ added.index,
+ newColumns);
+ changeEvent.setSourceDialectName(SOURCE_DIALECT);
+ events.add(changeEvent);
+ matchedRemoved.add(renameCandidate);
+ matchedAdded.add(added);
+ }
+
+ removedColumns.removeAll(matchedRemoved);
+ addedColumns.removeAll(matchedAdded);
+ }
+
+ private AlterTableAddColumnEvent buildAddEvent(
+ TableIdentifier tableIdentifier,
+ Column column,
+ int newIndex,
+ List<io.debezium.relational.Column> newColumns) {
+ if (newIndex == 0) {
+ return AlterTableAddColumnEvent.addFirst(tableIdentifier, column);
+ }
+ return AlterTableAddColumnEvent.addAfter(
+ tableIdentifier, column, newColumns.get(newIndex - 1).name());
+ }
+
+ private AlterTableModifyColumnEvent buildModifyEvent(
+ TableIdentifier tableIdentifier,
+ Column column,
+ int newIndex,
+ List<io.debezium.relational.Column> newColumns) {
+ if (newIndex == 0) {
+ return AlterTableModifyColumnEvent.modifyFirst(tableIdentifier,
column);
+ }
+ return AlterTableModifyColumnEvent.modifyAfter(
+ tableIdentifier, column, newColumns.get(newIndex - 1).name());
+ }
+
+ private AlterTableChangeColumnEvent buildChangeEvent(
+ TableIdentifier tableIdentifier,
+ String oldColumnName,
+ Column newColumn,
+ int newIndex,
+ List<io.debezium.relational.Column> newColumns) {
+ if (newIndex == 0) {
+ return AlterTableChangeColumnEvent.changeFirst(
+ tableIdentifier, oldColumnName, newColumn);
+ }
+ return AlterTableChangeColumnEvent.changeAfter(
+ tableIdentifier, oldColumnName, newColumn,
newColumns.get(newIndex - 1).name());
+ }
+
+ private boolean hasColumnChanged(
+ Column oldColumn, Column newColumn, int newIndex, List<Column>
previousColumns) {
+ if (oldColumn == null || newColumn == null) {
+ return true;
+ }
+ if (isSameColumnDefinition(oldColumn, newColumn)) {
+ int oldIndex = indexOfColumn(previousColumns, oldColumn.getName());
+ return oldIndex != newIndex;
+ }
+ return true;
+ }
+
+ private int indexOfColumn(List<Column> columns, String columnName) {
+ for (int index = 0; index < columns.size(); index++) {
+ if (isSameIdentifier(columns.get(index).getName(), columnName)) {
+ return index;
+ }
+ }
+ return -1;
+ }
+
+ private boolean hasTypeChanged(Column oldColumn, Column newColumn) {
+ return !Objects.equals(oldColumn.getDataType(),
newColumn.getDataType())
+ || !isLengthCompatible(oldColumn.getColumnLength(),
newColumn.getColumnLength())
+ || !isScaleCompatible(oldColumn.getScale(),
newColumn.getScale())
+ || !isSourceTypeCompatible(oldColumn.getSourceType(),
newColumn.getSourceType());
+ }
+
+ /**
+ * Returns {@code true} when the two columns share the same structural
definition (type, length,
+ * scale, nullability, default, comment), ignoring the column name. Used
to heuristically detect
+ * renames. Note: comment equality is included — a rename that also
changes the comment will not
+ * be detected as a rename but as DROP + ADD.
+ */
+ private boolean sameDefinitionExceptName(Column oldColumn, Column
newColumn) {
+ return isSameColumnDefinition(oldColumn, newColumn);
+ }
+
+ private Column convertColumn(io.debezium.relational.Column column) {
+ String sourceType = column.typeExpression();
+ if (StringUtils.isBlank(sourceType)) {
+ sourceType = column.typeName();
+ }
+ return PhysicalColumn.builder()
+ .name(column.name())
+ .dataType(SqlServerTypeUtils.convertFromColumn(column))
+ .columnLength(
+ column.length() ==
io.debezium.relational.Column.UNSET_INT_VALUE
+ ? null
+ : (long) column.length())
+ .scale(column.scale().orElse(null))
+ .nullable(column.isOptional())
+ .defaultValue(column.defaultValueExpression().orElse(null))
+ .comment(column.comment())
+ .sourceType(sourceType)
+ .build();
+ }
+
+ private boolean isSameColumnDefinition(Column oldColumn, Column newColumn)
{
+ return Objects.equals(oldColumn.getDataType(), newColumn.getDataType())
+ && isLengthCompatible(oldColumn.getColumnLength(),
newColumn.getColumnLength())
+ && isScaleCompatible(oldColumn.getScale(),
newColumn.getScale())
+ && Objects.equals(oldColumn.isNullable(),
newColumn.isNullable())
+ && Objects.equals(oldColumn.getDefaultValue(),
newColumn.getDefaultValue())
+ && Objects.equals(oldColumn.getComment(),
newColumn.getComment())
+ && isSourceTypeCompatible(oldColumn.getSourceType(),
newColumn.getSourceType());
+ }
+
+ private boolean isLengthCompatible(Long oldLength, Long newLength) {
+ return oldLength == null || newLength == null ||
Objects.equals(oldLength, newLength);
+ }
+
+ private boolean isScaleCompatible(Integer oldScale, Integer newScale) {
+ return oldScale == null || newScale == null ||
Objects.equals(oldScale, newScale);
+ }
+
+ private boolean isSourceTypeCompatible(String oldSourceType, String
newSourceType) {
+ if (StringUtils.isBlank(oldSourceType) ||
StringUtils.isBlank(newSourceType)) {
+ return true;
+ }
+ String normalizedOld = oldSourceType.replace(" ",
"").toLowerCase(Locale.ENGLISH);
+ String normalizedNew = newSourceType.replace(" ",
"").toLowerCase(Locale.ENGLISH);
+ if (StringUtils.equals(normalizedOld, normalizedNew)) {
+ return true;
+ }
+ // When the CDC migration path reads column metadata via JDBC
DatabaseMetaData, the
+ // typeExpression field may not be populated, causing convertColumn()
to fall back to
+ // typeName() which lacks length/precision (e.g. "varchar" instead of
"varchar(255)").
+ // Treat the types as compatible when their base names match and at
least one side is
+ // missing the length qualifier — the actual column type has not
changed in that case.
+ String baseOld =
+ normalizedOld.contains("(")
+ ? normalizedOld.substring(0,
normalizedOld.indexOf('('))
+ : normalizedOld;
+ String baseNew =
+ normalizedNew.contains("(")
+ ? normalizedNew.substring(0,
normalizedNew.indexOf('('))
+ : normalizedNew;
+ return StringUtils.equals(baseOld, baseNew)
+ && (!normalizedOld.contains("(") ||
!normalizedNew.contains("("));
+ }
+
+ private String extractLastIdentifierPart(String identifier) {
+ List<String> parts = parseIdentifierParts(identifier);
+ if (!parts.isEmpty()) {
+ return parts.get(parts.size() - 1);
+ }
+ return identifier;
+ }
+
+ private TablePath parseTablePathFromDdl(String ddl) {
+ if (StringUtils.isBlank(ddl)) {
+ return null;
+ }
+ Matcher matcher = ALTER_TABLE_PATTERN.matcher(ddl);
+ if (!matcher.find()) {
+ return null;
+ }
+ String tableIdentifier = matcher.group(1);
+ List<String> parts = parseIdentifierParts(tableIdentifier);
+ if (parts.isEmpty()) {
+ return null;
+ }
+ if (parts.size() == 1) {
+ return TablePath.of(null, null, parts.get(0));
+ }
+ if (parts.size() == 2) {
+ return TablePath.of(null, parts.get(0), parts.get(1));
+ }
+ return TablePath.of(
+ parts.get(parts.size() - 3),
+ parts.get(parts.size() - 2),
+ parts.get(parts.size() - 1));
+ }
+
+ private List<String> parseIdentifierParts(String rawIdentifier) {
+ if (StringUtils.isBlank(rawIdentifier)) {
+ return Collections.emptyList();
+ }
+ List<String> parts = new ArrayList<>();
+ Matcher matcher = TABLE_IDENTIFIER_PART_PATTERN.matcher(rawIdentifier);
+ while (matcher.find()) {
+ String part =
+ firstNonBlank(
+ matcher.group(1), matcher.group(2),
matcher.group(3), matcher.group(4));
+ if (StringUtils.isNotBlank(part)) {
+ parts.add(part.trim());
+ }
+ }
+ return parts;
+ }
+
+ private String firstNonBlank(String... values) {
+ for (String value : values) {
+ if (StringUtils.isNotBlank(value)) {
+ return value;
+ }
+ }
+ return null;
+ }
+
+ private boolean isSqlServerSchemaChangeEvent(SourceRecord record) {
+ if (record == null || record.keySchema() == null) {
+ return false;
+ }
+ String keySchemaName = record.keySchema().name();
+ return StringUtils.equalsIgnoreCase(keySchemaName,
SQLSERVER_SCHEMA_CHANGE_KEY)
+ || SourceRecordUtils.isSchemaChangeEvent(record);
+ }
+
+ private boolean isSameTablePath(TablePath left, TablePath right) {
+ return left != null
+ && right != null
+ && isSameIdentifier(left.getDatabaseName(),
right.getDatabaseName())
+ && isSameIdentifier(left.getSchemaName(),
right.getSchemaName())
+ && isSameIdentifier(left.getTableName(), right.getTableName());
+ }
+
+ private boolean isSameIdentifier(String left, String right) {
+ String normalizedLeft = normalizeIdentifier(left);
+ String normalizedRight = normalizeIdentifier(right);
+ if (StringUtils.isBlank(normalizedLeft) ||
StringUtils.isBlank(normalizedRight)) {
+ return StringUtils.equals(normalizedLeft, normalizedRight);
+ }
+ return StringUtils.equalsIgnoreCase(normalizedLeft, normalizedRight);
+ }
+
+ private String normalizeIdentifier(String identifier) {
+ if (identifier == null) {
+ return null;
+ }
+ String normalized = identifier.trim();
+ if (StringUtils.startsWith(normalized, "[") &&
StringUtils.endsWith(normalized, "]")) {
+ normalized = normalized.substring(1, normalized.length() - 1);
+ }
+ if (StringUtils.startsWith(normalized, "\"") &&
StringUtils.endsWith(normalized, "\"")) {
+ normalized = normalized.substring(1, normalized.length() - 1);
+ }
+ if (StringUtils.startsWith(normalized, "`") &&
StringUtils.endsWith(normalized, "`")) {
+ normalized = normalized.substring(1, normalized.length() - 1);
+ }
+ return normalized;
+ }
+
+ private static class ColumnWithIndex<T> {
+
+ private final T value;
+ private final int index;
+
+ private ColumnWithIndex(T value, int index) {
+ this.value = value;
+ this.index = index;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/io/debezium/connector/sqlserver/SqlServerConnectionTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/io/debezium/connector/sqlserver/SqlServerConnectionTest.java
index f74f545fd4..a9487185d9 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/io/debezium/connector/sqlserver/SqlServerConnectionTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/io/debezium/connector/sqlserver/SqlServerConnectionTest.java
@@ -30,6 +30,7 @@ import io.debezium.relational.Tables;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
+import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Types;
@@ -37,6 +38,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Optional;
+import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.mock;
@@ -58,13 +60,22 @@ public class SqlServerConnectionTest {
when(columnsRs.getString("TABLE_NAME")).thenReturn("user_info",
"userAinfo");
when(columnsRs.getString("TABLE_SCHEM")).thenReturn("dbo", "dbo");
when(columnsRs.getString("COLUMN_NAME")).thenReturn("id", "bad");
+ // doReadTableColumn reads column name and type by JDBC positional
index, not by label
+ when(columnsRs.getString(4)).thenReturn("id", "bad"); // COLUMN_NAME
+ when(columnsRs.getString(6)).thenReturn("INT", "INT"); // TYPE_NAME
DatabaseMetaData metadata = mock(DatabaseMetaData.class);
when(metadata.getColumns(eq(databaseName), eq("dbo"), eq("user_info"),
isNull()))
.thenReturn(columnsRs);
+ ResultSet emptyRs = mock(ResultSet.class);
+ when(emptyRs.next()).thenReturn(false);
+ PreparedStatement mockPs = mock(PreparedStatement.class);
+ when(mockPs.executeQuery()).thenReturn(emptyRs);
+
Connection jdbcConnection = mock(Connection.class);
when(jdbcConnection.getMetaData()).thenReturn(metadata);
+ when(jdbcConnection.prepareStatement(anyString())).thenReturn(mockPs);
TestSqlServerConnection connection = new
TestSqlServerConnection(jdbcConnection);
Table table = connection.getTableSchemaFromTable(databaseName,
changeTable);
@@ -93,8 +104,14 @@ public class SqlServerConnectionTest {
when(metadata.getColumns(eq(databaseName), eq("dbo"), eq("UserInfo"),
isNull()))
.thenReturn(columnsRs);
+ ResultSet emptyRs = mock(ResultSet.class);
+ when(emptyRs.next()).thenReturn(false);
+ PreparedStatement mockPs = mock(PreparedStatement.class);
+ when(mockPs.executeQuery()).thenReturn(emptyRs);
+
Connection jdbcConnection = mock(Connection.class);
when(jdbcConnection.getMetaData()).thenReturn(metadata);
+ when(jdbcConnection.prepareStatement(anyString())).thenReturn(mockPs);
TestSqlServerConnection connection = new
TestSqlServerConnection(jdbcConnection);
Table table = connection.getTableSchemaFromTable(databaseName,
changeTable);
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactoryTest.java
new file mode 100644
index 0000000000..f4f4ad604b
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/config/SqlServerSourceConfigFactoryTest.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.config;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class SqlServerSourceConfigFactoryTest {
+
+ @Test
+ void shouldEnableSchemaChangesInDebeziumConfigWhenSchemaEvolutionEnabled()
{
+ SqlServerSourceConfigFactory configFactory = createFactory();
+ configFactory.schemaChangeEnabled(true);
+
+ SqlServerSourceConfig sourceConfig = configFactory.create(0);
+
+ Assertions.assertTrue(
+ sourceConfig
+ .getDbzConfiguration()
+
.getBoolean(SqlServerSourceConfigFactory.SCHEMA_CHANGE_KEY));
+ }
+
+ private SqlServerSourceConfigFactory createFactory() {
+ return (SqlServerSourceConfigFactory)
+ new SqlServerSourceConfigFactory()
+ .hostname("localhost")
+ .port(1433)
+ .username("sa")
+ .password("Password!")
+ .databaseList("schema_change_test");
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerSchemaChangeResolverTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerSchemaChangeResolverTest.java
new file mode 100644
index 0000000000..b5d7be26cb
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerSchemaChangeResolverTest.java
@@ -0,0 +1,299 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source;
+
+import org.apache.seatunnel.api.event.EventType;
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+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.AlterTableColumnsEvent;
+import org.apache.seatunnel.api.table.schema.event.AlterTableDropColumnEvent;
+import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.type.BasicType;
+import
org.apache.seatunnel.connectors.cdc.debezium.ConnectTableChangeSerializer;
+
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import io.debezium.connector.AbstractSourceInfo;
+import io.debezium.data.Envelope;
+import io.debezium.relational.Table;
+import io.debezium.relational.history.HistoryRecord;
+import io.debezium.relational.history.TableChanges;
+
+import java.sql.Types;
+import java.util.Arrays;
+import java.util.Collections;
+
+public class SqlServerSchemaChangeResolverTest {
+
+ private static final String DATABASE_NAME = "test_db";
+ private static final String SCHEMA_NAME = "dbo";
+ private static final String TABLE_NAME = "customers";
+ private static final TableIdentifier TABLE_IDENTIFIER =
+ TableIdentifier.of(null, DATABASE_NAME, SCHEMA_NAME, TABLE_NAME);
+ private static final Schema KEY_SCHEMA =
+
SchemaBuilder.struct().name("io.debezium.connector.sqlserver.SchemaChangeKey").build();
+ private static final Schema SOURCE_SCHEMA =
+ SchemaBuilder.struct()
+ .field(AbstractSourceInfo.DATABASE_NAME_KEY,
Schema.STRING_SCHEMA)
+ .field(AbstractSourceInfo.SCHEMA_NAME_KEY,
Schema.STRING_SCHEMA)
+ .field(AbstractSourceInfo.TABLE_NAME_KEY,
Schema.STRING_SCHEMA)
+ .build();
+ private static final Schema VALUE_SCHEMA =
+ SchemaBuilder.struct()
+ .field(Envelope.FieldName.SOURCE, SOURCE_SCHEMA)
+ .field(HistoryRecord.Fields.DDL_STATEMENTS,
Schema.OPTIONAL_STRING_SCHEMA)
+ .field(
+ HistoryRecord.Fields.TABLE_CHANGES,
+
SchemaBuilder.array(ConnectTableChangeSerializer.CHANGE_SCHEMA).build())
+ .build();
+
+ private final SqlServerSchemaChangeResolver resolver = new
SqlServerSchemaChangeResolver();
+ private final ConnectTableChangeSerializer serializer = new
ConnectTableChangeSerializer();
+
+ @Test
+ void shouldRecognizeSqlServerSchemaChangeRecord() {
+ SourceRecord record =
createRecord(createAlterTableChange(varcharColumn("email", 3, 128)));
+
+ Assertions.assertTrue(resolver.support(record));
+ }
+
+ @Test
+ void shouldResolveAddColumnEventFromTableChanges() {
+ SourceRecord record =
createRecord(createAlterTableChange(varcharColumn("email", 3, 128)));
+
+ SchemaChangeEvent event =
+ resolver.resolve(record,
Collections.singletonList(createCatalogTable()));
+
+ Assertions.assertInstanceOf(AlterTableColumnsEvent.class, event);
+ AlterTableColumnsEvent columnsEvent = (AlterTableColumnsEvent) event;
+ Assertions.assertEquals(
+ EventType.SCHEMA_CHANGE_UPDATE_COLUMNS,
columnsEvent.getEventType());
+ Assertions.assertEquals(1, columnsEvent.getEvents().size());
+ Assertions.assertInstanceOf(
+ AlterTableAddColumnEvent.class,
columnsEvent.getEvents().get(0));
+ AlterTableAddColumnEvent addColumnEvent =
+ (AlterTableAddColumnEvent) columnsEvent.getEvents().get(0);
+ Assertions.assertEquals("email", addColumnEvent.getColumn().getName());
+ Assertions.assertEquals("name", addColumnEvent.getAfterColumn());
+ }
+
+ @Test
+ void shouldResolveRenameColumnEventFromTableChanges() {
+ TableChanges tableChanges = new TableChanges();
+ tableChanges.alter(
+ Table.editor()
+ .tableId(
+ new io.debezium.relational.TableId(
+ DATABASE_NAME, SCHEMA_NAME,
TABLE_NAME))
+ .setPrimaryKeyNames(Collections.singletonList("id"))
+ .setColumns(
+ Arrays.asList(
+ intColumn("id", 1),
varcharColumn("full_name", 2, 64)))
+ .create());
+ SourceRecord record =
+ createRecord(
+ tableChanges,
+ DATABASE_NAME,
+ SCHEMA_NAME,
+ TABLE_NAME,
+ "EXEC sp_rename 'dbo.customers.name', 'full_name',
'COLUMN'");
+
+ SchemaChangeEvent event =
+ resolver.resolve(record,
Collections.singletonList(createCatalogTable()));
+
+ Assertions.assertInstanceOf(AlterTableColumnsEvent.class, event);
+ AlterTableColumnsEvent columnsEvent = (AlterTableColumnsEvent) event;
+ Assertions.assertEquals(1, columnsEvent.getEvents().size());
+ Assertions.assertInstanceOf(
+ AlterTableChangeColumnEvent.class,
columnsEvent.getEvents().get(0));
+ AlterTableChangeColumnEvent changeEvent =
+ (AlterTableChangeColumnEvent) columnsEvent.getEvents().get(0);
+ Assertions.assertEquals("name", changeEvent.getOldColumn());
+ Assertions.assertEquals("full_name",
changeEvent.getColumn().getName());
+ Assertions.assertEquals("id", changeEvent.getAfterColumn());
+ }
+
+ @Test
+ void shouldEmitDropAndAddWhenDdlIsNotExplicitRename() {
+ TableChanges tableChanges = new TableChanges();
+ tableChanges.alter(
+ Table.editor()
+ .tableId(
+ new io.debezium.relational.TableId(
+ DATABASE_NAME, SCHEMA_NAME,
TABLE_NAME))
+ .setPrimaryKeyNames(Collections.singletonList("id"))
+ .setColumns(
+ Arrays.asList(
+ intColumn("id", 1),
varcharColumn("full_name", 2, 64)))
+ .create());
+ SourceRecord record =
+ createRecord(
+ tableChanges,
+ DATABASE_NAME,
+ SCHEMA_NAME,
+ TABLE_NAME,
+ "ALTER TABLE dbo.customers DROP COLUMN name; ALTER
TABLE dbo.customers ADD full_name VARCHAR(64)");
+
+ SchemaChangeEvent event =
+ resolver.resolve(record,
Collections.singletonList(createCatalogTable()));
+
+ Assertions.assertInstanceOf(AlterTableColumnsEvent.class, event);
+ AlterTableColumnsEvent columnsEvent = (AlterTableColumnsEvent) event;
+ Assertions.assertEquals(2, columnsEvent.getEvents().size());
+ Assertions.assertInstanceOf(
+ AlterTableAddColumnEvent.class,
columnsEvent.getEvents().get(0));
+ Assertions.assertInstanceOf(
+ AlterTableDropColumnEvent.class,
columnsEvent.getEvents().get(1));
+
+ AlterTableAddColumnEvent addEvent =
+ (AlterTableAddColumnEvent) columnsEvent.getEvents().get(0);
+ Assertions.assertEquals("full_name", addEvent.getColumn().getName());
+ Assertions.assertEquals("id", addEvent.getAfterColumn());
+
+ AlterTableDropColumnEvent dropEvent =
+ (AlterTableDropColumnEvent) columnsEvent.getEvents().get(1);
+ Assertions.assertEquals("name", dropEvent.getColumn());
+ }
+
+ @Test
+ void shouldResolveSchemaChangeWhenIdentifiersAreBracketed() {
+ SourceRecord record =
+ createRecord(
+ createAlterTableChange(varcharColumn("email", 3, 128)),
+ "[test_db]",
+ "[dbo]",
+ "[customers]",
+ "ALTER TABLE [test_db].[dbo].[customers] ADD [email]
VARCHAR(128)");
+
+ SchemaChangeEvent event =
+ resolver.resolve(record,
Collections.singletonList(createCatalogTable()));
+
+ Assertions.assertInstanceOf(AlterTableColumnsEvent.class, event);
+ AlterTableColumnsEvent columnsEvent = (AlterTableColumnsEvent) event;
+ Assertions.assertEquals(1, columnsEvent.getEvents().size());
+ Assertions.assertInstanceOf(
+ AlterTableAddColumnEvent.class,
columnsEvent.getEvents().get(0));
+ }
+
+ private CatalogTable createCatalogTable() {
+ return CatalogTable.of(
+ TABLE_IDENTIFIER,
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.builder()
+ .name("id")
+ .dataType(BasicType.INT_TYPE)
+ .nullable(false)
+ .columnLength(10L)
+ .sourceType("int")
+ .build())
+ .column(
+ PhysicalColumn.builder()
+ .name("name")
+ .dataType(BasicType.STRING_TYPE)
+ .nullable(true)
+ .columnLength(64L)
+ .sourceType("varchar(64)")
+ .build())
+ .build(),
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ null,
+ null);
+ }
+
+ private TableChanges createAlterTableChange(io.debezium.relational.Column
newColumn) {
+ TableChanges tableChanges = new TableChanges();
+ tableChanges.alter(
+ Table.editor()
+ .tableId(
+ new io.debezium.relational.TableId(
+ DATABASE_NAME, SCHEMA_NAME,
TABLE_NAME))
+ .setPrimaryKeyNames(Collections.singletonList("id"))
+ .setColumns(
+ Arrays.asList(
+ intColumn("id", 1),
+ varcharColumn("name", 2, 64),
+ newColumn))
+ .create());
+ return tableChanges;
+ }
+
+ private SourceRecord createRecord(TableChanges tableChanges) {
+ return createRecord(tableChanges, DATABASE_NAME, SCHEMA_NAME,
TABLE_NAME, "ALTER TABLE");
+ }
+
+ private SourceRecord createRecord(
+ TableChanges tableChanges,
+ String databaseName,
+ String schemaName,
+ String tableName,
+ String ddl) {
+ Struct value = new Struct(VALUE_SCHEMA);
+ value.put(
+ Envelope.FieldName.SOURCE,
+ new Struct(SOURCE_SCHEMA)
+ .put(AbstractSourceInfo.DATABASE_NAME_KEY,
databaseName)
+ .put(AbstractSourceInfo.SCHEMA_NAME_KEY, schemaName)
+ .put(AbstractSourceInfo.TABLE_NAME_KEY, tableName));
+ value.put(HistoryRecord.Fields.DDL_STATEMENTS, ddl);
+ value.put(HistoryRecord.Fields.TABLE_CHANGES,
serializer.serialize(tableChanges));
+ return new SourceRecord(
+ Collections.emptyMap(),
+ Collections.emptyMap(),
+ "topic",
+ null,
+ KEY_SCHEMA,
+ new Struct(KEY_SCHEMA),
+ VALUE_SCHEMA,
+ value);
+ }
+
+ private io.debezium.relational.Column intColumn(String name, int position)
{
+ return io.debezium.relational.Column.editor()
+ .name(name)
+ .jdbcType(Types.INTEGER)
+ .nativeType(Types.INTEGER)
+ .type("int", "int")
+ .position(position)
+ .optional(false)
+ .create();
+ }
+
+ private io.debezium.relational.Column varcharColumn(String name, int
position, int length) {
+ return io.debezium.relational.Column.editor()
+ .name(name)
+ .jdbcType(Types.VARCHAR)
+ .nativeType(Types.VARCHAR)
+ .type("varchar", "varchar(" + length + ")")
+ .length(length)
+ .position(position)
+ .optional(true)
+ .create();
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialect.java
index 7581f52f24..6645ea424c 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialect.java
@@ -292,7 +292,7 @@ public class SqlServerDialect implements JdbcDialect {
String sourceDialectName = event.getSourceDialectName();
boolean sameCatalog = StringUtils.equals(dialectName(),
sourceDialectName);
BasicTypeDefine typeDefine = getTypeConverter().reconvert(column);
- String columnType = sameCatalog ? column.getSourceType() :
typeDefine.getColumnType();
+ String columnType = resolveColumnType(sameCatalog, column, typeDefine);
// Build the SQL statement that add the column
StringBuilder sqlBuilder =
@@ -310,8 +310,22 @@ public class SqlServerDialect implements JdbcDialect {
}
if (!column.isNullable()) {
- // Handle null constraints
- sqlBuilder.append(" NOT NULL");
+ if (column.getDefaultValue() != null) {
+ // A DEFAULT is present — SQL Server can populate existing
rows, so NOT NULL is
+ // safe.
+ sqlBuilder.append(" NOT NULL");
+ } else {
+ // SQL Server forbids adding a NOT NULL column without a
DEFAULT to a non-empty
+ // table.
+ // Add as NULL so that existing rows are not affected;
subsequent CDC UPDATE events
+ // will fill in the actual values for those rows.
+ log.warn(
+ "Column '{}' in table {} is NOT NULL but has no
DEFAULT; adding as NULL "
+ + "to allow addition to a non-empty table.",
+ column.getName(),
+ tablePath.getFullName());
+ sqlBuilder.append(" NULL");
+ }
}
ddlSQL.add(sqlBuilder.toString());
@@ -331,16 +345,13 @@ public class SqlServerDialect implements JdbcDialect {
List<String> ddlSQL = new ArrayList<>();
if (event.getOldColumn() != null
&&
!(event.getColumn().getName().equals(event.getOldColumn()))) {
+ String renameObject = buildRenameColumnObject(tablePath,
event.getOldColumn());
StringBuilder sqlBuilder =
new StringBuilder()
- .append("EXEC sp_rename ")
- .append(
- String.format(
- "'%s.%s.%s.%s', ",
- tablePath.getDatabaseName(),
- tablePath.getSchemaName(),
- tablePath.getTableName(),
- event.getOldColumn()))
+ .append("EXEC ")
+ .append(buildRenameProcedure(tablePath))
+ .append(" ")
+ .append(String.format("'%s', ", renameObject))
.append(String.format("'%s', 'COLUMN';",
event.getColumn().getName()));
ddlSQL.add(sqlBuilder.toString());
}
@@ -363,7 +374,7 @@ public class SqlServerDialect implements JdbcDialect {
String sourceDialectName = event.getSourceDialectName();
boolean sameCatalog = StringUtils.equals(dialectName(),
sourceDialectName);
BasicTypeDefine typeDefine = getTypeConverter().reconvert(column);
- String columnType = sameCatalog ? column.getSourceType() :
typeDefine.getColumnType();
+ String columnType = resolveColumnType(sameCatalog, column, typeDefine);
List<String> ddlSQL = new ArrayList<>();
// Handle field default constraints.
if (column.getDefaultValue() != null) {
@@ -521,17 +532,57 @@ public class SqlServerDialect implements JdbcDialect {
private boolean columnIsNullable(Connection connection, TablePath
tablePath, String column)
throws SQLException {
+ // Prefix with the target database name so the query works even when
the JDBC connection
+ // is connected to a different database (e.g. master with no
databaseName in the URL).
+ String databaseName = tablePath.getDatabaseName();
+ String infoSchemaPrefix =
+ StringUtils.isNotBlank(databaseName)
+ ? quoteDatabaseIdentifier(databaseName) +
".INFORMATION_SCHEMA"
+ : "INFORMATION_SCHEMA";
String selectColumnSQL =
String.format(
- "SELECT IS_NULLABLE FROM information_schema.COLUMNS
WHERE %s AND COLUMN_NAME = '%s';",
- buildCommonWhereClause(tablePath), column);
+ "SELECT IS_NULLABLE FROM %s.COLUMNS WHERE %s AND
COLUMN_NAME = '%s';",
+ infoSchemaPrefix, buildCommonWhereClause(tablePath),
column);
try (Statement statement = connection.createStatement()) {
ResultSet rs = statement.executeQuery(selectColumnSQL);
- rs.next();
+ if (!rs.next()) {
+ // Column not found — default to non-nullable to avoid
incorrectly appending
+ // NULL to an ALTER COLUMN statement for a column that doesn't
allow nulls.
+ log.warn(
+ "Column '{}' not found in {}.COLUMNS for table {};
assuming NOT NULL",
+ column,
+ infoSchemaPrefix,
+ tablePath.getFullName());
+ return false;
+ }
return rs.getString("IS_NULLABLE").equals("YES");
}
}
+ /**
+ * Returns the SQL column type string to use in DDL statements.
+ *
+ * <p>When source and sink are the same catalog (SQL Server to SQL Server)
we prefer the
+ * original {@code sourceType} because it already carries the full type
expression (e.g. {@code
+ * varchar(255)}). However, when the CDC schema-change path produces a
column whose {@code
+ * sourceType} is a bare type name without length/precision (e.g. {@code
varchar} instead of
+ * {@code varchar(255)}), SQL Server rejects the resulting DDL statement.
In that case we fall
+ * back to {@code typeDefine.getColumnType()}, which is always fully
qualified.
+ */
+ private String resolveColumnType(
+ boolean sameCatalog, Column column, BasicTypeDefine typeDefine) {
+ if (!sameCatalog) {
+ return typeDefine.getColumnType();
+ }
+ String sourceType = column.getSourceType();
+ if (StringUtils.isBlank(sourceType) || !sourceType.contains("(")) {
+ // Bare type name (no length/precision); use the fully-qualified
type from
+ // the type converter to avoid SQL Server DDL syntax errors.
+ return typeDefine.getColumnType();
+ }
+ return sourceType;
+ }
+
private StringBuilder buildAlterTablePrefix(TablePath tablePath) {
return new StringBuilder("ALTER TABLE
").append(tableIdentifier(tablePath));
}
@@ -541,4 +592,21 @@ public class SqlServerDialect implements JdbcDialect {
"TABLE_CATALOG = '%s' AND TABLE_SCHEMA = '%s' AND TABLE_NAME =
'%s'",
tablePath.getDatabaseName(), tablePath.getSchemaName(),
tablePath.getTableName());
}
+
+ private String buildRenameColumnObject(TablePath tablePath, String
oldColumn) {
+ List<String> objectParts = new ArrayList<>();
+ if (StringUtils.isNotBlank(tablePath.getSchemaName())) {
+ objectParts.add(tablePath.getSchemaName());
+ }
+ objectParts.add(tablePath.getTableName());
+ objectParts.add(oldColumn);
+ return String.join(".", objectParts);
+ }
+
+ private String buildRenameProcedure(TablePath tablePath) {
+ if (StringUtils.isNotBlank(tablePath.getDatabaseName())) {
+ return quoteDatabaseIdentifier(tablePath.getDatabaseName()) +
".sys.sp_rename";
+ }
+ return "sp_rename";
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialectTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialectTest.java
new file mode 100644
index 0000000000..f10d828652
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialectTest.java
@@ -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.
+ */
+
+package
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.sqlserver;
+
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.schema.event.AlterTableChangeColumnEvent;
+
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.Statement;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class SqlServerDialectTest {
+
+ @Test
+ void shouldUseSchemaTableColumnForRenameWithoutDatabasePrefix() throws
Exception {
+ SqlServerDialect dialect = new SqlServerDialect();
+ Connection connection = mock(Connection.class);
+ Statement statement = mock(Statement.class);
+ when(connection.createStatement()).thenReturn(statement);
+
+ Column newColumn = mock(Column.class);
+ when(newColumn.getName()).thenReturn("add_column");
+ when(newColumn.getDataType()).thenReturn(null);
+
+ AlterTableChangeColumnEvent event =
+ AlterTableChangeColumnEvent.change(
+ TableIdentifier.of("SqlServer", "schema_change_test",
"dbo", "products"),
+ "add_column2",
+ newColumn);
+
+ dialect.applySchemaChange(
+ connection, TablePath.of("schema_change_test", "dbo",
"products_sink"), event);
+
+ verify(statement)
+ .execute(
+ "EXEC [schema_change_test].sys.sp_rename
'dbo.products_sink.add_column2', 'add_column', 'COLUMN';");
+ }
+}
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 b70f60843b..8df8f19a8f 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
@@ -88,6 +88,10 @@ public class SqlServerCDCIT extends TestSuiteBase implements
TestResource {
public static final String DATABASE_NAME = "column_type_test";
public static final String SCHEMA_NAME = "dbo";
+ public static final String SCHEMA_EVOLUTION_DATABASE_NAME =
"schema_change_test";
+ private static final String SCHEMA_EVOLUTION_SOURCE_TABLE = "products";
+ private static final String SCHEMA_EVOLUTION_SINK_TABLE = "products_sink";
+ private static final int INCREMENTAL_MARKER_ID = 1000;
private static final String DISABLE_DB_CDC =
"IF EXISTS(select 1 from sys.databases where name='#' AND
is_cdc_enabled=1)\n"
@@ -161,6 +165,16 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
+ " CONVERT(varchar(100), val_varbinary) as
val_varbinary,\n"
+ " val_udtdecimal\n"
+ "from %s order by id asc";
+ private static final String SELECT_SCHEMA_EVOLUTION_DATA_SQL =
+ "SELECT * FROM %s ORDER BY id ASC";
+ private static final String SELECT_SCHEMA_EVOLUTION_COLUMNS_SQL =
+ "SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, "
+ + "COALESCE(CAST(CHARACTER_MAXIMUM_LENGTH AS VARCHAR(20)),
'null'), "
+ + "COALESCE(CAST(NUMERIC_PRECISION AS VARCHAR(20)),
'null'), "
+ + "COALESCE(CAST(NUMERIC_SCALE AS VARCHAR(20)), 'null') "
+ + "FROM INFORMATION_SCHEMA.COLUMNS "
+ + "WHERE TABLE_CATALOG = '%s' AND TABLE_SCHEMA = '%s' AND
TABLE_NAME = '%s' "
+ + "ORDER BY ORDINAL_POSITION";
public static final MSSQLServerContainer MSSQL_SERVER_CONTAINER =
new
MSSQLServerContainer<>("mcr.microsoft.com/mssql/server:2019-latest")
@@ -505,6 +519,50 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
* Executes a JDBC statement using the default jdbc config without
autocommitting the
* connection.
*/
+ private void assertSchemaEvolutionTableStructureAndData(
+ String databaseName, String sourceTable, String sinkTable) {
+ String sourceTablePath = databaseName + "." + SCHEMA_NAME + "." +
sourceTable;
+ String sinkTablePath = databaseName + "." + SCHEMA_NAME + "." +
sinkTable;
+ await().atMost(120000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertIterableEquals(
+
querySql(SELECT_SCHEMA_EVOLUTION_DATA_SQL, sourceTablePath),
+
querySql(SELECT_SCHEMA_EVOLUTION_DATA_SQL, sinkTablePath)));
+ await().atMost(120000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertIterableEquals(
+ querySql(
+ String.format(
+
SELECT_SCHEMA_EVOLUTION_COLUMNS_SQL,
+ databaseName,
+ SCHEMA_NAME,
+ sourceTable)),
+ querySql(
+ String.format(
+
SELECT_SCHEMA_EVOLUTION_COLUMNS_SQL,
+ databaseName,
+ SCHEMA_NAME,
+ sinkTable))));
+ }
+
+ private void executeSqlFile(String sqlFile) {
+ final String ddlFile = String.format("ddl/%s.sql", sqlFile);
+ final URL ddlTestFile =
TestSuiteBase.class.getClassLoader().getResource(ddlFile);
+ Assertions.assertNotNull(ddlTestFile, "Cannot locate " + ddlFile);
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ List<String> statements =
+
parseStatements(Files.readAllLines(Paths.get(ddlTestFile.toURI())));
+ for (String stmt : statements) {
+ statement.execute(stmt);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
private void initializeSqlServerTable(String sqlFile) {
final String ddlFile = String.format("ddl/%s.sql", sqlFile);
final URL ddlTestFile =
TestSuiteBase.class.getClassLoader().getResource(ddlFile);
@@ -515,20 +573,7 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
String ddlContent = String.join("\n", ddlLines);
String actualDatabaseName = extractDatabaseName(ddlContent);
dropTestDatabase(connection, actualDatabaseName);
- final List<String> statements =
- Arrays.stream(
- ddlLines.stream()
- .map(String::trim)
- .filter(x -> !x.startsWith("--")
&& !x.isEmpty())
- .map(
- x -> {
- final Matcher m =
-
COMMENT_PATTERN.matcher(x);
- return m.matches() ?
m.group(1) : x;
- })
- .collect(Collectors.joining("\n"))
- .split(";"))
- .collect(Collectors.toList());
+ final List<String> statements = parseStatements(ddlLines);
for (String stmt : statements) {
statement.execute(stmt);
}
@@ -537,6 +582,23 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
}
}
+ private List<String> parseStatements(List<String> ddlLines) {
+ return Arrays.stream(
+ ddlLines.stream()
+ .map(String::trim)
+ .filter(x -> !x.startsWith("--") &&
!x.isEmpty())
+ .map(
+ x -> {
+ final Matcher m =
COMMENT_PATTERN.matcher(x);
+ return m.matches() ? m.group(1) :
x;
+ })
+ .collect(Collectors.joining("\n"))
+ .split(";"))
+ .map(String::trim)
+ .filter(x -> !x.isEmpty())
+ .collect(Collectors.toList());
+ }
+
private String extractDatabaseName(String ddlContent) {
Pattern createDbPattern =
Pattern.compile(
@@ -597,8 +659,9 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
}
private void executeSql(String sql) {
- try (Connection connection = getJdbcConnection()) {
- connection.createStatement().execute(sql);
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(sql);
} catch (SQLException e) {
throw new RuntimeException(e);
}
@@ -719,6 +782,94 @@ public class SqlServerCDCIT extends TestSuiteBase
implements TestResource {
});
}
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK},
+ disabledReason =
+ "This case validates SqlServer CDC schema evolution on the
Flink engine & zeta engine.")
+ public void testWithSchemaEvolution(TestContainer container) {
+ initializeSqlServerTable("schema_change_test");
+
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+
container.executeJob("/sqlservercdc_to_sqlserver_with_schema_change.conf");
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ return null;
+ });
+
+ assertSchemaEvolutionTableStructureAndData(
+ SCHEMA_EVOLUTION_DATABASE_NAME,
+ SCHEMA_EVOLUTION_SOURCE_TABLE,
+ SCHEMA_EVOLUTION_SINK_TABLE);
+
+ waitForSchemaEvolutionIncrementalStarted();
+
+ executeSqlFile("sqlserver_schema_change_add_columns");
+ assertSchemaEvolutionTableStructureAndData(
+ SCHEMA_EVOLUTION_DATABASE_NAME,
+ SCHEMA_EVOLUTION_SOURCE_TABLE,
+ SCHEMA_EVOLUTION_SINK_TABLE);
+
+ executeSqlFile("sqlserver_schema_change_drop_columns");
+ assertSchemaEvolutionTableStructureAndData(
+ SCHEMA_EVOLUTION_DATABASE_NAME,
+ SCHEMA_EVOLUTION_SOURCE_TABLE,
+ SCHEMA_EVOLUTION_SINK_TABLE);
+
+ executeSqlFile("sqlserver_schema_change_rename_columns");
+ assertSchemaEvolutionTableStructureAndData(
+ SCHEMA_EVOLUTION_DATABASE_NAME,
+ SCHEMA_EVOLUTION_SOURCE_TABLE,
+ SCHEMA_EVOLUTION_SINK_TABLE);
+
+ executeSqlFile("sqlserver_schema_change_modify_columns");
+ assertSchemaEvolutionTableStructureAndData(
+ SCHEMA_EVOLUTION_DATABASE_NAME,
+ SCHEMA_EVOLUTION_SOURCE_TABLE,
+ SCHEMA_EVOLUTION_SINK_TABLE);
+ }
+
+ private void waitForSchemaEvolutionIncrementalStarted() {
+ String sourceTablePath =
+ SCHEMA_EVOLUTION_DATABASE_NAME
+ + "."
+ + SCHEMA_NAME
+ + "."
+ + SCHEMA_EVOLUTION_SOURCE_TABLE;
+ String sinkTablePath =
+ SCHEMA_EVOLUTION_DATABASE_NAME
+ + "."
+ + SCHEMA_NAME
+ + "."
+ + SCHEMA_EVOLUTION_SINK_TABLE;
+ executeSql(
+ String.format(
+ "INSERT INTO %s VALUES (%d, 'incremental-marker',
'ensure-stream-phase', 9.9)",
+ sourceTablePath, INCREMENTAL_MARKER_ID));
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ querySql(
+ String.format(
+ "SELECT
COUNT(1) FROM %s WHERE id = %d",
+
sourceTablePath,
+
INCREMENTAL_MARKER_ID))
+ .get(0)
+ .get(0),
+ querySql(
+ String.format(
+ "SELECT
COUNT(1) FROM %s WHERE id = %d",
+ sinkTablePath,
+
INCREMENTAL_MARKER_ID))
+ .get(0)
+ .get(0)));
+ }
+
@TestTemplate
@DisabledOnContainer(
value = {},
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/schema_change_test.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/schema_change_test.sql
new file mode 100644
index 0000000000..dfbe560d18
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/schema_change_test.sql
@@ -0,0 +1,47 @@
+--
+-- 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.
+
+CREATE DATABASE schema_change_test;
+USE schema_change_test;
+EXEC sys.sp_cdc_enable_db;
+
+CREATE TABLE dbo.products (
+ id INT NOT NULL,
+ name VARCHAR(255) NOT NULL,
+ description VARCHAR(512),
+ weight FLOAT,
+ PRIMARY KEY (id)
+);
+
+CREATE TABLE dbo.products_sink (
+ id INT NOT NULL,
+ name VARCHAR(255) NOT NULL,
+ description VARCHAR(512),
+ weight FLOAT,
+ PRIMARY KEY (id)
+);
+
+INSERT INTO dbo.products VALUES (101, 'scooter', 'Small 2-wheel scooter',
3.14);
+INSERT INTO dbo.products VALUES (102, 'car battery', '12V car battery', 8.1);
+INSERT INTO dbo.products VALUES (103, '12-pack drill bits', '12-pack of drill
bits with sizes ranging from #40 to #3', 0.8);
+INSERT INTO dbo.products VALUES (104, 'hammer', '12oz carpenter''s hammer',
0.75);
+INSERT INTO dbo.products VALUES (105, 'hammer', '14oz carpenter''s hammer',
0.875);
+INSERT INTO dbo.products VALUES (106, 'hammer', '16oz carpenter''s hammer',
1.0);
+INSERT INTO dbo.products VALUES (107, 'rocks', 'box of assorted rocks', 5.3);
+INSERT INTO dbo.products VALUES (108, 'jacket', 'water resistant black wind
breaker', 0.1);
+INSERT INTO dbo.products VALUES (109, 'spare tire', '24 inch spare tire',
22.2);
+
+EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name =
'products', @role_name = NULL, @supports_net_changes = 0;
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_add_columns.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_add_columns.sql
new file mode 100644
index 0000000000..628c6e444d
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_add_columns.sql
@@ -0,0 +1,55 @@
+--
+-- 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.
+
+USE schema_change_test;
+
+INSERT INTO dbo.products VALUES (110, 'scooter', 'Small 2-wheel scooter',
3.14);
+UPDATE dbo.products SET name = 'updated-before-add' WHERE id = 101;
+DELETE FROM dbo.products WHERE id = 102;
+
+-- Add columns as NULL first so that existing rows are not silently filled by
SQL Server's
+-- DEFAULT mechanism (which does not generate CDC events). We will explicitly
set values
+-- for all pre-existing rows via UPDATE so that CDC can replicate them to the
sink.
+ALTER TABLE dbo.products
+ADD add_column1 VARCHAR(64) NULL,
+ add_column2 INT NULL;
+
+INSERT INTO dbo.products VALUES (120, 'car battery', '12V car battery', 8.1,
'xx', 2);
+INSERT INTO dbo.products VALUES (121, 'drill bits', 'sizes from #40 to #3',
0.8, 'xx', 3);
+
+ALTER TABLE dbo.products
+ADD add_column3 FLOAT NULL;
+
+ALTER TABLE dbo.products
+ADD add_column4 DATETIME2 NULL;
+
+EXEC sys.sp_cdc_enable_table
+ @source_schema = 'dbo',
+ @source_name = 'products',
+ @role_name = NULL,
+ @capture_instance = 'dbo_products_v2',
+ @supports_net_changes = 0;
+
+-- Backfill the newly added columns only after the new capture instance is
enabled,
+-- otherwise the source table is updated but the sink never receives the
non-null values.
+UPDATE dbo.products SET add_column1 = 'yy', add_column2 = 1
+WHERE id IN (101, 103, 104, 105, 106, 107, 108, 109, 110);
+
+UPDATE dbo.products SET add_column3 = 1.1, add_column4 = '2023-02-02T09:09:09'
+WHERE id IN (101, 103, 104, 105, 106, 107, 108, 109, 110, 120, 121);
+
+INSERT INTO dbo.products VALUES (128, 'scooter', 'Small 2-wheel scooter',
3.14, 'xx', 1, 1.1, '2023-02-02T09:09:09');
+INSERT INTO dbo.products VALUES (129, 'car battery', '12V car battery', 8.1,
'xx', 2, 1.2, '2023-02-02T09:09:09');
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_drop_columns.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_drop_columns.sql
new file mode 100644
index 0000000000..644c004c97
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_drop_columns.sql
@@ -0,0 +1,47 @@
+--
+-- 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.
+
+USE schema_change_test;
+
+EXEC sys.sp_cdc_disable_table
+ @source_schema = 'dbo',
+ @source_name = 'products',
+ @capture_instance = 'dbo_products_v2';
+
+ALTER TABLE dbo.products DROP COLUMN add_column4;
+
+INSERT INTO dbo.products VALUES (130, 'scooter', 'Small 2-wheel scooter',
3.14, 'xx', 1, 1.1);
+INSERT INTO dbo.products VALUES (131, 'car battery', '12V car battery', 8.1,
'xx', 2, 1.2);
+
+ALTER TABLE dbo.products DROP COLUMN add_column1, add_column3;
+
+EXEC sys.sp_cdc_enable_table
+ @source_schema = 'dbo',
+ @source_name = 'products',
+ @role_name = NULL,
+ @capture_instance = 'dbo_products_v3',
+ @supports_net_changes = 0;
+
+-- Rows inserted between capture-instance switches may be emitted by an older
+-- instance without add_column2. Touch them again after v3 is enabled so CDC
+-- emits the current schema and sink rows converge.
+UPDATE dbo.products
+SET add_column2 = CASE id WHEN 130 THEN 11 WHEN 131 THEN 12 END
+WHERE id IN (130, 131);
+
+INSERT INTO dbo.products VALUES (140, 'scooter', 'Small 2-wheel scooter',
3.14, 1);
+INSERT INTO dbo.products VALUES (141, 'car battery', '12V car battery', 8.1,
2);
+UPDATE dbo.products SET name = 'updated-after-drop' WHERE id = 140;
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_modify_columns.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_modify_columns.sql
new file mode 100644
index 0000000000..336d8526e8
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_modify_columns.sql
@@ -0,0 +1,36 @@
+--
+-- 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.
+
+USE schema_change_test;
+
+EXEC sys.sp_cdc_disable_table
+ @source_schema = 'dbo',
+ @source_name = 'products',
+ @capture_instance = 'dbo_products_v4';
+
+-- Use bounded length for stable source/sink metadata comparison in IT
assertions.
+ALTER TABLE dbo.products ALTER COLUMN name NVARCHAR(255) NULL;
+
+EXEC sys.sp_cdc_enable_table
+ @source_schema = 'dbo',
+ @source_name = 'products',
+ @role_name = NULL,
+ @capture_instance = 'dbo_products_v5',
+ @supports_net_changes = 0;
+
+UPDATE dbo.products SET name = N'updated-after-modify' WHERE id = 150;
+INSERT INTO dbo.products VALUES (160, N'scooter-modified', 'Small 2-wheel
scooter', 3.14, 1);
+INSERT INTO dbo.products VALUES (161, N'car battery-modified', '12V car
battery', 8.1, 2);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_rename_columns.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_rename_columns.sql
new file mode 100644
index 0000000000..dee84b1abe
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/ddl/sqlserver_schema_change_rename_columns.sql
@@ -0,0 +1,48 @@
+--
+-- 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.
+
+USE schema_change_test;
+
+EXEC sys.sp_cdc_disable_table
+ @source_schema = 'dbo',
+ @source_name = 'products',
+ @capture_instance = 'dbo_products_v3';
+
+EXEC sp_rename 'dbo.products.add_column2', 'add_column', 'COLUMN';
+
+EXEC sys.sp_cdc_enable_table
+ @source_schema = 'dbo',
+ @source_name = 'products',
+ @role_name = NULL,
+ @capture_instance = 'dbo_products_v4',
+ @supports_net_changes = 0;
+
+-- SQL Server exposes sp_rename as a capture-instance switch. SeaTunnel now
+-- treats it conservatively as ADD + DROP unless the downstream sink can
+-- preserve existing values itself, so force real post-switch row updates under
+-- the renamed column. A no-op assignment is not enough here because SQL Server
+-- may skip writing CDC rows when the value does not change.
+UPDATE dbo.products
+SET add_column = add_column + 1000
+WHERE id IN (101, 103, 104, 105, 106, 107, 108, 109, 110, 120, 121, 128, 129,
130, 131, 140, 141);
+
+UPDATE dbo.products
+SET add_column = add_column - 1000
+WHERE id IN (101, 103, 104, 105, 106, 107, 108, 109, 110, 120, 121, 128, 129,
130, 131, 140, 141);
+
+DELETE FROM dbo.products WHERE id = 130;
+INSERT INTO dbo.products VALUES (150, 'scooter', 'Small 2-wheel scooter',
3.14, 1);
+INSERT INTO dbo.products VALUES (151, 'car battery', '12V car battery', 8.1,
2);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_to_sqlserver_with_schema_change.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_to_sqlserver_with_schema_change.conf
new file mode 100644
index 0000000000..3ea0f5983c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/resources/sqlservercdc_to_sqlserver_with_schema_change.conf
@@ -0,0 +1,52 @@
+#
+# 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 validates SqlServer CDC schema evolution on Flink
+######
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ SqlServer-CDC {
+ plugin_output = "products"
+ username = "sa"
+ password = "Password!"
+ database-names = ["schema_change_test"]
+ table-names = ["schema_change_test.dbo.products"]
+ url =
"jdbc:sqlserver://sqlserver-host:1433;databaseName=schema_change_test"
+ schema-changes.enabled = true
+ }
+}
+
+sink {
+ Jdbc {
+ plugin_input = "products"
+ driver = "com.microsoft.sqlserver.jdbc.SQLServerDriver"
+ url = "jdbc:sqlserver://sqlserver-host:1433;encrypt=false"
+ user = "sa"
+ password = "Password!"
+ generate_sink_sql = true
+ database = "schema_change_test"
+ table = "dbo.products_sink"
+ batch_size = 1
+ primary_keys = ["id"]
+ }
+}
