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 0474b87e5b [core] Protect rollback-to-as-latest snapshots from expire
(#8450)
0474b87e5b is described below
commit 0474b87e5b427e032b1be5a3f92ff3e62802aae6
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Jul 3 22:33:17 2026 +0800
[core] Protect rollback-to-as-latest snapshots from expire (#8450)
Fix rollback-to-as-latest so files restored from an older snapshot are
protected from later expiration. When users roll back by `snapshot_id`,
the procedure now creates a dedicated non-expiring
`rollback-to-as-latest-<snapshotId>-<uuid>` tag and passes that `Tag`
into `TableCommitImpl`; when users roll back by `tag`, the provided tag
is used as-is.
---
.../apache/paimon/table/sink/TableCommitImpl.java | 5 +-
.../operation/ChainTablePartitionExpireTest.java | 9 ++-
.../paimon/operation/FileStoreCommitTest.java | 45 +++++++++++++
.../procedure/RollbackToAsLatestProcedure.java | 78 +++++++++++++++++++---
.../RollbackToAsLatestProcedureITCase.java | 53 +++++++++++++++
5 files changed, 177 insertions(+), 13 deletions(-)
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 e0ee31b21e..f5b8c7892b 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
@@ -32,6 +32,7 @@ import org.apache.paimon.operation.FileStoreCommit;
import org.apache.paimon.operation.PartitionExpire;
import org.apache.paimon.operation.metrics.CommitMetrics;
import org.apache.paimon.stats.Statistics;
+import org.apache.paimon.tag.Tag;
import org.apache.paimon.tag.TagAutoCreation;
import org.apache.paimon.tag.TagAutoManager;
import org.apache.paimon.tag.TagTimeExpire;
@@ -219,9 +220,9 @@ public class TableCommitImpl implements InnerTableCommit {
commit.compactManifest();
}
- public boolean rollbackToAsLatest(Snapshot targetSnapshot) {
+ public boolean rollbackToAsLatest(Tag targetTag) {
checkCommitted();
- boolean success = commit.rollbackToAsLatest(targetSnapshot);
+ boolean success =
commit.rollbackToAsLatest(targetTag.trimToSnapshot());
if (success) {
// Skip automatic expiration for the rollback path.
rollback_to_as_latest promises not
// to
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 0094addcf7..c8440a3506 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
@@ -616,8 +616,15 @@ public class ChainTablePartitionExpireTest {
// regular overwrite) instead of silently breaking the chain.
FileStoreTable snapshotBranch =
loadTable(tablePath).switchToBranch("snapshot");
Snapshot target = snapshotBranch.snapshotManager().snapshot(1);
+ String protectionTag = "rollback-to-as-latest-" + target.id() + "-" +
UUID.randomUUID();
+ snapshotBranch
+ .tagManager()
+ .createTag(target, protectionTag, null,
Collections.emptyList(), false);
try (TableCommitImpl commit = snapshotBranch.newCommit(commitUser)) {
- assertThatThrownBy(() -> commit.rollbackToAsLatest(target))
+ assertThatThrownBy(
+ () ->
+ commit.rollbackToAsLatest(
+
snapshotBranch.tagManager().getOrThrow(protectionTag)))
.hasMessageContaining("Snapshot partition cannot be
dropped");
}
// The dangerous rollback was aborted, so the latest snapshot is
unchanged.
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 e2210edea2..10bce28e94 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,10 +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.ExpireSnapshotsImpl;
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.sink.TableCommitImpl;
import org.apache.paimon.table.source.IncrementalSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.types.DataField;
@@ -81,6 +83,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -1056,6 +1059,48 @@ public class FileStoreCommitTest {
assertThat(splits).isEmpty();
}
+ @Test
+ public void testExpireSnapshotsKeepsFilesRestoredByRollbackToAsLatest()
throws Exception {
+ TestFileStore store = createStore(false, 1,
CoreOptions.ChangelogProducer.NONE);
+ KeyValue original = gen.nextInsert("20201110", 10, 1L, new int[] {1,
1}, "original");
+ KeyValue replacement = gen.nextInsert("20201111", 11, 2L, new int[]
{2, 2}, "replacement");
+ KeyValue retainedBeforeRollback =
+ gen.nextInsert("20201112", 12, 3L, new int[] {3, 3},
"retained");
+
+ Snapshot target =
+ store.commitData(Collections.singletonList(original),
gen::getPartition, kv -> 0)
+ .get(0);
+ String root = TraceableFileIO.SCHEME + "://" + tempDir.toString();
+ FileStoreTable table = FileStoreTableFactory.create(store.fileIO(),
new Path(root));
+ table.tagManager()
+ .createTag(
+ target, "expiring-target", Duration.ZERO,
Collections.emptyList(), false);
+ String protectionTag = "rollback-to-as-latest-" + target.id() + "-" +
UUID.randomUUID();
+ table.tagManager().createTag(target, protectionTag, null,
Collections.emptyList(), false);
+ store.overwriteData(
+ Collections.singletonList(replacement),
+ gen::getPartition,
+ kv -> 0,
+ Collections.emptyMap());
+ store.commitData(
+ Collections.singletonList(retainedBeforeRollback),
gen::getPartition, kv -> 0);
+
+ try (TableCommitImpl commit =
table.newCommit("rollback-to-as-latest-test")) {
+
assertThat(commit.rollbackToAsLatest(table.tagManager().getOrThrow(protectionTag)))
+ .isTrue();
+ }
+ Snapshot rolledBack = store.snapshotManager().latestSnapshot();
+ assertThat(table.tagManager().tags().get(target))
+ .contains("expiring-target")
+ .contains(protectionTag);
+
+ ((ExpireSnapshotsImpl) store.newExpire(1, 1,
Long.MAX_VALUE)).expireUntil(1, 3);
+
+ List<KeyValue> actual = store.readKvsFromSnapshot(rolledBack.id());
+ assertThat(store.toKvMap(actual))
+ .isEqualTo(store.toKvMap(Collections.singletonList(original)));
+ }
+
@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
index af1acf618b..46df2916c2 100644
---
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
@@ -25,9 +25,11 @@ 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.tag.Tag;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.StringUtils;
+import org.apache.paimon.utils.TagManager;
import org.apache.flink.table.annotation.ArgumentHint;
import org.apache.flink.table.annotation.DataTypeHint;
@@ -35,6 +37,7 @@ import org.apache.flink.table.annotation.ProcedureHint;
import org.apache.flink.table.procedure.ProcedureContext;
import org.apache.flink.types.Row;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
@@ -54,6 +57,7 @@ import java.util.UUID;
public class RollbackToAsLatestProcedure extends ProcedureBase {
public static final String IDENTIFIER = "rollback_to_as_latest";
+ private static final String ROLLBACK_TO_AS_LATEST_TAG_PREFIX =
"rollback-to-as-latest-";
@ProcedureHint(
argument = {
@@ -80,20 +84,39 @@ public class RollbackToAsLatestProcedure extends
ProcedureBase {
Preconditions.checkArgument(
hasTag != hasSnapshot, "Must specify exactly one of tag and
snapshot_id.");
+ TagManager tagManager = store.newTagManager();
+ Tag targetTag;
Snapshot targetSnapshot;
if (hasTag) {
- targetSnapshot =
store.newTagManager().getOrThrow(tagName).trimToSnapshot();
+ targetTag = tagManager.getOrThrow(tagName);
+ targetSnapshot = targetTag.trimToSnapshot();
} else {
- targetSnapshot = findSnapshot(store, snapshotId);
+ targetTag = null;
+ targetSnapshot = findSnapshot(store, tagManager, 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());
+ String createdRollbackTag = null;
+ String commitUser = "rollback-to-as-latest-" +
UUID.randomUUID().toString();
+ try {
+ if (!hasTag) {
+ createdRollbackTag = createRollbackToAsLatestTag(tagManager,
targetSnapshot);
+ targetTag = tagManager.getOrThrow(createdRollbackTag);
+ }
+
+ try (TableCommitImpl commit =
fileStoreTable.newCommit(commitUser)) {
+ boolean success = commit.rollbackToAsLatest(targetTag);
+ Preconditions.checkState(
+ success,
+ "Failed to roll back to snapshot %s as latest.",
+ targetSnapshot.id());
+ }
} catch (Exception e) {
+ try {
+ deleteCreatedRollbackTagIfNotCommitted(
+ store, tagManager, createdRollbackTag,
latestSnapshot.id() + 1, commitUser);
+ } catch (Exception cleanupException) {
+ e.addSuppressed(cleanupException);
+ }
throw new RuntimeException(
String.format(
"Failed to roll back to snapshot %s as latest.",
targetSnapshot.id()),
@@ -108,13 +131,48 @@ public class RollbackToAsLatestProcedure extends
ProcedureBase {
};
}
- private Snapshot findSnapshot(FileStore<?> store, long snapshotId) {
+ private String createRollbackToAsLatestTag(TagManager tagManager, Snapshot
targetSnapshot) {
+ String tagName =
+ ROLLBACK_TO_AS_LATEST_TAG_PREFIX
+ + targetSnapshot.id()
+ + "-"
+ + UUID.randomUUID().toString();
+ tagManager.createTag(targetSnapshot, tagName, null,
Collections.emptyList(), false);
+ return tagName;
+ }
+
+ private void deleteCreatedRollbackTagIfNotCommitted(
+ FileStore<?> store,
+ TagManager tagManager,
+ String createdRollbackTag,
+ long rollbackSnapshotId,
+ String commitUser) {
+ if (createdRollbackTag == null
+ || rollbackSnapshotCommitted(
+ store.snapshotManager(), rollbackSnapshotId,
commitUser)) {
+ return;
+ }
+
+ tagManager.deleteTag(
+ createdRollbackTag,
+ store.newTagDeletion(),
+ store.snapshotManager(),
+ Collections.emptyList());
+ }
+
+ private boolean rollbackSnapshotCommitted(
+ SnapshotManager snapshotManager, long rollbackSnapshotId, String
commitUser) {
+ return snapshotManager.snapshotExists(rollbackSnapshotId)
+ &&
commitUser.equals(snapshotManager.snapshot(rollbackSnapshotId).commitUser());
+ }
+
+ private Snapshot findSnapshot(FileStore<?> store, TagManager tagManager,
long snapshotId) {
SnapshotManager snapshotManager = store.snapshotManager();
if (snapshotManager.snapshotExists(snapshotId)) {
return snapshotManager.snapshot(snapshotId);
}
- SortedMap<Snapshot, List<String>> tags = store.newTagManager().tags();
+ SortedMap<Snapshot, List<String>> tags = tagManager.tags();
for (Map.Entry<Snapshot, List<String>> entry : tags.entrySet()) {
if (entry.getKey().id() == snapshotId) {
return entry.getKey();
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
index 383e9334e0..8e4e8a8103 100644
---
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
@@ -29,12 +29,18 @@ 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.tag.Tag;
+import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
import org.apache.flink.types.Row;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
+import java.util.Collections;
import java.util.List;
+import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -101,6 +107,47 @@ public class RollbackToAsLatestProcedureITCase extends
CatalogITCaseBase {
assertTrue(snapshotManager.snapshotExists(2));
assertTrue(snapshotManager.snapshotExists(3));
assertThat(sql("SELECT * FROM T")).containsExactly(Row.of(1, "a"));
+ assertThat(tagsOfSnapshot(table.tagManager(), 1))
+ .extracting(Pair::getRight)
+ .containsExactly("tag-1");
+ }
+
+ @Test
+ public void testRollbackCreatesNonExpiringProtectionTag() 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());
+
+ Snapshot target = snapshotManager.snapshot(1);
+ TagManager tagManager = table.tagManager();
+ tagManager.createTag(
+ target, "expiring-target", Duration.ZERO,
Collections.emptyList(), false);
+
+ assertThat(sql("CALL sys.rollback_to_as_latest(`table` => 'default.T',
snapshot_id => 1)"))
+ .containsExactly(Row.of(3L, 1L, 4L));
+
+ List<Pair<Tag, String>> targetTags = tagsOfSnapshot(tagManager,
target.id());
+ assertThat(targetTags)
+ .anyMatch(
+ tag ->
+ tag.getRight().equals("expiring-target")
+ && Duration.ZERO.equals(
+
tag.getLeft().getTagTimeRetained()));
+ assertThat(targetTags)
+ .anyMatch(
+ tag ->
+ tag.getRight()
+ .startsWith(
+
"rollback-to-as-latest-"
+ + target.id()
+ + "-")
+ && tag.getLeft().getTagTimeRetained()
== null);
}
@Test
@@ -199,6 +246,12 @@ public class RollbackToAsLatestProcedureITCase extends
CatalogITCaseBase {
assertThat(snapshot.deltaRecordCount()).isEqualTo(expectedDeltaRecordCount);
}
+ private List<Pair<Tag, String>> tagsOfSnapshot(TagManager tagManager, long
snapshotId) {
+ return tagManager.tagObjects().stream()
+ .filter(tag -> tag.getLeft().id() == snapshotId)
+ .collect(Collectors.toList());
+ }
+
private void commitRow(FileStoreTable table, int id, String name) throws
Exception {
BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
try (BatchTableWrite write = writeBuilder.newWrite();
