raunaqmorarka commented on code in PR #17754:
URL: https://github.com/apache/iceberg/pull/17754#discussion_r4035071761
##########
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:
Confirmed with a test, the rollback left the first attempt's merge in place.
A merge from an earlier attempt is now undone when its concurrent DV is no
longer present, which also restores the delete file removals the merge had
dropped. Added coverage for a rollback between attempts and for the branch
being removed.
--
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]
