[flink] Auto-tag savepoints on coordinator commit path

Mirror AutoTagForSavepointCommitterOperator on the JM-side commit path:
writers mark the savepoint bit on their buffered committables, the
coordinator rebuilds the pending-tag set from the replayed bit and
creates savepoint-<id> tags after commit (dropping them on abort).

Author: liubiao.leo

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CoordinatorCommittingRowDataStoreWriteOperator.java

@@ -33,9 +33,12 @@
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
 import org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy;
+import org.apache.flink.runtime.checkpoint.CheckpointOptions;
 import org.apache.flink.runtime.operators.coordination.OperatorEventGateway;
+import org.apache.flink.runtime.state.CheckpointStreamFactory;
 import org.apache.flink.runtime.state.StateInitializationContext;
 import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.OperatorSnapshotFutures;
 import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
 import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
 import org.apache.flink.streaming.api.watermark.Watermark;
@@ -158,23 +161,16 @@ public void snapshotState(StateSnapshotContext context) throws Exception {
         pendingCommittableState.addAll(new ArrayList<>(pendingCommittables.values()));
     }
 
-    @Override
-    public void notifyCheckpointComplete(long checkpointId) throws Exception {
-        super.notifyCheckpointComplete(checkpointId);
-        // operator state already persisted these; we no longer need to replay them on the next
-        // restore
-        pendingCommittables.headMap(checkpointId, true).clear();
-    }
-
     @Override
     protected void emitCommittables(boolean waitCompaction, long checkpointId) throws IOException {
+        // Runs before the checkpoint barrier (via prepareSnapshotPreBarrier) and at end-of-input.
+        // Produces and buffers this checkpoint's committables and forwards them downstream, but does
+        // NOT report to the coordinator yet: at this point a savepoint is indistinguishable from a
+        // normal checkpoint. The report is deferred to snapshotState / endInput.
         List<Committable> committables = prepareCommit(waitCompaction, checkpointId);
         CheckpointCommittables entry =
                 new CheckpointCommittables(
                         checkpointId, committables, currentWatermark, currentIdle);
-        // Emit an event per (subtask, checkpoint) regardless of whether committables is empty.
-        operatorEventGateway.sendEventToCoordinator(
-                CommittableEvent.create(checkpointId, entry, eventSerializer));
         // Always buffer the per-checkpoint entry so an empty barrier — even one that has not seen
         // a real watermark yet — survives restore. The coordinator relies on every subtask
         // having an entry for the checkpoint being aligned so its watermark min stays sound.
@@ -185,6 +181,48 @@ protected void emitCommittables(boolean waitCompaction, long checkpointId) throw
         committables.forEach(committable -> output.collect(new StreamRecord<>(committable)));
     }
 
