This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1d717f01e0 [flink] Add rollback_to_as_latest procedure (#8139)
1d717f01e0 is described below
commit 1d717f01e040da5b7b41900b299cf93f0b727ec1
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Fri Jul 3 20:18:21 2026 +0800
[flink] Add rollback_to_as_latest procedure (#8139)
This PR adds a non-destructive rollback procedure for Flink:
```sql
CALL sys.rollback_to_as_latest(`table` => 'default.T', snapshot_id => 3);
CALL sys.rollback_to_as_latest(`table` => 'default.T', tag => 'tag-1');
```
Unlike `rollback_to`, this procedure rolls the table back to the state
of a target snapshot or tag by materializing it as a new latest
snapshot. Later snapshots and tags are preserved.
---
docs/docs/flink/procedures.md | 30 +++
docs/docs/maintenance/manage-snapshots.mdx | 29 ++-
docs/docs/maintenance/manage-tags.mdx | 19 ++
.../apache/paimon/operation/FileStoreCommit.java | 3 +
.../paimon/operation/FileStoreCommitImpl.java | 155 +++++++++++++++
.../apache/paimon/table/sink/TableCommitImpl.java | 15 ++
.../operation/ChainTablePartitionExpireTest.java | 35 ++++
.../paimon/operation/FileStoreCommitTest.java | 103 ++++++++++
.../procedure/RollbackToAsLatestProcedure.java | 134 +++++++++++++
.../services/org.apache.paimon.factories.Factory | 3 +-
.../RollbackToAsLatestProcedureITCase.java | 210 +++++++++++++++++++++
11 files changed, 734 insertions(+), 2 deletions(-)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index b8726ebbd0..a0e43efbc3 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -498,6 +498,36 @@ All available procedures are listed below.
CALL sys.rollback_to(`table` => 'default.T', snapshot_id => 10)
</td>
</tr>
+ <tr>
+ <td>rollback_to_as_latest</td>
+ <td>
+ -- for Flink 1.18<br/>
+ -- roll back to a snapshot as the latest snapshot<br/>
+ CALL [catalog.]sys.rollback_to_as_latest('identifier', cast(null as
string), snapshotId)<br/><br/>
+ -- roll back to a tag as the latest snapshot<br/>
+ CALL [catalog.]sys.rollback_to_as_latest('identifier', 'tagName',
cast(null as bigint))<br/><br/>
+ -- for Flink 1.19 and later<br/>
+ -- roll back to a snapshot as the latest snapshot<br/>
+ CALL [catalog.]sys.rollback_to_as_latest(`table` => 'identifier',
snapshot_id => snapshotId)<br/><br/>
+ -- roll back to a tag as the latest snapshot<br/>
+ CALL [catalog.]sys.rollback_to_as_latest(`table` => 'identifier', tag
=> 'tagName')
+ </td>
+ <td>
+ To roll a table back to a specific version and materialize it as the
latest snapshot, without deleting later
+ snapshots or tags. Batch and time-travel reads are correct; for
deletion-vector tables, a rollback whose only
+ difference is a deletion-vector change is not guaranteed to be
observed by streaming overwrite readers.
+ Argument:
+ <li>table: the target table identifier. Cannot be empty.</li>
+ <li>snapshotId (Long): id of the snapshot to roll back to.</li>
+ <li>tagName: name of the tag to roll back to.</li>
+ </td>
+ <td>
+ -- for Flink 1.18<br/>
+ CALL sys.rollback_to_as_latest('default.T', cast(null as string),
10)<br/><br/>
+ -- for Flink 1.19 and later<br/>
+ CALL sys.rollback_to_as_latest(`table` => 'default.T', snapshot_id =>
10)
+ </td>
+ </tr>
<tr>
<td>rollback_to_timestamp</td>
<td>
diff --git a/docs/docs/maintenance/manage-snapshots.mdx
b/docs/docs/maintenance/manage-snapshots.mdx
index db2d9095be..69b972cb26 100644
--- a/docs/docs/maintenance/manage-snapshots.mdx
+++ b/docs/docs/maintenance/manage-snapshots.mdx
@@ -362,6 +362,33 @@ CALL sys.rollback(table => 'database_name.table_name',
snapshot => snasphot_id);
</Tabs>
+## Rollback to Snapshot as Latest
+
+Roll a table back to the state of a specific snapshot id by materializing that
state as a new latest snapshot. Unlike
+`rollback_to`, this operation does not delete snapshots or tags whose snapshot
id is larger than the target snapshot.
+
+<Tabs groupId="rollback-to-as-latest">
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+Run the following command:
+
+```sql
+CALL sys.rollback_to_as_latest(`table` => 'database_name.table_name',
snapshot_id => <snapshot-id>);
+```
+
+</TabItem>
+
+</Tabs>
+
+:::note
+Batch and time-travel reads of the new latest snapshot are correct, because it
points to the target snapshot's index
+manifest. However, for tables using deletion vectors, a rollback whose only
difference from the current latest is a
+deletion-vector change (the data files are identical) produces an empty data
delta, so it is not guaranteed to be
+observed by streaming overwrite readers — the same as a plain deletion-vector
delete, which is only streamable through a
+changelog producer.
+:::
+
## Remove Orphan Files
Paimon files are deleted physically only when expiring snapshots. However, it
is possible that some unexpected errors occurred
@@ -411,4 +438,4 @@ The table can be `*` to clean all tables in the database.
</TabItem>
-</Tabs>
\ No newline at end of file
+</Tabs>
diff --git a/docs/docs/maintenance/manage-tags.mdx
b/docs/docs/maintenance/manage-tags.mdx
index 78e588e5f5..f53538c3a8 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');
</TabItem>
</Tabs>
+
+## Rollback to Tag as Latest
+
+Roll a table back to the state of a specific tag by materializing that state
as a new latest snapshot. Unlike
+`rollback_to`, this operation does not delete snapshots or tags whose snapshot
id is larger than the target tag.
+
+<Tabs groupId="rollback-to-as-latest">
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+Run the following command:
+
+```sql
+CALL sys.rollback_to_as_latest(`table` => 'database_name.table_name', tag =>
'tag_name');
+```
+
+</TabItem>
+
+</Tabs>
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 2dab92854c..42211d56ca 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
@@ -79,6 +79,9 @@ public interface FileStoreCommit extends AutoCloseable {
/** Compact the manifest entries only. */
void compactManifest();
+ /** Roll back to the target snapshot and materialize it as the latest
snapshot. */
+ boolean rollbackToAsLatest(Snapshot targetSnapshot);
+
/** Abort an unsuccessful commit. The data files will be deleted. */
void abort(List<CommitMessage> 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 6495d10c27..d5807e33d8 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
@@ -111,6 +111,7 @@ import static
org.apache.paimon.operation.commit.RowTrackingCommitUtils.assignRo
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}.
@@ -1287,6 +1288,160 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
return commitSnapshotImpl(newSnapshot, emptyList());
}
+ @Override
+ public boolean rollbackToAsLatest(Snapshot targetSnapshot) {
+ Snapshot latest =
+ checkNotNull(
+ snapshotManager.latestSnapshot(),
+ "Latest snapshot is null, can not roll back.");
+
+ 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(manifestFile.write(new
ArrayList<>(latestEntries.values())));
+ Pair<String, Long> deltaManifestList =
manifestList.write(manifestFile.write(deltaFiles));
+ // For row-tracking tables nextRowId must stay monotonic: a rollback
to 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,
+ targetSnapshot.schemaId(),
+ baseManifestList.getKey(),
+ baseManifestList.getRight(),
+ deltaManifestList.getKey(),
+ deltaManifestList.getRight(),
+ null,
+ null,
+ targetSnapshot.indexManifest(),
+ commitUser,
+ Long.MAX_VALUE,
+ CommitKind.OVERWRITE,
+ System.currentTimeMillis(),
+ targetSnapshot.totalRecordCount(),
+ recordCountAdd(deltaFiles) -
recordCountDelete(deltaFiles),
+ null,
+ targetSnapshot.watermark(),
+ targetSnapshot.statistics(),
+ targetSnapshot.properties(),
+ nextRowId);
+
+ // The rollback 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<SimpleFileEntry> baseFiles =
+ SimpleFileEntry.from(new ArrayList<>(latestEntries.values()));
+ List<IndexManifestEntry> indexChanges = rollbackIndexChanges(latest,
targetSnapshot);
+
+ // Like a regular commit, run the pre-commit callbacks before the
snapshot becomes visible.
+ // They may veto the rollback 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 rollback snapshot is never created.
+ commitPreCallbacks.forEach(
+ callback -> callback.call(baseFiles, deltaFiles, indexChanges,
newSnapshot));
+
+ boolean success =
+ commitSnapshotImpl(newSnapshot, new
ArrayList<>(PartitionEntry.merge(deltaFiles)));
+ if (success) {
+ // Notify the post-commit callbacks so external views stay in sync
with the rolled-back
+ // state (e.g. Iceberg compatibility metadata and chain-table
overwrite handling).
+ CommitCallback.Context context =
+ new CommitCallback.Context(
+ baseFiles,
+ 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 rollback 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<IndexManifestEntry> rollbackIndexChanges(Snapshot latest,
Snapshot target) {
+ Set<IndexManifestEntry> latestIndexEntries =
readIndexEntries(latest.indexManifest());
+ Set<IndexManifestEntry> targetIndexEntries =
readIndexEntries(target.indexManifest());
+
+ List<IndexManifestEntry> 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<IndexManifestEntry> readIndexEntries(@Nullable String
indexManifest) {
+ if (indexManifest == null) {
+ return Collections.emptySet();
+ }
+ return new HashSet<>(indexManifestFile.read(indexManifest));
+ }
+
+ @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 9ec742b354..e0ee31b21e 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
@@ -219,6 +219,21 @@ public class TableCommitImpl implements InnerTableCommit {
commit.compactManifest();
}
+ public boolean rollbackToAsLatest(Snapshot targetSnapshot) {
+ checkCommitted();
+ boolean success = commit.rollbackToAsLatest(targetSnapshot);
+ if (success) {
+ // Skip automatic expiration for the rollback path.
rollback_to_as_latest promises not
+ // to
+ // delete snapshots or tags whose snapshot id is larger than the
target snapshot, but
+ // the newly committed latest snapshot would otherwise let
expiration (e.g. a low
+ // snapshot.num-retained.max) immediately remove the rolled-back
snapshot and the later
+ // snapshots/tags it is meant to preserve.
+ maintain(COMMIT_IDENTIFIER, maintainExecutor, false);
+ }
+ return success;
+ }
+
private void checkCommitted() {
checkState(!batchCommitted, "BatchTableCommit only support one-time
committing.");
batchCommitted = true;
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 5ff5edbf71..0094addcf7 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,40 @@ public class ChainTablePartitionExpireTest {
"Partition " + region + "|" + dt + "
not found"));
}
+ @Test
+ public void
testRollbackToAsLatestRejectedWhenDroppingAnchorSnapshotPartition()
+ throws Exception {
+ Path tablePath = tablePath("rollback_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");
+
+ // Rolling back the snapshot branch to snapshot #1 would drop
CN/20250301, the only anchor
+ // of the CN/20250315 delta. The pre-commit callback must reject this
rollback (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.rollbackToAsLatest(target))
+ .hasMessageContaining("Snapshot partition cannot be
dropped");
+ }
+ // The dangerous rollback 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);
}
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 5a386947c1..e2210edea2 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
@@ -56,8 +56,12 @@ import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.ColStats;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.stats.StatsFileHandler;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.source.IncrementalSplit;
+import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
@@ -953,6 +957,105 @@ public class FileStoreCommitTest {
assertThat(dvs.get("f2").isDeleted(3)).isTrue();
}
+ @Test
+ public void testRollbackToAsLatestFileLevelDeleteIsVisibleToStreaming()
throws Exception {
+ // Contrast with the DV-only case: when a rollback removes whole data
files, the delete is
+ // file-level (FileKind.DELETE) and IS visible to streaming readers,
no DV needed.
+ TestAppendFileStore store =
TestAppendFileStore.createAppendStore(tempDir, new HashMap<>());
+ BinaryRow partition = gen.getPartition(gen.next());
+
+ // snapshot 1 (target): f1
+ store.commit(store.writeDataFiles(partition, 0,
Collections.singletonList("f1")));
+ Snapshot target = store.snapshotManager().latestSnapshot();
+
+ // snapshot 2 (latest): add f2
+ store.commit(store.writeDataFiles(partition, 0,
Collections.singletonList("f2")));
+
+ // roll back to snapshot 1 — f2 must be removed
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ assertThat(commit.rollbackToAsLatest(target)).isTrue();
+ }
+ Snapshot rolledBack = store.snapshotManager().latestSnapshot();
+
+ String root = TraceableFileIO.SCHEME + "://" + tempDir.toString();
+ FileStoreTable table = FileStoreTableFactory.create(store.fileIO(),
new Path(root));
+ List<Split> splits =
+
table.newSnapshotReader().withSnapshot(rolledBack).readChanges().splits();
+
+ // f2 is retracted (before side) — the file-level delete is visible to
streaming
+ assertThat(splits).isNotEmpty();
+ IncrementalSplit split = (IncrementalSplit) splits.get(0);
+ boolean f2Retracted = split.beforeFiles().stream().anyMatch(f ->
f.fileName().equals("f2"));
+ assertThat(f2Retracted).isTrue();
+ }
+
+ @Test
+ public void testNormalDeletionVectorDeleteIsInvisibleToStreamingDelta()
throws Exception {
+ // Baseline (no rollback involved): without a changelog producer, a
plain DV-only delete
+ // produces an empty data delta, so the streaming delta read sees no
change at all. DV
+ // deletes are only streamable via a changelog producer, not via the
delta/overwrite path.
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ TestAppendFileStore store =
TestAppendFileStore.createAppendStore(tempDir, options);
+ BinaryRow partition = gen.getPartition(gen.next());
+
+ // snapshot 1: data file f1, no deletion vectors
+ store.commit(store.writeDataFiles(partition, 0,
Collections.singletonList("f1")));
+
+ // snapshot 2: a DV-only delete on f1 (data file unchanged, only an
index increment)
+ store.commit(
+ store.writeDVIndexFiles(
+ partition, 0, Collections.singletonMap("f1",
Arrays.asList(1, 3))));
+ Snapshot dvDelete = store.snapshotManager().latestSnapshot();
+ assertThat(store.scanDVIndexFiles(partition, 0)).isNotEmpty();
+
+ String root = TraceableFileIO.SCHEME + "://" + tempDir.toString();
+ FileStoreTable table = FileStoreTableFactory.create(store.fileIO(),
new Path(root));
+ List<Split> splits =
+
table.newSnapshotReader().withSnapshot(dvDelete).readChanges().splits();
+ assertThat(splits).isEmpty();
+ }
+
+ @Test
+ public void
testRollbackToAsLatestDeletionVectorChangeIsInvisibleToStreaming()
+ throws Exception {
+ // A DV-only rollback changes only the index manifest (the data files
are identical), so it
+ // produces an empty data delta and is invisible to streaming readers
— consistent with a
+ // plain DV delete, which is also invisible to streaming (DV deletes
are only streamable via
+ // a changelog producer, not via the delta/overwrite path).
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ TestAppendFileStore store =
TestAppendFileStore.createAppendStore(tempDir, options);
+ BinaryRow partition = gen.getPartition(gen.next());
+
+ // snapshot 1: data file f1, no deletion vectors (the rollback target)
+ store.commit(store.writeDataFiles(partition, 0,
Collections.singletonList("f1")));
+ Snapshot target = store.snapshotManager().latestSnapshot();
+
+ // snapshot 2: DV-only change — add a deletion vector for f1, data
file unchanged
+ store.commit(
+ store.writeDVIndexFiles(
+ partition, 0, Collections.singletonMap("f1",
Arrays.asList(1, 3))));
+ // sanity check: f1 really has a deletion vector now
+ assertThat(store.scanDVIndexFiles(partition, 0)).isNotEmpty();
+
+ // snapshot 3: roll back to snapshot 1
+ try (FileStoreCommitImpl commit = store.newCommit()) {
+ assertThat(commit.rollbackToAsLatest(target)).isTrue();
+ }
+ Snapshot rolledBack = store.snapshotManager().latestSnapshot();
+
+ // The rollback's data delta is empty, so the streaming change read
produces no splits — the
+ // DV-only rollback is invisible to streaming (batch / time-travel
reads remain correct
+ // since
+ // the snapshot points to the target's index manifest).
+ String root = TraceableFileIO.SCHEME + "://" + tempDir.toString();
+ FileStoreTable table = FileStoreTableFactory.create(store.fileIO(),
new Path(root));
+ List<Split> splits =
+
table.newSnapshotReader().withSnapshot(rolledBack).readChanges().splits();
+ assertThat(splits).isEmpty();
+ }
+
@Test
public void testManifestCompact() throws Exception {
TestFileStore store = createStore(false);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedure.java
new file mode 100644
index 0000000000..af1acf618b
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedure.java
@@ -0,0 +1,134 @@
+/*
+ * 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;
+
+/**
+ * Rollback to as latest procedure. Usage:
+ *
+ * <pre><code>
+ * -- roll back to a snapshot as the latest snapshot
+ * CALL sys.rollback_to_as_latest(`table` => 'tableId', snapshot_id =>
snapshotId)
+ *
+ * -- roll back to a tag as the latest snapshot
+ * CALL sys.rollback_to_as_latest(`table` => 'tableId', tag => 'tagName')
+ * </code></pre>
+ */
+public class RollbackToAsLatestProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "rollback_to_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<previous_snapshot_id BIGINT, rolled_back_snapshot_id BIGINT,
current_snapshot_id BIGINT>")
+ 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 roll back.");
+
+ 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("rollback-to-as-latest-" +
UUID.randomUUID().toString())) {
+ Preconditions.checkState(
+ commit.rollbackToAsLatest(targetSnapshot),
+ "Failed to roll back to snapshot %s as latest.",
+ targetSnapshot.id());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to roll back to 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<Snapshot, List<String>> tags = store.newTagManager().tags();
+ for (Map.Entry<Snapshot, List<String>> entry : tags.entrySet()) {
+ if (entry.getKey().id() == snapshotId) {
+ return entry.getKey();
+ } else if (entry.getKey().id() > snapshotId) {
+ break;
+ }
+ }
+
+ throw new IllegalArgumentException(
+ String.format("Snapshot '%s' to roll back to 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 db2777a3a0..26a1a2b9cb 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.RollbackToAsLatestProcedure
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/RollbackToAsLatestProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedureITCase.java
new file mode 100644
index 0000000000..383e9334e0
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RollbackToAsLatestProcedureITCase.java
@@ -0,0 +1,210 @@
+/*
+ * 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.Snapshot;
+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;
+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 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;
+
+/** IT cases for rollback_to_as_latest procedure. */
+public class RollbackToAsLatestProcedureITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testRollbackToSnapshotAsLatest() 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.rollback_to_as_latest(`table` => 'default.T',
snapshot_id => 1)"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+
+ assertEquals(4, snapshotManager.latestSnapshotId());
+ assertRollbackDelta(table, 4, 0, 2, -2L);
+ assertTrue(snapshotManager.snapshotExists(2));
+ assertTrue(snapshotManager.snapshotExists(3));
+ assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "a"));
+
+ assertThat(sql("CALL sys.rollback_to_as_latest(`table` => 'default.T',
snapshot_id => 3)"))
+ .containsExactly(Row.of(4L, 3L, 5L));
+
+ assertEquals(5, snapshotManager.latestSnapshotId());
+ assertRollbackDelta(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(2, "b"), Row.of(3, "c"),
Row.of(4, "d"));
+ }
+
+ @Test
+ public void testRollbackToTagAsLatest() 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.rollback_to_as_latest(`table` => 'default.T',
tag => 'tag-1')"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+
+ assertEquals(4, snapshotManager.latestSnapshotId());
+ assertRollbackDelta(table, 4, 0, 2, -2L);
+ assertTrue(snapshotManager.snapshotExists(2));
+ assertTrue(snapshotManager.snapshotExists(3));
+ assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "a"));
+ }
+
+ @Test
+ public void testRollbackDoesNotExpireKeptSnapshots() 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.rollback_to_as_latest(`table` => 'default.T',
snapshot_id => 1)"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+
+ // rollback_to_as_latest must not delete snapshots whose id is larger
than the target one,
+ // even with snapshot.num-retained.max = 1. The rollback path skips
automatic expiration.
+ assertEquals(4, snapshotManager.latestSnapshotId());
+ assertTrue(snapshotManager.snapshotExists(1));
+ assertTrue(snapshotManager.snapshotExists(2));
+ assertTrue(snapshotManager.snapshotExists(3));
+ }
+
+ @Test
+ public void testRollbackKeepsRowIdMonotonic() 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();
+
+ // Roll back to the first snapshot, whose own nextRowId is smaller
than the current latest.
+ assertThat(sql("CALL sys.rollback_to_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 rollback
+ // snapshot keeps the larger of the previous latest and the target
nextRowId.
+ Snapshot rolledBack = table.snapshot(4);
+ assertThat(rolledBack.nextRowId()).isEqualTo(latestNextRowId);
+ }
+
+ @Test
+ public void testRollbackTriggersCommitCallback() 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.rollback_to_as_latest(`table` => 'default.T',
snapshot_id => 1)"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+ assertEquals(4, snapshotManager.latestSnapshotId());
+
+ // The rollback must trigger the commit callbacks like a regular
commit, so external views
+ // stay in sync with the rolled-back state. With Iceberg compatibility
enabled, that means
+ // Iceberg metadata is generated for the rollback snapshot.
+ Path icebergMetadata = new Path(table.location(),
"metadata/v4.metadata.json");
+ assertTrue(table.fileIO().exists(icebergMetadata));
+ }
+
+ private void assertRollbackDelta(
+ 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();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(id, BinaryString.fromString(name)));
+ commit.commit(write.prepareCommit());
+ }
+ }
+}
