JingsongLi commented on code in PR #8139:
URL: https://github.com/apache/paimon/pull/8139#discussion_r3484993122
##########
paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java:
##########
@@ -1164,6 +1167,288 @@ 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()));
+ }
+ }
+
+ // Include data files whose deletion vector changed between the
previous latest and the
+ // target while the data file itself stayed the same. A DV-only
delete/update does not
+ // rewrite the data file, so the identifier diff above misses it and
leaves an empty data
+ // delta that streaming overwrite readers would skip. Re-emitting such
files as
+ // DELETE(latest) + ADD(target) makes the restore transition visible.
The physical row count
+ // is unchanged, so the pair nets to zero in the delta record count,
consistent with the
+ // unchanged totalRecordCount.
+ addDeletionVectorOnlyChanges(
Review Comment:
This still does not make DV-only restores correct for streaming readers.
`readChanges()` always calls `toIncrementalPlan(true, ...)`, and
`SnapshotReaderImpl` only scans DV indexes when `!isStreaming`, so the
`IncrementalSplit` built for this DELETE+ADD pair has both
`beforeDeletionFiles` and `afterDeletionFiles` as null.
`IncrementalChangelogReadProvider` then reads the same physical data file on
both sides without applying either snapshot's DV: restoring from a DV snapshot
back to a no-DV snapshot emits a full-file retract plus a full-file add instead
of only adding the previously deleted rows (and the opposite direction
similarly fails to delete only the DV rows). The new test only checks that the
delta manifest is non-empty, so it does not catch this consumer-path behavior.
Please either make the restore streaming delta carry the snapshot-specific DV
files, or cover this with a streaming changelog/diff read test that proves the
logical rows produced by a DV-only restore are co
rrect.
--
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]