+    @Override
+    public OperatorSnapshotFutures snapshotState(
+            long checkpointId,
+            long timestamp,
+            CheckpointOptions checkpointOptions,
+            CheckpointStreamFactory storageLocation)
+            throws Exception {
+        // emitCommittables already ran in prepareSnapshotPreBarrier and buffered this checkpoint's
+        // entry. The savepoint intent only becomes observable here, so stamp it on the buffered
+        // entry before reporting so both the event and the persisted state carry it.
+        if (checkpointOptions.getCheckpointType().isSavepoint()) {
+            pendingCommittables.computeIfPresent(
+                    checkpointId, (id, entry) -> entry.withSavepoint(true));
+        }
+        reportToCoordinator(checkpointId);
+        // super drives snapshotState(StateSnapshotContext) which persists the buffered pending map.
+        return super.snapshotState(checkpointId, timestamp, checkpointOptions, storageLocation);
+    }
+
+    @Override
+    public void endInput() throws Exception {
+        // endInput emits the Long.MAX_VALUE entry via emitCommittables but is not followed by a
+        // snapshotState, so report it here. End-of-input is never a savepoint.
+        super.endInput();
+        reportToCoordinator(Long.MAX_VALUE);
+    }
+
+    @Override
+    public void notifyCheckpointComplete(long checkpointId) throws Exception {
+        super.notifyCheckpointComplete(checkpointId);
+        // operator state already persisted these; we no longer need to replay them on the next
+        // restore
+        pendingCommittables.headMap(checkpointId, true).clear();
+    }
+
+    /** Sends the buffered entry for {@code checkpointId} to the coordinator, one per checkpoint. */
+    private void reportToCoordinator(long checkpointId) throws IOException {
+        operatorEventGateway.sendEventToCoordinator(
+                CommittableEvent.create(
+                        checkpointId, pendingCommittables.get(checkpointId), eventSerializer));
+    }
+
     @Override
     public void processWatermark(Watermark mark) throws Exception {
         super.processWatermark(mark);

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java

@@ -400,14 +400,6 @@ static void checkCoordinatorCommitPreconditions(
                         + PRECOMMIT_COMPACT.key()
                         + " = false.");
 
-        // The OperatorCoordinator cannot tell a savepoint from a normal checkpoint.
-        // TODO support savepoint auto-tag.
-        checkArgument(
-                !options.get(SINK_AUTO_TAG_FOR_SAVEPOINT),
-                "Could not enable coordinator commit because "
-                        + SINK_AUTO_TAG_FOR_SAVEPOINT.key()
-                        + " is enabled, which is not supported yet.");
-
         // TODO concurrent checkpoints are not supported yet.
         checkArgument(
                 checkpointConfig.getMaxConcurrentCheckpoints() == 1,

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowAppendTableSink.java

@@ -110,7 +110,17 @@ private static class CoordinatorCommittingFactory extends RowDataStoreWriteOpera
         public OperatorCoordinator.Provider getCoordinatorProvider(
                 String operatorName, OperatorID operatorID) {
             return new CommittingWriteOperatorCoordinator.Provider(
-                    operatorID, committerFactory, streamingCheckpointEnabled, initialCommitUser);
+                    operatorID,
+                    committerFactory,
+                    streamingCheckpointEnabled,
+                    initialCommitUser,
+                    new Options(table.options())
+                            .get(FlinkConnectorOptions.SINK_AUTO_TAG_FOR_SAVEPOINT),
+                    table::snapshotManager,
+                    table::tagManager,
+                    () -> table.store().newTagDeletion(),
+                    () -> table.store().createTagCallbacks(table),
+                    table.coreOptions().tagDefaultTimeRetained());
         }
 
         @Override

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittables.java

@@ -35,19 +35,34 @@ public class CheckpointCommittables {
     // Idle bit is frozen at barrier time together with watermark; mirrors what Flink's
     // StatusWatermarkValve would have observed on the writer's input at the moment of the barrier.
     private final boolean idle;
+    // Whether the checkpoint that produced these committables is a Flink savepoint. Only the writer
+    // can observe this (via CheckpointOptions); it rides along so the JM-side coordinator, whose
+    // OperatorCoordinator API carries no savepoint indicator, can auto-tag the resulting snapshot.
+    private final boolean savepoint;
 
     public CheckpointCommittables(
-            long checkpointId, List<Committable> committables, long watermark, boolean idle) {
+            long checkpointId,
+            List<Committable> committables,
+            long watermark,
+            boolean idle,
+            boolean savepoint) {
         this.checkpointId = checkpointId;
         this.committables = committables;
         this.watermark = watermark;
         this.idle = idle;
+        this.savepoint = savepoint;
+    }
+
+    // Convenience for callers that are not savepoint-aware yet.
+    public CheckpointCommittables(
+            long checkpointId, List<Committable> committables, long watermark, boolean idle) {
+        this(checkpointId, committables, watermark, idle, false);
     }
 
     // Convenience for callers that only need the pre-idle-aware shape (ACTIVE writer).
     public CheckpointCommittables(
             long checkpointId, List<Committable> committables, long watermark) {
-        this(checkpointId, committables, watermark, false);
+        this(checkpointId, committables, watermark, false, false);
     }
 
     public long checkpointId() {
@@ -66,6 +81,15 @@ public boolean idle() {
         return idle;
     }
 
+    public boolean savepoint() {
+        return savepoint;
+    }
+
+    /** Returns a copy with the savepoint bit set; all other fields preserved. */
+    public CheckpointCommittables withSavepoint(boolean savepoint) {
+        return new CheckpointCommittables(checkpointId, committables, watermark, idle, savepoint);
+    }
+
     public int size() {
         return committables.size();
     }
@@ -77,7 +101,7 @@ public boolean isEmpty() {
     @Override
     public String toString() {
         return String.format(
-                "CheckpointCommittables{checkpointId=%d, watermark=%d, idle=%s, committables=%s}",
-                checkpointId, watermark, idle, committables);
+                "CheckpointCommittables{checkpointId=%d, watermark=%d, idle=%s, savepoint=%s, committables=%s}",
+                checkpointId, watermark, idle, savepoint, committables);
     }
 }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CheckpointCommittablesSerializer.java

@@ -43,7 +43,8 @@ public CheckpointCommittablesSerializer(CommittableSerializer committableSeriali
     public int getVersion() {
         // v1: checkpointId + watermark + committables
         // v2: v1 + idle bit (appended before the committable list to keep the ordering explicit)
-        return 2;
+        // v3: v2 + savepoint bit (appended after idle, still before the committable list)
+        return 3;
     }
 
     @Override
@@ -52,6 +53,7 @@ public byte[] serialize(CheckpointCommittables value) throws IOException {
         out.writeLong(value.checkpointId());
         out.writeLong(value.watermark());
         out.writeBoolean(value.idle());
+        out.writeBoolean(value.savepoint());
         // Nested serializer version comes before the list so the reader can pick the right decoder
         // before touching any list bytes — mirrors ManifestCommittableSerializer's layout.
         out.writeInt(committableSerializer.getVersion());
@@ -67,7 +69,7 @@ public byte[] serialize(CheckpointCommittables value) throws IOException {
 
     @Override
     public CheckpointCommittables deserialize(int version, byte[] serialized) throws IOException {
-        if (version != 1 && version != 2) {
+        if (version != 1 && version != 2 && version != 3) {
             throw new IOException("Unknown version " + version);
         }
         DataInputDeserializer in = new DataInputDeserializer(serialized);
@@ -76,6 +78,8 @@ public CheckpointCommittables deserialize(int version, byte[] serialized) throws
         // v1 payloads pre-date idle tracking; default to ACTIVE (idle=false), which reproduces
         // the pre-idle-aware behaviour: every subtask contributes to the min unconditionally.
         boolean idle = version >= 2 && in.readBoolean();
+        // v1/v2 payloads pre-date savepoint tracking; default to false (a normal checkpoint).
+        boolean savepoint = version >= 3 && in.readBoolean();
         int committableVersion = in.readInt();
         int count = in.readInt();
         List<Committable> committables = new ArrayList<>(count);
@@ -85,6 +89,6 @@ public CheckpointCommittables deserialize(int version, byte[] serialized) throws
             in.readFully(bytes);
             committables.add(committableSerializer.deserialize(committableVersion, bytes));
         }
-        return new CheckpointCommittables(checkpointId, committables, watermark, idle);
+        return new CheckpointCommittables(checkpointId, committables, watermark, idle, savepoint);
     }
 }