[minor][pipeline-connector/sqlserver] Fix SQL Server schema restore without per-record JDBC

Restore CreateTableEvent state from source splits during reader initialization so savepoint recovery does not fall back to per-record JDBC lookups.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: leonardBang

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlServerPipelineRecordEmitter.java

@@ -23,6 +23,7 @@
 import org.apache.flink.cdc.common.schema.Schema;
 import org.apache.flink.cdc.connectors.base.options.StartupOptions;
 import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitState;
 import org.apache.flink.cdc.connectors.base.source.metrics.SourceReaderMetrics;
 import org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceRecordEmitter;
@@ -91,8 +92,7 @@ protected void processElement(
             SourceRecord element, SourceOutput<T> output, SourceSplitState splitState)
             throws Exception {
         if (isSchemaChangeEvent(element) && splitState.isStreamSplitState()) {
-            restoreCreateTableEventsFromSplitSchemas(
-                    splitState.asStreamSplitState().getTableSchemas());
+            cacheCreateTableEventsFromSchemas(splitState.asStreamSplitState().getTableSchemas());
         }
 
         if (shouldEmitAllCreateTableEventsInSnapshotMode && isBounded) {
@@ -104,15 +104,20 @@ protected void processElement(
             // to downstream to avoid checkpoint timeout.
             io.debezium.relational.TableId tableId =
                     splitState.asSnapshotSplitState().toSourceSplit().getTableId();
-            emitCreateTableEventIfNeeded(tableId, output);
+            emitCreateTableEventIfNeeded(tableId, output, splitState);
         } else if (isDataChangeRecord(element)) {
             // Handle data change events, schema change events are handled downstream directly
             io.debezium.relational.TableId tableId = getTableId(element);
-            emitCreateTableEventIfNeeded(tableId, output);
+            emitCreateTableEventIfNeeded(tableId, output, splitState);
         }
         super.processElement(element, output, splitState);
     }
 
+    @Override
+    public void applySplit(SourceSplitBase split) {
+        cacheCreateTableEventsFromSchemas(split.getTableSchemas());
+    }
+
     @SuppressWarnings("unchecked")
     private void emitAllCreateTableEvents(SourceOutput<T> output) {
         createTableEventCache.forEach(
@@ -124,33 +129,28 @@ private void emitAllCreateTableEvents(SourceOutput<T> output) {
 
     @SuppressWarnings("unchecked")
     private void emitCreateTableEventIfNeeded(
-            io.debezium.relational.TableId tableId, SourceOutput<T> output) {
+            io.debezium.relational.TableId tableId,
+            SourceOutput<T> output,
+            SourceSplitState splitState) {
         if (alreadySendCreateTableTables.contains(tableId)) {
             return;
         }
 
+        cacheCreateTableEventsFromSchemas(splitState.toSourceSplit().getTableSchemas());
         CreateTableEvent createTableEvent = createTableEventCache.get(tableId);
-        if (createTableEvent != null) {
-            output.collect((T) createTableEvent);
-        } else {
-            // Table not in cache, fetch schema from database
-            try (SqlServerConnection jdbc =
-                    createSqlServerConnection(sourceConfig.getDbzConnectorConfig())) {
-                createTableEvent = buildCreateTableEvent(jdbc, tableId);
-                output.collect((T) createTableEvent);
-                createTableEventCache.put(tableId, createTableEvent);
-            } catch (SQLException e) {
-                throw new RuntimeException("Failed to get table schema for " + tableId, e);
-            }
+        if (createTableEvent == null) {
+            throw new IllegalStateException(
+                    "Missing CreateTableEvent for table "
+                            + tableId
+                            + ". Table schema should have been restored before processing records.");
         }
+        output.collect((T) createTableEvent);
         alreadySendCreateTableTables.add(tableId);
     }
 
-    private void restoreCreateTableEventsFromSplitSchemas(
+    private void cacheCreateTableEventsFromSchemas(
             Map<io.debezium.relational.TableId, TableChange> tableSchemas) {
-        if (!sourceConfig.isIncludeSchemaChanges()
-                || tableSchemas == null
-                || tableSchemas.isEmpty()) {
+        if (tableSchemas == null || tableSchemas.isEmpty()) {
             return;
         }
         for (Map.Entry<io.debezium.relational.TableId, TableChange> entry :