JingsongLi commented on code in PR #8139:
URL: https://github.com/apache/paimon/pull/8139#discussion_r3464787760
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:
##########
@@ -1164,6 +1165,150 @@ 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);
+
+ boolean success =
Review Comment:
This restore path still bypasses the normal pre-commit callbacks. Regular
commits call commitPreCallbacks before commitSnapshotImpl, and
ChainTableCommitPreCallback uses that hook to reject unsafe pure-DELETE
overwrite commits on the snapshot branch. restoreAsLatest also creates an
OVERWRITE delta, so restoring a snapshot branch back to an older state can
delete snapshot partitions that a normal overwrite would validate and
potentially abort. Please run the same pre-callback path (with the restore base
files, delta files, index changes, and new snapshot) before making the restore
snapshot visible.
--
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]
