[flink] Fix restore_as_latest overwrite delta

Ensure restore_as_latest writes an overwrite delta so streaming overwrite readers can observe restored file changes.

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

Author: zhuxiangyi

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -1171,17 +1171,61 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
                 checkNotNull(
                         snapshotManager.latestSnapshot(),
                         "Latest snapshot is null, can not restore.");
+
+        Map<FileEntry.Identifier, ManifestEntry> latestEntries = new HashMap<>();
+        FileEntry.mergeEntries(
+                manifestFile,
+                manifestList.readDataManifests(latest),
+                latestEntries,
+                options.scanManifestParallelism());
+
+        latestEntries.entrySet().removeIf(entry -> entry.getValue().kind() != FileKind.ADD);
+
+        Map<FileEntry.Identifier, ManifestEntry> targetEntries = new HashMap<>();
+        FileEntry.mergeEntries(
+                manifestFile,
+                manifestList.readDataManifests(targetSnapshot),
+                targetEntries,
+                options.scanManifestParallelism());
+        targetEntries.entrySet().removeIf(entry -> entry.getValue().kind() != FileKind.ADD);
+
+        List<ManifestEntry> deltaFiles = new ArrayList<>();
+        for (Map.Entry<FileEntry.Identifier, ManifestEntry> entry : latestEntries.entrySet()) {
+            if (!targetEntries.containsKey(entry.getKey())) {
+                ManifestEntry manifestEntry = entry.getValue();
+                deltaFiles.add(
+                        ManifestEntry.create(
+                                FileKind.DELETE,
+                                manifestEntry.partition(),
+                                manifestEntry.bucket(),
+                                manifestEntry.totalBuckets(),
+                                manifestEntry.file()));
+            }
+        }
+        for (Map.Entry<FileEntry.Identifier, ManifestEntry> entry : targetEntries.entrySet()) {
+            if (!latestEntries.containsKey(entry.getKey())) {
+                ManifestEntry manifestEntry = entry.getValue();
+                deltaFiles.add(
+                        ManifestEntry.create(
+                                FileKind.ADD,
+                                manifestEntry.partition(),
+                                manifestEntry.bucket(),
+                                manifestEntry.totalBuckets(),
+                                manifestEntry.file()));
+            }
+        }
+
         Pair<String, Long> baseManifestList =
-                manifestList.write(manifestList.readDataManifests(targetSnapshot));
-        Pair<String, Long> emptyDeltaManifestList = manifestList.write(emptyList());
+                manifestList.write(manifestFile.write(new ArrayList<>(latestEntries.values())));
+        Pair<String, Long> deltaManifestList = manifestList.write(manifestFile.write(deltaFiles));
         Snapshot newSnapshot =
                 new Snapshot(
                         latest.id() + 1,
                         targetSnapshot.schemaId(),
                         baseManifestList.getKey(),
                         baseManifestList.getRight(),
-                        emptyDeltaManifestList.getKey(),
-                        emptyDeltaManifestList.getRight(),
+                        deltaManifestList.getKey(),
+                        deltaManifestList.getRight(),
                         null,
                         null,
                         targetSnapshot.indexManifest(),
@@ -1190,14 +1234,14 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
                         CommitKind.OVERWRITE,
                         System.currentTimeMillis(),
                         targetSnapshot.totalRecordCount(),
-                        0L,
+                        recordCountAdd(deltaFiles) - recordCountDelete(deltaFiles),
                         null,
                         targetSnapshot.watermark(),
                         targetSnapshot.statistics(),
                         targetSnapshot.properties(),
                         targetSnapshot.nextRowId());
 
-        return commitSnapshotImpl(newSnapshot, emptyList());
+        return commitSnapshotImpl(newSnapshot, new ArrayList<>(PartitionEntry.merge(deltaFiles)));
     }
 
     public void compactManifest() {

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java

@@ -18,9 +18,12 @@
 
 package org.apache.paimon.flink.procedure;
 
+import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.BatchTableCommit;
 import org.apache.paimon.table.sink.BatchTableWrite;
@@ -30,6 +33,8 @@
 import org.apache.flink.types.Row;
 import org.junit.jupiter.api.Test;
 
+import java.util.List;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -53,14 +58,24 @@ public void testRestoreSnapshotAsLatest() throws Exception {
                 .containsExactly(Row.of(3L, 1L, 4L));
 
         assertEquals(4, snapshotManager.latestSnapshotId());
+        assertRestoreDelta(table, 4, 0, 2, -2L);
         assertTrue(snapshotManager.snapshotExists(2));
         assertTrue(snapshotManager.snapshotExists(3));
         assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "a"));
 
-        commitRow(table, 4, "d");
+        assertThat(sql("CALL sys.restore_as_latest(`table` => 'default.T', snapshot_id => 3)"))
+                .containsExactly(Row.of(4L, 3L, 5L));
+
         assertEquals(5, snapshotManager.latestSnapshotId());
+        assertRestoreDelta(table, 5, 2, 0, 2L);
+        assertThat(sql("SELECT * FROM T"))
+                .containsExactlyInAnyOrder(Row.of(1, "a"), Row.of(2, "b"), Row.of(3, "c"));
+
+        commitRow(table, 4, "d");
+        assertEquals(6, snapshotManager.latestSnapshotId());
         assertThat(sql("SELECT * FROM T"))
-                .containsExactlyInAnyOrder(Row.of(1, "a"), Row.of(4, "d"));
+                .containsExactlyInAnyOrder(
+                        Row.of(1, "a"), Row.of(2, "b"), Row.of(3, "c"), Row.of(4, "d"));
     }
 
     @Test
@@ -81,11 +96,28 @@ public void testRestoreTagAsLatest() throws Exception {
                 .containsExactly(Row.of(3L, 1L, 4L));
 
         assertEquals(4, snapshotManager.latestSnapshotId());
+        assertRestoreDelta(table, 4, 0, 2, -2L);
         assertTrue(snapshotManager.snapshotExists(2));
         assertTrue(snapshotManager.snapshotExists(3));
         assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "a"));
     }
 
+    private void assertRestoreDelta(
+            FileStoreTable table,
+            long snapshotId,
+            long expectedNumAddedFiles,
+            long expectedNumDeletedFiles,
+            long expectedDeltaRecordCount) {
+        Snapshot snapshot = table.snapshot(snapshotId);
+        ManifestList manifestList = table.store().manifestListFactory().create();
+        List<ManifestFileMeta> deltaManifests = manifestList.readDeltaManifests(snapshot);
+
+        assertThat(deltaManifests).hasSize(1);
+        assertThat(deltaManifests.get(0).numAddedFiles()).isEqualTo(expectedNumAddedFiles);
+        assertThat(deltaManifests.get(0).numDeletedFiles()).isEqualTo(expectedNumDeletedFiles);
+        assertThat(snapshot.deltaRecordCount()).isEqualTo(expectedDeltaRecordCount);
+    }
+
     private void commitRow(FileStoreTable table, int id, String name) throws Exception {
         BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
         try (BatchTableWrite write = writeBuilder.newWrite();