huaxingao commented on code in PR #17754:
URL: https://github.com/apache/iceberg/pull/17754#discussion_r4034049102
##########
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java:
##########
@@ -908,19 +949,80 @@ private void validateAddedDVs(
for (ManifestEntry<DeleteFile> entry : entries) {
DeleteFile file = entry.file();
- if (newSnapshotIds.contains(entry.snapshotId()) &&
ContentFileUtil.isDV(file)) {
+ if (newSnapshotIds.contains(entry.snapshotId())
+ && ContentFileUtil.isDV(file)
+ && dvsByReferencedFile.containsKey(file.referencedDataFile())) {
ValidationException.check(
- !dvsByReferencedFile.containsKey(file.referencedDataFile()),
+ concurrentDVs != null &&
mergeableSnapshotIds.contains(entry.snapshotId()),
"Found concurrently added DV for %s: %s",
file.referencedDataFile(),
ContentFileUtil.dvDesc(file));
+ concurrentDVs.add(file.copy());
}
}
} catch (IOException e) {
throw new UncheckedIOException(e);
}
}
+ /**
+ * Merges concurrently added DVs into this operation's DVs.
+ *
+ * <p>The format requires a DV to include all deleted positions from the DV
it replaces, so the
+ * newest concurrent DV for a data file contains the content of all older
DVs for that file. That
+ * DV is removed from the table and its content is merged into this
operation's DV for the same
+ * data file. Pending removals of delete files for that data file are
dropped because the
+ * concurrent commits have already removed them.
+ */
+ private void mergeConcurrentDVs(List<DeleteFile> concurrentDVs) {
+ Map<String, DeleteFile> newestDVByReferencedFile = Maps.newHashMap();
+ for (DeleteFile dv : concurrentDVs) {
+ newestDVByReferencedFile.merge(
+ dv.referencedDataFile(),
+ dv,
+ (dv1, dv2) -> dv1.dataSequenceNumber() >= dv2.dataSequenceNumber() ?
dv1 : dv2);
+ }
+
+ for (Map.Entry<String, DeleteFile> entry :
newestDVByReferencedFile.entrySet()) {
+ String referencedDataFile = entry.getKey();
+ DeleteFile concurrentDV = entry.getValue();
+
+ DeleteFile previouslyMerged =
mergedConcurrentDVsByFile.get(referencedDataFile);
+ if (previouslyMerged != null) {
+ if (isSameDV(previouslyMerged, concurrentDV)) {
+ // already merged in a previous commit attempt
+ continue;
+ }
+
+ // the previously merged DV was replaced by a newer commit that merged
its content
+ dvsByReferencedFile.get(referencedDataFile).remove(previouslyMerged);
Review Comment:
I think this handles the concurrent DV being replaced between attempts, but
not it disappearing.
For example, a data file has rows 0-3. Alice deletes 0,1 and commits a DV.
Bob deletes 2,3. Bob's attempt 1 merges Alice's DV, so his DV now covers
0,1,2,3, then his commit fails and retries. In between, the branch is rolled
back to before Alice's commit.
Attempt 2 finds no concurrent DV, so line 933 skips mergeConcurrentDVs and
attempt 1's merge is never reconsidered — dvsByReferencedFile is never cleared,
and the retry calls apply() on the same instance. Bob still points at Alice's
DV file, which the rollback didn't delete, so he commits 0,1,2,3. Rows 0 and 1
end up deleted even though Alice's delete was rolled back and Bob never asked
for them.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
