rdblue commented on code in PR #15006:
URL: https://github.com/apache/iceberg/pull/15006#discussion_r2719137541
##########
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java:
##########
@@ -1073,6 +1088,125 @@ private List<ManifestFile> newDeleteFilesAsManifests() {
return cachedNewDeleteManifests;
}
+ // Merge duplicates, internally takes care of updating newDeleteFilesBySpec
to remove
+ // duplicates and add the newly merged DV
+ private void mergeDVsAndWrite() {
+ Map<String, DeleteFileSet> dataFilesWithDuplicateDVs =
+ dvsByReferencedFile.entrySet().stream()
+ .filter(entry -> entry.getValue().size() > 1)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+ List<MergedDVContent> mergedDVs =
Collections.synchronizedList(Lists.newArrayList());
+ Tasks.foreach(dataFilesWithDuplicateDVs.entrySet())
+ .executeWith(ThreadPools.getDeleteWorkerPool())
+ .stopOnFailure()
+ .throwFailureWhenFinished()
+ .run(
+ entry -> {
+ String referencedLocation = entry.getKey();
+ DeleteFileSet duplicateDVs = entry.getValue();
+ mergedDVs.add(mergePositions(referencedLocation, duplicateDVs));
+ });
+
+ // Update newDeleteFilesBySpec to remove all the duplicates
+ mergedDVs.forEach(
+ mergedDV ->
newDeleteFilesBySpec.get(mergedDV.specId).removeAll(mergedDV.duplicateDVs));
+
+ writeMergedDVs(mergedDVs);
+ }
+
+ // Produces a Puffin per partition spec containing the merged DVs for that
spec
+ private void writeMergedDVs(List<MergedDVContent> mergedDVs) {
+ try (DVFileWriter dvFileWriter =
+ new BaseDVFileWriter(
+ // Use an unpartitioned spec for the location provider for the
puffin containing
+ // all the merged DVs
+ OutputFileFactory.builderFor(
+ ops(), PartitionSpec.unpartitioned(), FileFormat.PUFFIN,
1, 1)
+ .build(),
+ path -> null)) {
+
+ for (MergedDVContent mergedDV : mergedDVs) {
+ LOG.warn(
+ "Merged {} duplicate deletion vectors for data file {} in table
{}. The duplicate DVs are orphaned, and writers should merge DVs per file
before committing",
+ mergedDV.duplicateDVs.size(),
+ mergedDV.referencedLocation,
+ tableName);
+ dvFileWriter.delete(
+ mergedDV.referencedLocation,
+ mergedDV.mergedPositions,
+ spec(mergedDV.specId),
+ mergedDV.partition);
+ }
+
+ dvFileWriter.close();
+ DeleteWriteResult result = dvFileWriter.result();
+ result.deleteFiles().forEach(this::addPendingDelete);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private void addPendingDelete(DeleteFile file) {
+ newDeleteFilesBySpec
+ .get(file.specId())
+ .add(Delegates.pendingDeleteFile(file, file.dataSequenceNumber()));
+ }
+
+ // Data class for referenced file, the duplicate DVs, the merged position
delete index,
+ // partition spec and tuple
+ private static class MergedDVContent {
+ private final DeleteFileSet duplicateDVs;
+ private final String referencedLocation;
+ private final PositionDeleteIndex mergedPositions;
+ private final int specId;
+ private final StructLike partition;
+
+ MergedDVContent(
+ String referencedLocation,
+ DeleteFileSet duplicateDVs,
+ PositionDeleteIndex mergedPositions,
+ int specId,
+ StructLike partition) {
+ this.referencedLocation = referencedLocation;
+ this.duplicateDVs = duplicateDVs;
+ this.mergedPositions = mergedPositions;
+ this.specId = specId;
+ this.partition = partition;
+ }
+ }
+
+ // Merges the position indices for the duplicate DVs for a given referenced
file
+ private MergedDVContent mergePositions(String referencedLocation,
DeleteFileSet duplicateDVs) {
Review Comment:
Currently, this is called in tasks submitted to a threadpool, but I wonder
if there is a better model for parallelizing DV reads. We know that most DVs
that are being merged are going to be in different Puffin files, so each DV
read is going to be a separate IO to storage. If that's the case, why not use
more memory and run `Deletes.readDV` in a threadpool for all DV files, then
merge them in memory just before writing into Puffin? That way all of the reads
are parallelized rather than sequentially reading within a group.
--
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]
