zhuxiangyi commented on code in PR #8139:
URL: https://github.com/apache/paimon/pull/8139#discussion_r3448726080
##########
paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java:
##########
@@ -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);
Review Comment:
Good catch, thanks. You're right — calling `maintain(..., true)` here lets
snapshot/tag/partition expiration run right after the restore, which can
immediately delete the very snapshots/tags that restore-as-latest promises to
keep (e.g. with `snapshot.num-retained.max=1`).
Fixed by skipping automatic expiration on this path: the restore now calls
`maintain(..., false)`, so no snapshot/tag/partition/consumer expiration is
triggered by the restore commit. Normal expiration resumes on the next regular
commit. (`tagAutoCreation`, which is not gated by `doExpire`, is
non-destructive and left as-is.)
Added an IT case `testRestoreDoesNotExpireKeptSnapshots` that sets
`snapshot.num-retained.min/max=1` and asserts the older snapshots still exist
after the restore.
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:
##########
@@ -1164,6 +1165,85 @@ 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.");
+
+ 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));
+ 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(),
+ targetSnapshot.nextRowId());
Review Comment:
Thanks, agreed — copying `targetSnapshot.nextRowId()` directly can move
`nextRowId` backwards when restoring an older snapshot, so subsequent appends
would reuse row ids already assigned by the snapshots between the target and
the previous latest, breaking the global uniqueness of `_ROW_ID`.
Fixed by keeping the larger of the previous latest and the target
`nextRowId`: `maxNextRowId(latest.nextRowId(), targetSnapshot.nextRowId())`
(null-safe, so non-row-tracking tables stay `null`). This keeps `nextRowId`
monotonic without leaving gaps in the id space.
Added an IT case `testRestoreKeepsRowIdMonotonic` on a
`row-tracking.enabled` table that restores an older snapshot and asserts the
restored snapshot's `nextRowId` equals the previous latest (not the smaller
target value).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
