huaxingao commented on code in PR #17754:
URL: https://github.com/apache/iceberg/pull/17754#discussion_r4051537521
##########
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java:
##########
@@ -908,19 +955,108 @@ 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.
+ *
+ * <p>A merge from an earlier commit attempt is undone when the concurrent
DV it merged is no
+ * longer present, which happens when the commit that added it is replaced
or rolled back.
+ */
+ 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 (String referencedDataFile :
ImmutableList.copyOf(mergedConcurrentDVsByFile.keySet())) {
+ MergedConcurrentDV merged =
mergedConcurrentDVsByFile.get(referencedDataFile);
+ DeleteFile concurrentDV =
newestDVByReferencedFile.get(referencedDataFile);
+ if (concurrentDV == null || !isSameDV(merged.concurrentDV(),
concurrentDV)) {
+ undoConcurrentDVMerge(referencedDataFile, merged);
+ }
+ }
+
+ for (Map.Entry<String, DeleteFile> entry :
newestDVByReferencedFile.entrySet()) {
+ String referencedDataFile = entry.getKey();
+ DeleteFile concurrentDV = entry.getValue();
+
+ if (mergedConcurrentDVsByFile.containsKey(referencedDataFile)) {
+ // already merged in a previous commit attempt
+ continue;
+ }
+
+ LOG.info(
+ "Merging concurrently added DV {} for {} into the new DV in table
{}",
+ concurrentDV.location(),
+ referencedDataFile,
+ tableName);
+
+ List<DeleteFile> droppedRemovals =
+ deleteFilterManager.filesToBeDeleted().stream()
+ .filter(file ->
referencedDataFile.equals(file.referencedDataFile()))
+ .collect(ImmutableList.toImmutableList());
+ droppedRemovals.forEach(deleteFilterManager::dropDelete);
+
+ DeleteFile pendingDV = Delegates.pendingDeleteFile(concurrentDV, null);
Review Comment:
nit: Maybe add a comment on why null instead of the concurrent DV's own
sequence number. `DVUtil.validateCanMerge` requires all DVs for a data file to
share a `dataSequenceNumber`, and this operation's DVs are pending (null), so
null is what makes the merge legal.
--
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]
