[flink][cdc] Track schema event progress in CDC source reader (#8311)

The CDC source reader restored split progress only from data row
`recordsToSkip`. Schema change events can be emitted before any data
rows are consumed, but that progress was not checkpointed independently.
After recovery, the reader could emit the same schema change events
again.

This PR adds explicit schema event progress to
`TableAwareFileStoreSourceSplit` and tracks it with a CDC-specific split
state. The split reader skips already emitted schema change events on
restore, while preserving V1 checkpoint compatibility through a legacy
progress marker.

Author: QuakeWang

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/TableAwareFileStoreSourceSplit.java

@@ -43,6 +43,8 @@ public class TableAwareFileStoreSourceSplit extends FileStoreSourceSplit {
     private final Identifier identifier;
     private final @Nullable Long lastSchemaId;
     private final long schemaId;
+    private final long schemaChangeEventsToSkip;
+    private final boolean legacySchemaProgress;
 
     public TableAwareFileStoreSourceSplit(
             String id,
@@ -51,10 +53,43 @@ public TableAwareFileStoreSourceSplit(
             Identifier identifier,
             @Nullable Long lastSchemaId,
             long schemaId) {
+        this(id, split, recordsToSkip, identifier, lastSchemaId, schemaId, 0L);
+    }
+
+    public TableAwareFileStoreSourceSplit(
+            String id,
+            Split split,
+            long recordsToSkip,
+            Identifier identifier,
+            @Nullable Long lastSchemaId,
+            long schemaId,
+            long schemaChangeEventsToSkip) {
+        this(
+                id,
+                split,
+                recordsToSkip,
+                identifier,
+                lastSchemaId,
+                schemaId,
+                schemaChangeEventsToSkip,
+                false);
+    }
+
+    private TableAwareFileStoreSourceSplit(
+            String id,
+            Split split,
+            long recordsToSkip,
+            Identifier identifier,
+            @Nullable Long lastSchemaId,
+            long schemaId,
+            long schemaChangeEventsToSkip,
+            boolean legacySchemaProgress) {
         super(id, split, recordsToSkip);
         this.identifier = identifier;
         this.lastSchemaId = lastSchemaId;
         this.schemaId = schemaId;
+        this.schemaChangeEventsToSkip = schemaChangeEventsToSkip;
+        this.legacySchemaProgress = legacySchemaProgress;
     }
 
     public Identifier getIdentifier() {
@@ -69,10 +104,38 @@ public long getSchemaId() {
         return schemaId;
     }
 
+    public long schemaChangeEventsToSkip() {
+        return schemaChangeEventsToSkip;
+    }
+
+    public TableAwareFileStoreSourceSplit updateWithProgress(
+            long recordsToSkip, long schemaChangeEventsToSkip) {
+        return new TableAwareFileStoreSourceSplit(
+                splitId(),
+                split(),
+                recordsToSkip,
+                identifier,
+                lastSchemaId,
+                schemaId,
+                schemaChangeEventsToSkip,
+                legacySchemaProgress);
+    }
+
+    public boolean isLegacySchemaProgress() {
+        return legacySchemaProgress;
+    }
+
     @Override
     public TableAwareFileStoreSourceSplit updateWithRecordsToSkip(long recordsToSkip) {
         return new TableAwareFileStoreSourceSplit(
-                splitId(), split(), recordsToSkip, identifier, lastSchemaId, schemaId);
+                splitId(),
+                split(),
+                recordsToSkip,
+                identifier,
+                lastSchemaId,
+                schemaId,
+                schemaChangeEventsToSkip,
+                legacySchemaProgress);
     }
 
     @Override
@@ -87,13 +150,22 @@ public boolean equals(Object o) {
                 && recordsToSkip() == other.recordsToSkip()
                 && identifier.equals(other.identifier)
                 && Objects.equals(lastSchemaId, other.lastSchemaId)
-                && schemaId == other.schemaId;
+                && schemaId == other.schemaId
+                && schemaChangeEventsToSkip == other.schemaChangeEventsToSkip
+                && legacySchemaProgress == other.legacySchemaProgress;
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(
-                splitId(), split(), recordsToSkip(), identifier, lastSchemaId, schemaId);
+                splitId(),
+                split(),
+                recordsToSkip(),
+                identifier,
+                lastSchemaId,
+                schemaId,
+                schemaChangeEventsToSkip,
+                legacySchemaProgress);
     }
 
     @Override
@@ -112,17 +184,23 @@ public String toString() {
                 + lastSchemaId
                 + ", schemaId="
                 + schemaId
+                + ", schemaChangeEventsToSkip="
+                + schemaChangeEventsToSkip
+                + ", legacySchemaProgress="
+                + legacySchemaProgress
                 + '}';
     }
 
     /** The serializer for {@link TableAwareFileStoreSourceSplit}. */
     public static class Serializer
             implements SimpleVersionedSerializer<TableAwareFileStoreSourceSplit> {
+        private static final int VERSION_1 = 1;
+        private static final int VERSION_2 = 2;
         private static final Long NULL_SCHEMA_ID = -1L;
 
         @Override
         public int getVersion() {
-            return 1;
+            return VERSION_2;
         }
 
         @Override
@@ -136,12 +214,18 @@ public byte[] serialize(TableAwareFileStoreSourceSplit split) throws IOException
             view.writeLong(
                     split.getLastSchemaId() == null ? NULL_SCHEMA_ID : split.getLastSchemaId());
             view.writeLong(split.getSchemaId());
+            view.writeLong(split.schemaChangeEventsToSkip());
+            view.writeBoolean(split.isLegacySchemaProgress());
             return out.toByteArray();
         }
 
         @Override
         public TableAwareFileStoreSourceSplit deserialize(int version, byte[] serialized)
                 throws IOException {
+            if (version != VERSION_1 && version != VERSION_2) {
+                throw new IOException(
+                        "Unsupported TableAwareFileStoreSourceSplit version: " + version);
+            }
             ByteArrayInputStream in = new ByteArrayInputStream(serialized);
             DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in);
             String splitId = view.readUTF();
@@ -158,8 +242,18 @@ public TableAwareFileStoreSourceSplit deserialize(int version, byte[] serialized
                 lastSchemaId = null;
             }
             long schemaId = view.readLong();
+            long schemaChangeEventsToSkip = version == VERSION_2 ? view.readLong() : 0L;
+            boolean legacySchemaProgress =
+                    version == VERSION_2 ? view.readBoolean() : version == VERSION_1;
             return new TableAwareFileStoreSourceSplit(
-                    splitId, split, recordsToSkip, identifier, lastSchemaId, schemaId);
+                    splitId,
+                    split,
+                    recordsToSkip,
+                    identifier,
+                    lastSchemaId,
+                    schemaId,
+                    schemaChangeEventsToSkip,
+                    legacySchemaProgress);
         }
     }
 }

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCRecordsWithSplitIds.java

