[Feature][Connector-V2][SqlServer CDC] support sqlserver schema evolution (#10890)

Author: CloverDew

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/utils/SourceRecordUtils.java

@@ -51,7 +51,8 @@ private 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";

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 Lsn getMinLsn(String databaseName, String changeTableName) throws SQLExce
     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 @@ private Optional<ColumnEditor> doReadTableColumn(
         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 List<SqlServerChangeTable> getNewChangeTables(
                 });
     }
 
+    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 Table getTableSchemaFromTable(String databaseName, SqlServerChangeTable c
                     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 String getNameOfChangeTable(String captureName) {
         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.
      *