zhuxiangyi commented on code in PR #8139:
URL: https://github.com/apache/paimon/pull/8139#discussion_r3460523402
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:
##########
@@ -1164,6 +1165,101 @@ 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));
+ // 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,
+ 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);
+
+ return commitSnapshotImpl(newSnapshot, new
ArrayList<>(PartitionEntry.merge(deltaFiles)));
Review Comment:
Good catch, thanks. `restoreAsLatest` committed via `commitSnapshotImpl`
directly and skipped the commit callbacks, so external views (Iceberg metadata,
chain-table overwrite) could stay at the pre-restore state.
Fixed by notifying the callbacks after a successful restore, like a regular
commit. The context uses the restore delta (DELETE previous-only + ADD
target-only) and an index delta derived the same way from the previous-latest
and target index manifests. Both callbacks are idempotent, so retries stay
correct.
Added an IT case `testRestoreTriggersCommitCallback` asserting Iceberg
metadata is generated for the restore snapshot.
--
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]