@@ -18,7 +18,6 @@
 
 package org.apache.paimon.flink.pipeline.cdc.source.reader;
 
-import org.apache.paimon.flink.source.FileStoreSourceSplitState;
 import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
 import org.apache.paimon.utils.Reference;
 
@@ -109,7 +108,7 @@ public static void emitRecord(
             SourceReaderContext context,
             RecordIterator<Event> element,
             SourceOutput<Event> output,
-            FileStoreSourceSplitState state,
+            CDCSourceSplitState state,
             FileStoreSourceReaderMetrics metrics) {
         long timestamp = TimestampAssigner.NO_TIMESTAMP;
         if (metrics.getLatestFileCreationTime() != FileStoreSourceReaderMetrics.UNDEFINED) {

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceReader.java

@@ -21,7 +21,6 @@
 import org.apache.paimon.disk.IOManager;
 import org.apache.paimon.flink.pipeline.cdc.source.CDCSource;
 import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
-import org.apache.paimon.flink.source.FileStoreSourceSplitState;
 import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
 
 import org.apache.flink.api.connector.source.SourceReader;
@@ -37,10 +36,7 @@
 /** A {@link SourceReader} that read records from {@link TableAwareFileStoreSourceSplit}. */
 public class CDCSourceReader
         extends SingleThreadMultiplexSourceReaderBase<
-                RecordIterator<Event>,
-                Event,
-                TableAwareFileStoreSourceSplit,
-                FileStoreSourceSplitState> {
+                RecordIterator<Event>, Event, TableAwareFileStoreSourceSplit, CDCSourceSplitState> {
     private static final Logger LOG = LoggerFactory.getLogger(CDCSourceReader.class);
 
     private final IOManager ioManager;
@@ -69,7 +65,7 @@ public void start() {
     }
 
     @Override
-    protected void onSplitFinished(Map<String, FileStoreSourceSplitState> finishedSplitIds) {
+    protected void onSplitFinished(Map<String, CDCSourceSplitState> finishedSplitIds) {
         // this method is called each time when we consume one split
         // it is possible that one response from the coordinator contains multiple splits
         // we should only require for more splits after we've consumed all given splits
@@ -79,16 +75,16 @@ protected void onSplitFinished(Map<String, FileStoreSourceSplitState> finishedSp
     }
 
     @Override
-    protected FileStoreSourceSplitState initializedState(TableAwareFileStoreSourceSplit split) {
+    protected CDCSourceSplitState initializedState(TableAwareFileStoreSourceSplit split) {
         LOG.info("Initializing split {}", split);
-        return new FileStoreSourceSplitState(split);
+        return new CDCSourceSplitState(split);
     }
 
     @Override
     protected TableAwareFileStoreSourceSplit toSplitType(
-            String splitId, FileStoreSourceSplitState splitState) {
+            String splitId, CDCSourceSplitState splitState) {
         LOG.info("Converting split state {} with id {} to split", splitState, splitId);
-        return (TableAwareFileStoreSourceSplit) splitState.toSourceSplit();
+        return splitState.toSourceSplit();
     }
 
     @Override

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitReader.java

@@ -231,15 +231,33 @@ private void checkSplitOrStartNext() throws IOException {
         currentReader = createLazyRecordReader(nextSplit.split());
         currentDataRowsRead = nextSplit.recordsToSkip();
         currentSchemaChangeEvents.clear();
-        if (currentDataRowsRead == 0) {
-            currentSchemaChangeEvents.addAll(schemaChangeEvents);
+        long schemaChangeEventsToSkip =
+                schemaChangeEventsToSkip(nextSplit, schemaChangeEvents.size());
+        for (int i = (int) schemaChangeEventsToSkip; i < schemaChangeEvents.size(); i++) {
+            currentSchemaChangeEvents.add(schemaChangeEvents.get(i));
         }
 
         if (currentDataRowsRead > 0) {
             seek(currentDataRowsRead);
         }
     }
 
+    private long schemaChangeEventsToSkip(
+            TableAwareFileStoreSourceSplit split, int schemaChangeEventCount) throws IOException {
+        long schemaChangeEventsToSkip =
+                split.isLegacySchemaProgress() && split.recordsToSkip() > 0
+                        ? schemaChangeEventCount
+                        : split.schemaChangeEventsToSkip();
+        if (schemaChangeEventsToSkip < 0 || schemaChangeEventsToSkip > schemaChangeEventCount) {
+            throw new IOException(
+                    String.format(
+                            "Invalid schema change event skip count %s for split %s. "
+                                    + "The split has only %s schema change events.",
+                            schemaChangeEventsToSkip, split.splitId(), schemaChangeEventCount));
+        }
+        return schemaChangeEventsToSkip;
+    }
+
     @VisibleForTesting
     protected LazyRecordReader createLazyRecordReader(Split split) {
         return new LazyRecordReader(split, currentTableReaderInfo, tableManager);

paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/pipeline/cdc/source/reader/CDCSourceSplitState.java

@@ -0,0 +1,78 @@
+/*
+ * 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.paimon.flink.pipeline.cdc.source.reader;
+
+import org.apache.paimon.flink.pipeline.cdc.source.TableAwareFileStoreSourceSplit;
+
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.RecordAndPosition;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** State of the CDC source reader with independent data row and schema event progress. */
+public class CDCSourceSplitState {
+
+    private final TableAwareFileStoreSourceSplit split;
+
+    private long recordsToSkip;
+
+    private long schemaChangeEventsToSkip;
+
+    public CDCSourceSplitState(TableAwareFileStoreSourceSplit split) {
+        this.split = checkNotNull(split);
+        this.recordsToSkip = split.recordsToSkip();
+        this.schemaChangeEventsToSkip = split.schemaChangeEventsToSkip();
+    }
+
+    public void setPosition(RecordAndPosition<Event> position) {
+        checkArgument(position.getOffset() == CheckpointedPosition.NO_OFFSET);
+        if (position.getRecord() instanceof SchemaChangeEvent) {
+            schemaChangeEventsToSkip++;
+        } else {
+            recordsToSkip = position.getRecordSkipCount();
+        }
+    }
+
+    public long recordsToSkip() {
+        return recordsToSkip;
+    }
+
+    public long schemaChangeEventsToSkip() {
+        return schemaChangeEventsToSkip;
+    }
+
+    public TableAwareFileStoreSourceSplit toSourceSplit() {
+        return split.updateWithProgress(recordsToSkip, schemaChangeEventsToSkip);
+    }
+
+    @Override
+    public String toString() {
+        return "CDCSourceSplitState{"
+                + "split="
+                + split
+                + ", recordsToSkip="
+                + recordsToSkip
+                + ", schemaChangeEventsToSkip="
+                + schemaChangeEventsToSkip
+                + '}';
+    }
+}