| rollback_to_timestamp |
diff --git a/docs/docs/maintenance/manage-snapshots.mdx b/docs/docs/maintenance/manage-snapshots.mdx
index 821aa0b7559f..edbbf0ca0ddf 100644
--- a/docs/docs/maintenance/manage-snapshots.mdx
+++ b/docs/docs/maintenance/manage-snapshots.mdx
@@ -353,6 +353,25 @@ CALL sys.rollback(table => 'database_name.table_name', snapshot => snasphot_id);
+## Restore Snapshot as Latest
+
+Restore a table to the state of a specific snapshot ID by creating a new latest snapshot. Unlike rollback, this operation
+does not delete snapshots or tags whose snapshot id is larger than the restored snapshot.
+
+
+
+
+
+Run the following command:
+
+```sql
+CALL sys.restore_as_latest(`table` => 'database_name.table_name', snapshot_id => );
+```
+
+
+
+
+
## Remove Orphan Files
Paimon files are deleted physically only when expiring snapshots. However, it is possible that some unexpected errors occurred
@@ -402,4 +421,4 @@ The table can be `*` to clean all tables in the database.
-
\ No newline at end of file
+
diff --git a/docs/docs/maintenance/manage-tags.mdx b/docs/docs/maintenance/manage-tags.mdx
index 78e588e5f53f..173dbe5e7ab2 100644
--- a/docs/docs/maintenance/manage-tags.mdx
+++ b/docs/docs/maintenance/manage-tags.mdx
@@ -300,3 +300,22 @@ CALL sys.rollback(table => 'test.t', version => '2');
+
+## Restore Tag as Latest
+
+Restore a table to the state of a specific tag by creating a new latest snapshot. Unlike rollback, this operation does not
+delete snapshots or tags whose snapshot id is larger than the restored tag.
+
+
+
+
+
+Run the following command:
+
+```sql
+CALL sys.restore_as_latest(`table` => 'database_name.table_name', tag => 'tag_name');
+```
+
+
+
+
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
index 31fb3c52cab6..a283360ca3e9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
@@ -74,6 +74,9 @@ int overwritePartition(
/** Compact the manifest entries only. */
void compactManifest();
+ /** Restore the target snapshot as the latest snapshot. */
+ boolean restoreAsLatest(Snapshot targetSnapshot);
+
/** Abort an unsuccessful commit. The data files will be deleted. */
void abort(List commitMessages);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 7cca259cbf9f..dd406d5c0dc2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -107,6 +107,7 @@
import static org.apache.paimon.partition.PartitionPredicate.createBinaryPartitions;
import static org.apache.paimon.partition.PartitionPredicate.createPartitionPredicate;
import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
/**
* Default implementation of {@link FileStoreCommit}.
@@ -1164,6 +1165,41 @@ public boolean replaceManifestList(
return commitSnapshotImpl(newSnapshot, emptyList());
}
+ @Override
+ public boolean restoreAsLatest(Snapshot targetSnapshot) {
+ Snapshot latest =
+ checkNotNull(
+ snapshotManager.latestSnapshot(),
+ "Latest snapshot is null, can not restore.");
+ Pair baseManifestList =
+ manifestList.write(manifestList.readDataManifests(targetSnapshot));
+ Pair emptyDeltaManifestList = manifestList.write(emptyList());
+ Snapshot newSnapshot =
+ new Snapshot(
+ latest.id() + 1,
+ targetSnapshot.schemaId(),
+ baseManifestList.getKey(),
+ baseManifestList.getRight(),
+ emptyDeltaManifestList.getKey(),
+ emptyDeltaManifestList.getRight(),
+ null,
+ null,
+ targetSnapshot.indexManifest(),
+ commitUser,
+ Long.MAX_VALUE,
+ CommitKind.OVERWRITE,
+ System.currentTimeMillis(),
+ targetSnapshot.totalRecordCount(),
+ 0L,
+ null,
+ targetSnapshot.watermark(),
+ targetSnapshot.statistics(),
+ targetSnapshot.properties(),
+ targetSnapshot.nextRowId());
+
+ return commitSnapshotImpl(newSnapshot, emptyList());
+ }
+
public void compactManifest() {
int retryCount = 0;
long startMillis = System.currentTimeMillis();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index 0311c9bbe4d8..167394f18b20 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.sink;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.consumer.ConsumerManager;
import org.apache.paimon.fs.Path;
@@ -203,6 +204,15 @@ public void compactManifests() {
commit.compactManifest();
}
+ public boolean restoreAsLatest(Snapshot targetSnapshot) {
+ checkCommitted();
+ boolean success = commit.restoreAsLatest(targetSnapshot);
+ if (success) {
+ maintain(COMMIT_IDENTIFIER, maintainExecutor, true);
+ }
+ return success;
+ }
+
private void checkCommitted() {
checkState(!batchCommitted, "BatchTableCommit only support one-time committing.");
batchCommitted = true;
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedure.java
new file mode 100644
index 000000000000..8aeb8576d70a
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedure.java
@@ -0,0 +1,133 @@
+/*
+ * 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.procedure;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+import org.apache.flink.types.Row;
+
+import java.util.List;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.UUID;
+
+/**
+ * Restore as latest procedure. Usage:
+ *
+ *
+ * -- restore a snapshot as the latest snapshot
+ * CALL sys.restore_as_latest(`table` => 'tableId', snapshot_id => snapshotId)
+ *
+ * -- restore a tag as the latest snapshot
+ * CALL sys.restore_as_latest(`table` => 'tableId', tag => 'tagName')
+ *
+ */
+public class RestoreAsLatestProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "restore_as_latest";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "tag", type = @DataTypeHint("STRING"), isOptional = true),
+ @ArgumentHint(
+ name = "snapshot_id",
+ type = @DataTypeHint("BIGINT"),
+ isOptional = true)
+ })
+ public @DataTypeHint(
+ "ROW")
+ Row[] call(ProcedureContext procedureContext, String tableId, String tagName, Long snapshotId)
+ throws Catalog.TableNotExistException {
+ Table table = catalog.getTable(Identifier.fromString(tableId));
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+
+ FileStore store = fileStoreTable.store();
+ Snapshot latestSnapshot = store.snapshotManager().latestSnapshot();
+ Preconditions.checkNotNull(latestSnapshot, "Latest snapshot is null, can not restore.");
+
+ boolean hasTag = !StringUtils.isNullOrWhitespaceOnly(tagName);
+ boolean hasSnapshot = snapshotId != null;
+ Preconditions.checkArgument(
+ hasTag != hasSnapshot, "Must specify exactly one of tag and snapshot_id.");
+
+ Snapshot targetSnapshot;
+ if (hasTag) {
+ targetSnapshot = store.newTagManager().getOrThrow(tagName).trimToSnapshot();
+ } else {
+ targetSnapshot = findSnapshot(store, snapshotId);
+ }
+
+ try (TableCommitImpl commit =
+ fileStoreTable.newCommit("restore-as-latest-" + UUID.randomUUID().toString())) {
+ Preconditions.checkState(
+ commit.restoreAsLatest(targetSnapshot),
+ "Failed to restore snapshot %s as latest.",
+ targetSnapshot.id());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format("Failed to restore snapshot %s as latest.", targetSnapshot.id()),
+ e);
+ }
+
+ return new Row[] {
+ Row.of(
+ latestSnapshot.id(),
+ targetSnapshot.id(),
+ store.snapshotManager().latestSnapshotId())
+ };
+ }
+
+ private Snapshot findSnapshot(FileStore store, long snapshotId) {
+ SnapshotManager snapshotManager = store.snapshotManager();
+ if (snapshotManager.snapshotExists(snapshotId)) {
+ return snapshotManager.snapshot(snapshotId);
+ }
+
+ SortedMap> tags = store.newTagManager().tags();
+ for (Map.Entry> entry : tags.entrySet()) {
+ if (entry.getKey().id() == snapshotId) {
+ return entry.getKey();
+ } else if (entry.getKey().id() > snapshotId) {
+ break;
+ }
+ }
+
+ throw new IllegalArgumentException(
+ String.format("Restore snapshot '%s' doesn't exist.", snapshotId));
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index db2777a3a0fc..b999168f2cbb 100644
--- a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -69,6 +69,7 @@ org.apache.paimon.flink.procedure.DropPartitionProcedure
org.apache.paimon.flink.procedure.MergeIntoProcedure
org.apache.paimon.flink.procedure.ResetConsumerProcedure
org.apache.paimon.flink.procedure.RollbackToProcedure
+org.apache.paimon.flink.procedure.RestoreAsLatestProcedure
org.apache.paimon.flink.procedure.RollbackToTimestampProcedure
org.apache.paimon.flink.procedure.RollbackToWatermarkProcedure
org.apache.paimon.flink.procedure.MigrateTableProcedure
@@ -106,4 +107,4 @@ org.apache.paimon.flink.procedure.DataEvolutionMergeIntoProcedure
org.apache.paimon.flink.procedure.ReassignRowIdProcedure
org.apache.paimon.flink.procedure.CreateGlobalIndexProcedure
org.apache.paimon.flink.procedure.VectorSearchProcedure
-org.apache.paimon.flink.procedure.DropGlobalIndexProcedure
\ No newline at end of file
+org.apache.paimon.flink.procedure.DropGlobalIndexProcedure
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
new file mode 100644
index 000000000000..a7c83f596690
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
@@ -0,0 +1,97 @@
+/*
+ * 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.procedure;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** IT cases for restore_as_latest procedure. */
+public class RestoreAsLatestProcedureITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testRestoreSnapshotAsLatest() throws Exception {
+ sql("CREATE TABLE T (id INT, name STRING)");
+
+ FileStoreTable table = paimonTable("T");
+ SnapshotManager snapshotManager = table.snapshotManager();
+
+ commitRow(table, 1, "a");
+ commitRow(table, 2, "b");
+ commitRow(table, 3, "c");
+ assertEquals(3, snapshotManager.latestSnapshotId());
+
+ assertThat(sql("CALL sys.restore_as_latest(`table` => 'default.T', snapshot_id => 1)"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+
+ assertEquals(4, snapshotManager.latestSnapshotId());
+ assertTrue(snapshotManager.snapshotExists(2));
+ assertTrue(snapshotManager.snapshotExists(3));
+ assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "a"));
+
+ commitRow(table, 4, "d");
+ assertEquals(5, snapshotManager.latestSnapshotId());
+ assertThat(sql("SELECT * FROM T"))
+ .containsExactlyInAnyOrder(Row.of(1, "a"), Row.of(4, "d"));
+ }
+
+ @Test
+ public void testRestoreTagAsLatest() throws Exception {
+ sql("CREATE TABLE T (id INT, name STRING)");
+
+ FileStoreTable table = paimonTable("T");
+ SnapshotManager snapshotManager = table.snapshotManager();
+
+ commitRow(table, 1, "a");
+ commitRow(table, 2, "b");
+ commitRow(table, 3, "c");
+ assertEquals(3, snapshotManager.latestSnapshotId());
+
+ sql("CALL sys.create_tag(`table` => 'default.T', tag => 'tag-1', snapshot_id => 1)");
+
+ assertThat(sql("CALL sys.restore_as_latest(`table` => 'default.T', tag => 'tag-1')"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+
+ assertEquals(4, snapshotManager.latestSnapshotId());
+ assertTrue(snapshotManager.snapshotExists(2));
+ assertTrue(snapshotManager.snapshotExists(3));
+ assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "a"));
+ }
+
+ private void commitRow(FileStoreTable table, int id, String name) throws Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(id, BinaryString.fromString(name)));
+ commit.commit(write.prepareCommit());
+ }
+ }
+}
From 1c523acd9cd3e5530c0771f36abe65620c4a232c Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Mon, 8 Jun 2026 10:14:17 +0800
Subject: [PATCH 2/8] [flink] Fix restore_as_latest overwrite delta
Ensure restore_as_latest writes an overwrite delta so streaming overwrite readers can observe restored file changes.
---
.../paimon/operation/FileStoreCommitImpl.java | 56 +++++++++++++++++--
.../RestoreAsLatestProcedureITCase.java | 36 +++++++++++-
2 files changed, 84 insertions(+), 8 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index dd406d5c0dc2..455efb1fcf1e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/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 latestEntries = new HashMap<>();
+ FileEntry.mergeEntries(
+ manifestFile,
+ manifestList.readDataManifests(latest),
+ latestEntries,
+ options.scanManifestParallelism());
+
+ latestEntries.entrySet().removeIf(entry -> entry.getValue().kind() != FileKind.ADD);
+
+ Map targetEntries = new HashMap<>();
+ FileEntry.mergeEntries(
+ manifestFile,
+ manifestList.readDataManifests(targetSnapshot),
+ targetEntries,
+ options.scanManifestParallelism());
+ targetEntries.entrySet().removeIf(entry -> entry.getValue().kind() != FileKind.ADD);
+
+ List deltaFiles = new ArrayList<>();
+ for (Map.Entry 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 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 baseManifestList =
- manifestList.write(manifestList.readDataManifests(targetSnapshot));
- Pair emptyDeltaManifestList = manifestList.write(emptyList());
+ manifestList.write(manifestFile.write(new ArrayList<>(latestEntries.values())));
+ Pair 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() {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
index a7c83f596690..e2bf7fcb661f 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
+++ b/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 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();
From b2a52ed89fbb0b28948d202fe5341f6109fc6bd7 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Mon, 22 Jun 2026 00:30:10 +0800
Subject: [PATCH 3/8] [flink] Fix restore_as_latest expiration and row id
regression
Skip automatic expiration on the restore-as-latest path so it no longer
deletes the snapshots/tags it promises to keep (e.g. with
snapshot.num-retained.max=1), and keep nextRowId monotonic by taking the
max of the previous latest and target snapshot, preventing row id reuse
that breaks _ROW_ID global uniqueness on row-tracking tables.
Add IT cases covering both fixes.
---
.../paimon/operation/FileStoreCommitImpl.java | 18 +++++-
.../paimon/table/sink/TableCommitImpl.java | 7 ++-
.../RestoreAsLatestProcedureITCase.java | 55 +++++++++++++++++++
3 files changed, 78 insertions(+), 2 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 455efb1fcf1e..e107914635ee 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1218,6 +1218,11 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
Pair baseManifestList =
manifestList.write(manifestFile.write(new ArrayList<>(latestEntries.values())));
Pair deltaManifestList = manifestList.write(manifestFile.write(deltaFiles));
+ // For row-tracking tables nextRowId must stay monotonic: restoring an older snapshot must
+ // not move it backwards, otherwise new appends would reuse row ids already assigned by the
+ // snapshots between the target and the previous latest, breaking the global uniqueness of
+ // _ROW_ID. Keep the larger of the previous latest and the target nextRowId.
+ Long nextRowId = maxNextRowId(latest.nextRowId(), targetSnapshot.nextRowId());
Snapshot newSnapshot =
new Snapshot(
latest.id() + 1,
@@ -1239,11 +1244,22 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
targetSnapshot.watermark(),
targetSnapshot.statistics(),
targetSnapshot.properties(),
- targetSnapshot.nextRowId());
+ nextRowId);
return commitSnapshotImpl(newSnapshot, new ArrayList<>(PartitionEntry.merge(deltaFiles)));
}
+ @Nullable
+ private static Long maxNextRowId(@Nullable Long left, @Nullable Long right) {
+ if (left == null) {
+ return right;
+ }
+ if (right == null) {
+ return left;
+ }
+ return Math.max(left, right);
+ }
+
public void compactManifest() {
int retryCount = 0;
long startMillis = System.currentTimeMillis();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index 167394f18b20..849b25d54e8b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -208,7 +208,12 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
checkCommitted();
boolean success = commit.restoreAsLatest(targetSnapshot);
if (success) {
- maintain(COMMIT_IDENTIFIER, maintainExecutor, true);
+ // Skip automatic expiration for the restore path. Restore-as-latest promises not to
+ // delete snapshots or tags whose snapshot id is larger than the restored snapshot, but
+ // the newly committed latest snapshot would otherwise let expiration (e.g. a low
+ // snapshot.num-retained.max) immediately remove the restored snapshot and the later
+ // snapshots/tags it is meant to preserve.
+ maintain(COMMIT_IDENTIFIER, maintainExecutor, false);
}
return success;
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
index e2bf7fcb661f..d5a977d4e39e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
@@ -102,6 +102,61 @@ public void testRestoreTagAsLatest() throws Exception {
assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "a"));
}
+ @Test
+ public void testRestoreDoesNotExpireKeptSnapshots() throws Exception {
+ sql("CREATE TABLE T (id INT, name STRING)");
+
+ FileStoreTable table = paimonTable("T");
+ SnapshotManager snapshotManager = table.snapshotManager();
+
+ commitRow(table, 1, "a");
+ commitRow(table, 2, "b");
+ commitRow(table, 3, "c");
+ assertEquals(3, snapshotManager.latestSnapshotId());
+
+ // Configure aggressive expiration that would otherwise drop every snapshot but the latest.
+ sql(
+ "ALTER TABLE T SET ("
+ + "'snapshot.num-retained.min' = '1', "
+ + "'snapshot.num-retained.max' = '1')");
+
+ assertThat(sql("CALL sys.restore_as_latest(`table` => 'default.T', snapshot_id => 1)"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+
+ // Restore-as-latest must not delete snapshots whose id is larger than the restored one,
+ // even with snapshot.num-retained.max = 1. The restore path skips automatic expiration.
+ assertEquals(4, snapshotManager.latestSnapshotId());
+ assertTrue(snapshotManager.snapshotExists(1));
+ assertTrue(snapshotManager.snapshotExists(2));
+ assertTrue(snapshotManager.snapshotExists(3));
+ }
+
+ @Test
+ public void testRestoreKeepsRowIdMonotonic() throws Exception {
+ sql("CREATE TABLE RT (id INT, name STRING) WITH ('row-tracking.enabled' = 'true')");
+
+ FileStoreTable table = paimonTable("RT");
+ SnapshotManager snapshotManager = table.snapshotManager();
+
+ commitRow(table, 1, "a");
+ commitRow(table, 2, "b");
+ commitRow(table, 3, "c");
+ assertEquals(3, snapshotManager.latestSnapshotId());
+
+ Long latestNextRowId = snapshotManager.latestSnapshot().nextRowId();
+ assertThat(latestNextRowId).isNotNull();
+
+ // Restore the first snapshot, whose own nextRowId is smaller than the current latest.
+ assertThat(sql("CALL sys.restore_as_latest(`table` => 'default.RT', snapshot_id => 1)"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+
+ // nextRowId must not move backwards: otherwise new appends would reuse row ids already
+ // assigned by snapshots 2 and 3, breaking the global uniqueness of _ROW_ID. The restore
+ // snapshot keeps the larger of the previous latest and the target nextRowId.
+ Snapshot restored = table.snapshot(4);
+ assertThat(restored.nextRowId()).isEqualTo(latestNextRowId);
+ }
+
private void assertRestoreDelta(
FileStoreTable table,
long snapshotId,
From 255ecf73032f2fbc6a42424d06ff059036643040 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Tue, 23 Jun 2026 22:21:30 +0800
Subject: [PATCH 4/8] [flink] Trigger commit callbacks on restore_as_latest
restoreAsLatest committed the restore snapshot directly via
commitSnapshotImpl, bypassing the commit callbacks that a regular commit
runs. External views that depend on those callbacks (Iceberg
compatibility metadata, chain-table overwrite handling) could therefore
stay at the pre-restore state.
Notify the callbacks after a successful restore using the restored
base/delta/index files and the new snapshot. The index changes are
derived from the previous latest and target index manifests, mirroring
how the data delta files are computed.
Add an IT case asserting Iceberg metadata is generated for the restore
snapshot.
---
.../paimon/operation/FileStoreCommitImpl.java | 50 ++++++++++++++++++-
.../RestoreAsLatestProcedureITCase.java | 26 ++++++++++
2 files changed, 75 insertions(+), 1 deletion(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index e107914635ee..9718a92b8ecf 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1246,7 +1246,55 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
targetSnapshot.properties(),
nextRowId);
- return commitSnapshotImpl(newSnapshot, new ArrayList<>(PartitionEntry.merge(deltaFiles)));
+ boolean success =
+ commitSnapshotImpl(newSnapshot, new ArrayList<>(PartitionEntry.merge(deltaFiles)));
+ if (success) {
+ // Like a regular commit, notify the commit callbacks so external views stay in sync
+ // with the restored state (e.g. Iceberg compatibility metadata and chain-table
+ // overwrite handling). The restore is an overwrite from the previous latest to the
+ // target, so the delta files and index changes describe the transition the callbacks
+ // need.
+ List indexChanges = restoreIndexChanges(latest, targetSnapshot);
+ CommitCallback.Context context =
+ new CommitCallback.Context(
+ SimpleFileEntry.from(new ArrayList<>(latestEntries.values())),
+ deltaFiles,
+ indexChanges,
+ newSnapshot,
+ newSnapshot.commitIdentifier());
+ commitCallbacks.forEach(callback -> callback.call(context));
+ }
+ return success;
+ }
+
+ /**
+ * Computes the index file changes between the previous latest snapshot and the restore target,
+ * mirroring how the data delta files are derived: entries that only exist in the previous latest
+ * are marked as {@link FileKind#DELETE}, entries that only exist in the target are kept as ADD.
+ */
+ private List restoreIndexChanges(Snapshot latest, Snapshot target) {
+ Set latestIndexEntries = readIndexEntries(latest.indexManifest());
+ Set targetIndexEntries = readIndexEntries(target.indexManifest());
+
+ List indexChanges = new ArrayList<>();
+ for (IndexManifestEntry entry : latestIndexEntries) {
+ if (!targetIndexEntries.contains(entry)) {
+ indexChanges.add(entry.toDeleteEntry());
+ }
+ }
+ for (IndexManifestEntry entry : targetIndexEntries) {
+ if (!latestIndexEntries.contains(entry)) {
+ indexChanges.add(entry);
+ }
+ }
+ return indexChanges;
+ }
+
+ private Set readIndexEntries(@Nullable String indexManifest) {
+ if (indexManifest == null) {
+ return Collections.emptySet();
+ }
+ return new HashSet<>(indexManifestFile.read(indexManifest));
}
@Nullable
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
index d5a977d4e39e..c89d58e6496f 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RestoreAsLatestProcedureITCase.java
@@ -22,6 +22,7 @@
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.table.FileStoreTable;
@@ -157,6 +158,31 @@ public void testRestoreKeepsRowIdMonotonic() throws Exception {
assertThat(restored.nextRowId()).isEqualTo(latestNextRowId);
}
+ @Test
+ public void testRestoreTriggersCommitCallback() throws Exception {
+ sql(
+ "CREATE TABLE T (id INT, name STRING) WITH ("
+ + "'metadata.iceberg.storage' = 'table-location')");
+
+ FileStoreTable table = paimonTable("T");
+ SnapshotManager snapshotManager = table.snapshotManager();
+
+ commitRow(table, 1, "a");
+ commitRow(table, 2, "b");
+ commitRow(table, 3, "c");
+ assertEquals(3, snapshotManager.latestSnapshotId());
+
+ assertThat(sql("CALL sys.restore_as_latest(`table` => 'default.T', snapshot_id => 1)"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+ assertEquals(4, snapshotManager.latestSnapshotId());
+
+ // The restore must trigger the commit callbacks like a regular commit, so external views
+ // stay in sync with the restored state. With Iceberg compatibility enabled, that means
+ // Iceberg metadata is generated for the restore snapshot.
+ Path icebergMetadata = new Path(table.location(), "metadata/v4.metadata.json");
+ assertTrue(table.fileIO().exists(icebergMetadata));
+ }
+
private void assertRestoreDelta(
FileStoreTable table,
long snapshotId,
From 41aacdb1807a9d9ad431dd8aec3949468179c203 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Tue, 23 Jun 2026 23:38:19 +0800
Subject: [PATCH 5/8] [flink] Apply spotless format to restore_as_latest
callback change
---
.../org/apache/paimon/operation/FileStoreCommitImpl.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 9718a92b8ecf..4ac12a72de41 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1269,8 +1269,9 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
/**
* Computes the index file changes between the previous latest snapshot and the restore target,
- * mirroring how the data delta files are derived: entries that only exist in the previous latest
- * are marked as {@link FileKind#DELETE}, entries that only exist in the target are kept as ADD.
+ * mirroring how the data delta files are derived: entries that only exist in the previous
+ * latest are marked as {@link FileKind#DELETE}, entries that only exist in the target are kept
+ * as ADD.
*/
private List restoreIndexChanges(Snapshot latest, Snapshot target) {
Set latestIndexEntries = readIndexEntries(latest.indexManifest());
From 7b99d1f1998a95724fa4dac98587b2c0998ad5be Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Wed, 24 Jun 2026 19:57:02 +0800
Subject: [PATCH 6/8] [flink] Run pre-commit callbacks on restore_as_latest
restoreAsLatest only ran the post-commit callbacks; it skipped the
pre-commit callbacks that a regular commit runs before the snapshot
becomes visible. ChainTableCommitPreCallback uses that hook to reject a
pure-DELETE overwrite on a chain-table snapshot branch that would drop a
snapshot partition still anchoring delta partitions. Since restore also
creates an OVERWRITE delta, restoring such a branch to an older state
could silently break the chain.
Invoke commitPreCallbacks before commitSnapshotImpl, sharing the restore
base/delta/index files with the post callbacks, so an unsafe restore is
aborted before the snapshot is created.
Add an IT case asserting the dangerous restore is rejected and the latest
snapshot is left unchanged.
---
.../paimon/operation/FileStoreCommitImpl.java | 24 +++++++++----
.../ChainTablePartitionExpireTest.java | 34 +++++++++++++++++++
2 files changed, 51 insertions(+), 7 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 4ac12a72de41..58795d3a7f2e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1246,18 +1246,28 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
targetSnapshot.properties(),
nextRowId);
+ // The restore is an overwrite from the previous latest to the target, so the base files,
+ // delta files and index changes describe the transition the callbacks need. These are
+ // shared by the pre- and post-commit callbacks below.
+ List baseFiles =
+ SimpleFileEntry.from(new ArrayList<>(latestEntries.values()));
+ List indexChanges = restoreIndexChanges(latest, targetSnapshot);
+
+ // Like a regular commit, run the pre-commit callbacks before the snapshot becomes visible.
+ // They may veto the restore by throwing (e.g. a chain-table snapshot branch rejects a
+ // pure-DELETE overwrite that would drop a snapshot partition still anchoring delta
+ // partitions), in which case the restore snapshot is never created.
+ commitPreCallbacks.forEach(
+ callback -> callback.call(baseFiles, deltaFiles, indexChanges, newSnapshot));
+
boolean success =
commitSnapshotImpl(newSnapshot, new ArrayList<>(PartitionEntry.merge(deltaFiles)));
if (success) {
- // Like a regular commit, notify the commit callbacks so external views stay in sync
- // with the restored state (e.g. Iceberg compatibility metadata and chain-table
- // overwrite handling). The restore is an overwrite from the previous latest to the
- // target, so the delta files and index changes describe the transition the callbacks
- // need.
- List indexChanges = restoreIndexChanges(latest, targetSnapshot);
+ // Notify the post-commit callbacks so external views stay in sync with the restored
+ // state (e.g. Iceberg compatibility metadata and chain-table overwrite handling).
CommitCallback.Context context =
new CommitCallback.Context(
- SimpleFileEntry.from(new ArrayList<>(latestEntries.values())),
+ baseFiles,
deltaFiles,
indexChanges,
newSnapshot,
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java
index 5ff5edbf71b4..8550edf358fc 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/ChainTablePartitionExpireTest.java
@@ -19,6 +19,7 @@
package org.apache.paimon.operation;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
@@ -594,6 +595,39 @@ private BinaryRow findGroupedPartition(FileStoreTable table, String region, Stri
"Partition " + region + "|" + dt + " not found"));
}
+ @Test
+ public void testRestoreRejectedWhenDroppingAnchorSnapshotPartition() throws Exception {
+ Path tablePath = tablePath("restore_reject_anchor");
+ createChainTable(tablePath, true);
+
+ FileStoreTable snapshotTable = loadTable(tablePath).switchToBranch("snapshot");
+ FileStoreTable deltaTable = loadTable(tablePath).switchToBranch("delta");
+
+ // snapshot branch: an unrelated group (US) first, then the CN anchor.
+ writeGrouped(snapshotTable, "US", "20250101", "v1"); // snapshot branch snapshot #1
+ writeGrouped(snapshotTable, "CN", "20250301", "v2"); // snapshot #2 adds CN/20250301
+
+ // delta branch: a CN delta that uses CN/20250301 as its only anchor.
+ writeGrouped(deltaTable, "CN", "20250315", "v3");
+
+ // Restoring the snapshot branch back to snapshot #1 would drop CN/20250301, the only anchor
+ // of the CN/20250315 delta. The pre-commit callback must reject this restore (same as a
+ // regular overwrite) instead of silently breaking the chain.
+ FileStoreTable snapshotBranch = loadTable(tablePath).switchToBranch("snapshot");
+ Snapshot target = snapshotBranch.snapshotManager().snapshot(1);
+ try (TableCommitImpl commit = snapshotBranch.newCommit(commitUser)) {
+ assertThatThrownBy(() -> commit.restoreAsLatest(target))
+ .hasMessageContaining("Snapshot partition cannot be dropped");
+ }
+ // The dangerous restore was aborted, so the latest snapshot is unchanged.
+ assertThat(
+ loadTable(tablePath)
+ .switchToBranch("snapshot")
+ .snapshotManager()
+ .latestSnapshotId())
+ .isEqualTo(2L);
+ }
+
private Path tablePath(String tableName) {
return new Path(tempDir.toUri().toString(), tableName);
}
From 0ccca3c502138aec9572ccc7de2721e460491967 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Wed, 24 Jun 2026 23:14:12 +0800
Subject: [PATCH 7/8] [flink] Include deletion-vector changes in
restore_as_latest delta
The restore delta was built only from data-file identifier differences.
A deletion-vector-only delete/update does not rewrite the data file, so
such a change was missed, leaving an empty data delta with
deltaRecordCount 0. Streaming overwrite readers read the DELTA data
manifests and do not load DV indexes, so they could skip the restore
entirely.
Compare the deletion vectors (via index-manifest dvRanges) between the
previous latest and the target, and re-emit each data file whose DV
changed as DELETE(latest) + ADD(target). The physical row count is
unchanged, so the pair nets to zero in the delta record count, matching
the unchanged totalRecordCount.
Add a FileStoreCommitTest covering a DV-only restore transition.
---
.../paimon/operation/FileStoreCommitImpl.java | 130 ++++++++++++++++++
.../paimon/operation/FileStoreCommitTest.java | 32 +++++
2 files changed, 162 insertions(+)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 58795d3a7f2e..367b5efd8fee 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -26,6 +26,7 @@
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
import org.apache.paimon.manifest.FileEntry;
@@ -89,6 +90,7 @@
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -1215,6 +1217,16 @@ public boolean restoreAsLatest(Snapshot targetSnapshot) {
}
}
+ // Include data files whose deletion vector changed between the previous latest and the
+ // target while the data file itself stayed the same. A DV-only delete/update does not
+ // rewrite the data file, so the identifier diff above misses it and leaves an empty data
+ // delta that streaming overwrite readers would skip. Re-emitting such files as
+ // DELETE(latest) + ADD(target) makes the restore transition visible. The physical row count
+ // is unchanged, so the pair nets to zero in the delta record count, consistent with the
+ // unchanged totalRecordCount.
+ addDeletionVectorOnlyChanges(
+ latest, targetSnapshot, latestEntries, targetEntries, deltaFiles);
+
Pair baseManifestList =
manifestList.write(manifestFile.write(new ArrayList<>(latestEntries.values())));
Pair deltaManifestList = manifestList.write(manifestFile.write(deltaFiles));
@@ -1308,6 +1320,124 @@ private Set readIndexEntries(@Nullable String indexManifest)
return new HashSet<>(indexManifestFile.read(indexManifest));
}
+ /**
+ * Adds, to {@code deltaFiles}, data files whose deletion vector differs between the previous
+ * latest and the target snapshot but whose data file itself is unchanged (i.e. a DV-only
+ * delete/update). Such files are present in both snapshots, so the data-file identifier diff
+ * does not emit them; without this, a DV-only restore produces an empty data delta that
+ * streaming overwrite readers skip. They are re-emitted as DELETE(latest) + ADD(target).
+ */
+ private void addDeletionVectorOnlyChanges(
+ Snapshot latest,
+ Snapshot target,
+ Map latestEntries,
+ Map targetEntries,
+ List deltaFiles) {
+ Map latestDvs =
+ readDeletionVectors(latest.indexManifest());
+ Map targetDvs =
+ readDeletionVectors(target.indexManifest());
+ if (latestDvs.isEmpty() && targetDvs.isEmpty()) {
+ return;
+ }
+
+ Map latestByFile =
+ indexByDataFile(latestEntries.values());
+ Map targetByFile =
+ indexByDataFile(targetEntries.values());
+
+ Set keys = new HashSet<>(latestDvs.keySet());
+ keys.addAll(targetDvs.keySet());
+ for (DeletionVectorKey key : keys) {
+ if (Objects.equals(latestDvs.get(key), targetDvs.get(key))) {
+ continue;
+ }
+ ManifestEntry latestEntry = latestByFile.get(key);
+ ManifestEntry targetEntry = targetByFile.get(key);
+ // Only handle DV-only changes here: the data file must be present in both snapshots. A
+ // file added/removed across the restore is already in deltaFiles via the identifier
+ // diff.
+ if (latestEntry != null && targetEntry != null) {
+ deltaFiles.add(toDeltaEntry(FileKind.DELETE, latestEntry));
+ deltaFiles.add(toDeltaEntry(FileKind.ADD, targetEntry));
+ }
+ }
+ }
+
+ private Map readDeletionVectors(
+ @Nullable String indexManifest) {
+ if (indexManifest == null) {
+ return Collections.emptyMap();
+ }
+ Map result = new HashMap<>();
+ for (IndexManifestEntry entry : indexManifestFile.read(indexManifest)) {
+ if (entry.kind() != FileKind.ADD
+ || !DELETION_VECTORS_INDEX.equals(entry.indexFile().indexType())) {
+ continue;
+ }
+ LinkedHashMap dvRanges = entry.indexFile().dvRanges();
+ if (dvRanges == null) {
+ continue;
+ }
+ for (DeletionVectorMeta meta : dvRanges.values()) {
+ result.put(
+ new DeletionVectorKey(
+ entry.partition(), entry.bucket(), meta.dataFileName()),
+ meta);
+ }
+ }
+ return result;
+ }
+
+ private static Map indexByDataFile(
+ Collection entries) {
+ Map result = new HashMap<>();
+ for (ManifestEntry entry : entries) {
+ result.put(
+ new DeletionVectorKey(
+ entry.partition(), entry.bucket(), entry.file().fileName()),
+ entry);
+ }
+ return result;
+ }
+
+ private static ManifestEntry toDeltaEntry(FileKind kind, ManifestEntry entry) {
+ return ManifestEntry.create(
+ kind, entry.partition(), entry.bucket(), entry.totalBuckets(), entry.file());
+ }
+
+ /** Identifies a data file (partition, bucket, file name) for deletion-vector comparison. */
+ private static class DeletionVectorKey {
+ private final BinaryRow partition;
+ private final int bucket;
+ private final String dataFileName;
+
+ DeletionVectorKey(BinaryRow partition, int bucket, String dataFileName) {
+ this.partition = partition;
+ this.bucket = bucket;
+ this.dataFileName = dataFileName;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DeletionVectorKey that = (DeletionVectorKey) o;
+ return bucket == that.bucket
+ && Objects.equals(partition, that.partition)
+ && Objects.equals(dataFileName, that.dataFileName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(partition, bucket, dataFileName);
+ }
+ }
+
@Nullable
private static Long maxNextRowId(@Nullable Long left, @Nullable Long right) {
if (left == null) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index e13b11d474b6..251866454fd8 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -39,6 +39,7 @@
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.mergetree.compact.DeduplicateMergeFunction;
import org.apache.paimon.operation.commit.ConflictDetection;
import org.apache.paimon.operation.commit.RetryCommitResult;
@@ -889,6 +890,37 @@ public void testDVIndexFiles(boolean bitmap64) throws Exception {
assertThat(dvs.get("f2").isDeleted(3)).isTrue();
}
+ @Test
+ public void testRestoreIncludesDeletionVectorOnlyChanges() throws Exception {
+ TestAppendFileStore store = TestAppendFileStore.createAppendStore(tempDir, new HashMap<>());
+ BinaryRow partition = gen.getPartition(gen.next());
+
+ // snapshot 1: data files f1, f2, no deletion vectors
+ store.commit(store.writeDataFiles(partition, 0, Arrays.asList("f1", "f2")));
+ Snapshot target = store.snapshotManager().latestSnapshot();
+
+ // snapshot 2: DV-only change — add a deletion vector for f1, data files unchanged
+ store.commit(
+ store.writeDVIndexFiles(
+ partition, 0, Collections.singletonMap("f1", Arrays.asList(1, 3))));
+
+ // restore back to snapshot 1
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ assertThat(commit.restoreAsLatest(target)).isTrue();
+ }
+
+ // f1's deletion vector changed even though the data file is unchanged, so the restore delta
+ // must re-emit f1 as DELETE + ADD instead of being empty. An empty data delta would let
+ // streaming overwrite readers skip the restore entirely.
+ Snapshot restored = store.snapshotManager().latestSnapshot();
+ ManifestList manifestList = store.manifestListFactory().create();
+ List deltaManifests = manifestList.readDeltaManifests(restored);
+ long added = deltaManifests.stream().mapToLong(ManifestFileMeta::numAddedFiles).sum();
+ long deleted = deltaManifests.stream().mapToLong(ManifestFileMeta::numDeletedFiles).sum();
+ assertThat(added).isEqualTo(1);
+ assertThat(deleted).isEqualTo(1);
+ }
+
@Test
public void testManifestCompact() throws Exception {
TestFileStore store = createStore(false);
From ceb6cf4bfc4068adc36c057c29e33a25610c8f38 Mon Sep 17 00:00:00 2001
From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com>
Date: Thu, 2 Jul 2026 00:35:37 +0800
Subject: [PATCH 8/8] [flink] Rename restore_as_latest to rollback_to_as_latest
and drop DV delta workaround
Rename the procedure and its supporting API from restore_as_latest to
rollback_to_as_latest (RollbackToAsLatestProcedure, FileStoreCommit#rollbackToAsLatest,
docs, tests, output field rolled_back_snapshot_id). The operation is a
non-destructive rollback: it materializes an older snapshot/tag as a new
latest snapshot without deleting later snapshots or tags.
Also revert the deletion-vector delta workaround: for a DV-only difference
(data files identical, only the index manifest differs) the rollback delta
is left empty again instead of re-emitting the data file as DELETE+ADD.
Streaming incremental reads intentionally do not apply deletion vectors
(they are only streamable via a changelog producer), so the DELETE+ADD pair
would make streaming overwrite readers retract and re-add the full physical
file, which is worse than leaving the change invisible. File-level rollbacks
remain streaming-visible; batch and time-travel reads stay correct because
the snapshot points to the target's index manifest. Document the limitation.
Add characterization tests for the streaming visibility (file-level visible,
DV-only invisible).
---
docs/docs/flink/procedures.md | 30 ++--
docs/docs/maintenance/manage-snapshots.mdx | 18 +-
docs/docs/maintenance/manage-tags.mdx | 10 +-
.../paimon/operation/FileStoreCommit.java | 4 +-
.../paimon/operation/FileStoreCommitImpl.java | 156 ++----------------
.../paimon/table/sink/TableCommitImpl.java | 11 +-
.../ChainTablePartitionExpireTest.java | 13 +-
.../paimon/operation/FileStoreCommitTest.java | 107 ++++++++++--
....java => RollbackToAsLatestProcedure.java} | 29 ++--
.../org.apache.paimon.factories.Factory | 2 +-
...=> RollbackToAsLatestProcedureITCase.java} | 52 +++---
11 files changed, 193 insertions(+), 239 deletions(-)
rename paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/{RestoreAsLatestProcedure.java => RollbackToAsLatestProcedure.java} (80%)
rename paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/{RestoreAsLatestProcedureITCase.java => RollbackToAsLatestProcedureITCase.java} (78%)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index 325ce57235ed..907d183e212e 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -499,31 +499,33 @@ All available procedures are listed below.
|