[FLINK-29802][state] Changelog supports native savepoint

Author: masteryhx

flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractChangelogStateBackend.java

@@ -19,6 +19,7 @@
 package org.apache.flink.state.changelog;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.execution.SavepointFormatType;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.runtime.execution.Environment;
 import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
@@ -108,6 +109,11 @@ public StateBackend getDelegatedStateBackend() {
         return delegatedStateBackend;
     }
 
+    @Override
+    public boolean supportsSavepointFormat(SavepointFormatType formatType) {
+        return delegatedStateBackend.supportsSavepointFormat(formatType);
+    }
+
     protected abstract <K> CheckpointableKeyedStateBackend<K> restore(
             Environment env,
             String operatorIdentifier,

flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackend.java

@@ -30,6 +30,7 @@
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.runtime.checkpoint.CheckpointOptions;
 import org.apache.flink.runtime.checkpoint.CheckpointType;
+import org.apache.flink.runtime.checkpoint.SnapshotType;
 import org.apache.flink.runtime.state.AbstractKeyedStateBackend;
 import org.apache.flink.runtime.state.CheckpointStateOutputStream;
 import org.apache.flink.runtime.state.CheckpointStorageLocationReference;
@@ -67,6 +68,7 @@
 import org.apache.flink.state.changelog.restore.ChangelogRestoreTarget;
 import org.apache.flink.state.changelog.restore.FunctionDelegationHelper;
 import org.apache.flink.state.common.PeriodicMaterializationManager.MaterializationTarget;
+import org.apache.flink.util.concurrent.FutureUtils;
 
 import org.apache.flink.shaded.guava31.com.google.common.io.Closer;
 
@@ -89,6 +91,7 @@
 import java.util.TreeMap;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
 import java.util.concurrent.RunnableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -379,6 +382,11 @@ public RunnableFuture<SnapshotResult<KeyedStateHandle>> snapshot(
             @Nonnull CheckpointStreamFactory streamFactory,
             @Nonnull CheckpointOptions checkpointOptions)
             throws Exception {
+
+        if (checkpointOptions.getCheckpointType().isSavepoint()) {
+            return nativeSavepoint(checkpointId, timestamp, streamFactory, checkpointOptions);
+        }
+
         // The range to upload may overlap with the previous one(s). To reuse them, we could store
         // the previous results either here in the backend or in the writer. However,
         // materialization may truncate only a part of the previous result and the backend would
@@ -416,6 +424,57 @@ public RunnableFuture<SnapshotResult<KeyedStateHandle>> snapshot(
                         .thenApply(this::castSnapshotResult));
     }
 
+    private RunnableFuture<SnapshotResult<KeyedStateHandle>> nativeSavepoint(
+            long checkpointId,
+            long timestamp,
+            @Nonnull CheckpointStreamFactory streamFactory,
+            @Nonnull CheckpointOptions checkpointOptions)
+            throws Exception {
+
+        SnapshotType.SharingFilesStrategy sharingFilesStrategy =
+                checkpointOptions.getCheckpointType().getSharingFilesStrategy();
+        if (sharingFilesStrategy != SnapshotType.SharingFilesStrategy.NO_SHARING) {
+            throw new UnsupportedOperationException(
+                    "ChangelogKeyedStateBackend doesn't support native savepoint with SharingFilesStrategy: "
+                            + sharingFilesStrategy);
+        }
+
+        long materializationID = materializedId++;
+        // For NO_SHARING native savepoint, trigger delegated one
+        RunnableFuture<SnapshotResult<KeyedStateHandle>> delegatedSnapshotResult =
+                keyedStateBackend.snapshot(
+                        materializationID, timestamp, streamFactory, checkpointOptions);
+
+        materializationIdByCheckpointId.put(checkpointId, materializationID);
+        return new FutureTask<SnapshotResult<KeyedStateHandle>>(
+                () -> {
+                    SnapshotResult<KeyedStateHandle> result =
+                            FutureUtils.runIfNotDoneAndGet(delegatedSnapshotResult);
+                    return castSnapshotResult(
+                            buildSnapshotResult(
+                                    checkpointId,
+                                    SnapshotResult.empty(),
+                                    new ChangelogSnapshotState(
+                                            getMaterializedResult(result), materializationID)));
+                }) {
+            @Override
+            public boolean cancel(boolean mayInterruptIfRunning) {
+                return delegatedSnapshotResult.cancel(mayInterruptIfRunning)
+                        && super.cancel(mayInterruptIfRunning);
+            }
+
+            @Override
+            public boolean isCancelled() {
+                return delegatedSnapshotResult.isCancelled() && super.isCancelled();
+            }
+
+            @Override
+            public boolean isDone() {
+                return delegatedSnapshotResult.isDone() && super.isDone();
+            }
+        };
+    }
+
     @SuppressWarnings("unchecked")
     private SnapshotResult<KeyedStateHandle> castSnapshotResult(SnapshotResult<?> snapshotResult) {
         return (SnapshotResult<KeyedStateHandle>) snapshotResult;
@@ -1013,6 +1072,19 @@ private class ChangelogSnapshotState {
         /** ID of this materialization corresponding to the nested backend checkpoint ID. */
         private final long materializationID;
 
+        /**
+         * Construct a ChangelogSnapshotState with empty non-materialized part, which could be used
+         * when triggering manual materialization.
+         */
+        public ChangelogSnapshotState(
+                List<KeyedStateHandle> materializedSnapshot, long materializationID) {
+            this(
+                    materializedSnapshot,
+                    Collections.emptyList(),
+                    SequenceNumber.of(Long.MAX_VALUE),
+                    materializationID);
+        }
+
         public ChangelogSnapshotState(
                 List<KeyedStateHandle> materializedSnapshot,
                 List<ChangelogStateHandle> restoredNonMaterialized,