Core: Detect and merge duplicate DVs for a data file and merge them before committing#15006
Core: Detect and merge duplicate DVs for a data file and merge them before committing#15006amogh-jahagirdar wants to merge 30 commits intoapache:mainfrom
Conversation
|
Still cleaning some stuff up, so leaving in draft but feel free to comment. But basically there are some cases in Spark where a file can be split across multiple tasks, and if deletes happen to touch every single part in the task we'd incorrectly produce multiple DVs for a given data file (discovered this recently with a user when they had Spark AQE enabled, but I think file splitting can happen in more cases). We currently throw on read in such cases, but ideally we can try and prevent this on write by detecting and merging pre-commit. The reason this is done behind the API is largely so that we are defensive from a library perspective that in case an engine/integration happens to produce multiple DVs, we can at least fix it up pre-commit. In the case there are too many to reasonably rewrite on a single node, then engines could do distributed writes to fix up before handing off the files to the API, but arguably from a library perspective it seems reasonable to pay this overhead to prevent bad commits across any integration. |
core/src/main/java/org/apache/iceberg/io/OutputFileFactory.java
Outdated
Show resolved
Hide resolved
| addedFilesSummary.addedFile(spec, file); | ||
| hasNewDeleteFiles = true; | ||
| if (ContentFileUtil.isDV(file)) { | ||
| newDVRefs.add(file.referencedDataFile()); |
There was a problem hiding this comment.
Probably keep a boolean in-case we detect a duplicate. That way we don't have to pay the price of grouping by referenced file everytime to detect possible duplicates; only if we detect it at the time of adding it, we can do the dedupe/merge
There was a problem hiding this comment.
We also could just keep a mapping specific for duplicates. That shrinks down how much work we need to do because instead of trying to group by every referenced data file in case of duplicates, we just go through the duplicates set. It's maybe a little more memory but if we consider that we expect duplicates to generally be rare it feels like a generally better solution
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
374b567 to
c04d0e0
Compare
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
| DeleteFileSet deleteFiles = | ||
| newDeleteFilesBySpec.computeIfAbsent(spec.specId(), ignored -> DeleteFileSet.create()); | ||
| if (deleteFiles.add(file)) { | ||
| addedFilesSummary.addedFile(spec, file); |
There was a problem hiding this comment.
because we may be merging duplicates, we don't update the summary for delete files until after we dedupe and are just about to write the new manifests
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
| Pair<List<PositionDelete<?>>, DeleteFile> deletesA = | ||
| deleteFile(tab, dataFileA, new Object[] {"aa"}, new Object[] {"a"}); | ||
| Pair<List<PositionDelete<?>>, DeleteFile> deletesB = | ||
| deleteFile(tab, dataFileA, new Object[] {"bb"}, new Object[] {"b"}); |
There was a problem hiding this comment.
This fix surfaced an issue in some of the TestPositionDeletesTable tests where we were setting the wrong data file for delete file; we'd just add a DV for the same data file, and then it'd get merged with the new logic , and break some of the later assertions.
| // Add Data Files with EQ and POS deletes | ||
| DeleteFile fileADeletes = fileADeletes(); | ||
| DeleteFile fileA2Deletes = fileA2Deletes(); | ||
| DeleteFile fileBDeletes = fileBDeletes(); |
There was a problem hiding this comment.
This test had to be fixed after the recent changes because the file paths for data file B and B2 were set to the same before, so the DVs for both referenced the same file (but that probably wasn't the intention of these tests) so it was a duplicate. After this change we'd merge the DVs in the commit, and then it'd actually get treated as a dangling delete and fail some of the assertions.
Since these tests are just testing the eq. delete case we could just simplify it by removing the usage of fileB deletes, it's a more minimal test that tests the same thing.
Also note, generally I'd take this in a separate PR but I think there's a good argument that this change should be in a 1.10.2 patch release to prevent invalid table states; in that case we'd need to keep these changes together.
| String referencedLocation = dvsToMergeForDataFile.getKey(); | ||
| mergedDVs.put( | ||
| referencedLocation, | ||
| mergeAndWriteDV(referencedLocation, dvsToMergeForDataFile.getValue())); |
There was a problem hiding this comment.
LIttle weird that we are now writing delete vectors on the driver?
spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestPositionDeletesTable.java
Show resolved
Hide resolved
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
| OutputFileFactory fileFactory = | ||
| OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); | ||
|
|
||
| DeleteFile deleteFile1 = dvWithPositions(dataFile, fileFactory, 0, 2); |
There was a problem hiding this comment.
Probably doesn't matter but in the real world these could also potentitally have existing overlapping deletes
Ie
Task 1 has existing DV and merges a few new Deletes
Task 2 has existing DV and merges a few new deletes
I think the logic is fine though
|
I'm mostly on board here but I have a few concerns,
|
| // Add position deletes for both partitions | ||
| Pair<List<PositionDelete<?>>, DeleteFile> deletesA = deleteFile(tab, dataFileA, "a"); | ||
| Pair<List<PositionDelete<?>>, DeleteFile> deletesB = deleteFile(tab, dataFileA, "b"); | ||
| Pair<List<PositionDelete<?>>, DeleteFile> deletesB = deleteFile(tab, dataFileB, "b"); |
There was a problem hiding this comment.
These tests are failing with the fix to core, right? I think we generally try to update one Spark version per commit unless there are failures that would leave main in a broken state.
There was a problem hiding this comment.
Yeah that's right, it's failing with the changes to core in this PR because the test was setting up duplicates (unintentionally as far as I can tell) and now we'd be merging them and breaking some of the assertions in the test.
core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java
Outdated
Show resolved
Hide resolved
f39d27e to
afdb314
Compare
| private final Map<Integer, DeleteFileSet> newDeleteFilesBySpec = Maps.newHashMap(); | ||
| private final Set<String> newDVRefs = Sets.newHashSet(); | ||
| private final List<DeleteFile> v2Deletes = Lists.newArrayList(); | ||
| private final Map<String, List<DeleteFile>> dvsByReferencedFile = Maps.newLinkedHashMap(); |
There was a problem hiding this comment.
I commented this elsehwere but the reason this field is a LinkedHashMap is because there's quite a lot of tests which verify the exact ordering of entries in a manifest. I do think we should change those tests because we really shouldn't be guaranteeing any ordering in the entries in the output manifests but that's a larger change. The current implementation which tracks newDeleteFilesBySpec uses a DeleteFileSet which is an insertion-ordering preserving structure.
There was a problem hiding this comment.
I am not sure about the ordering here, since I think that is something we may want to guarentee or at least make determinsitic in the future. We can always talk about that later though.
No problems with this usage here of LinkedHashMap
afdb314 to
1efc041
Compare
| return toPositionIndexes(posDeletes, null /* unknown delete file */); | ||
| } | ||
|
|
||
| public static PositionDeleteIndex readDV(DeleteFile deleteFile, FileIO fileIO) { |
There was a problem hiding this comment.
Should this be here or in DVUtil?
Deletes seems to be utilities for working with position delete structs coming from an engine. DVUtil seems more appropriate to me.
| * Merges duplicate DVs for the same data file and writes the merged DV Puffin files. If there is | ||
| * exactly 1 DV for a given data file then it is return as is | ||
| * | ||
| * @param dvsByFile map of data file location to DVs |
There was a problem hiding this comment.
Missing quite a few params here.
| */ | ||
| static List<DeleteFile> mergeAndWriteDVsIfRequired( | ||
| Map<String, List<DeleteFile>> dvsByFile, | ||
| Supplier<OutputFile> dvOutputFile, |
There was a problem hiding this comment.
Since this has a FileIO, it's a little awkward to have a Supplier<OutputFile> passed in since a big part of that is the job of FileIO. Also, if this supplier is only called once then it doesn't need to be a supplier. What about passing a string location instead?
| * @return a list containing both any newly merged DVs and any DVs that are already valid | ||
| */ | ||
| static List<DeleteFile> mergeAndWriteDVsIfRequired( | ||
| Map<String, List<DeleteFile>> dvsByFile, |
| for (int i = 0; i < duplicateDVPositions.length; i++) { | ||
| PositionDeleteIndex dvPositions = duplicateDVPositions[i]; | ||
| DeleteFile dv = duplicateDVs.get(i); | ||
| mergedIndexByFile.merge( |
There was a problem hiding this comment.
I think this is correct, but I find a more straightforward implementation that uses get instead of merge easier to read:
Map<String, PositionDeleteIndex> mergedDVs = Maps.newHashMap();
for (int i = 0; i < duplicatedDVPositions.length; i++) {
PositionDeleteIndex previousDV = mergedDVs.get(duplicateDVs.get(i).referencedDataFile());
if (previousDV != null) {
previousDV.merge(duplicatedDVPositions[i]);
} else {
mergedDVs.put(delete.referencedDataFile(), duplicatedDVPositions[i]);
}
}|
|
||
| List<DeleteFile> mergedDVs = Lists.newArrayList(); | ||
| Map<String, PartitionSpec> specByFile = Maps.newHashMap(); | ||
| Map<String, StructLike> partitionByFile = Maps.newHashMap(); |
There was a problem hiding this comment.
Rather than 2 maps and lookups, why not use Map<String, Pair<PartitionSpec, StructLike>>?
| FileIO fileIO, | ||
| Map<Integer, PartitionSpec> specs, | ||
| ExecutorService pool) { | ||
| Map<String, List<DeleteFile>> duplicateDVsByFile = |
There was a problem hiding this comment.
There are a couple of things that make this method fairly complicated. First, using a map of lists ends up causing this to do a fair amount of stream manipulation. Using a Guava multimap cleans that up quite a bit.
Second, this doesn't need to process the input map multiple times. This is cleaner if you process it once, keep the DVs that don't need to be merged in an output list, and also accumulate the partition and sequence number info for later at the same time. Doing that also simplifies the write and validate methods because you can identify the expected data sequence number and partition information here rather than complicating the loops in later methods.
Here's what I came up with:
static List<DeleteFile> mergeAndWriteDVsIfRequired(
Map<String, List<DeleteFile>> dvsByFile,
Supplier<OutputFile> dvOutputFile,
FileIO fileIO,
Map<Integer, PartitionSpec> specs,
ExecutorService pool) {
List<DeleteFile> finalDVs = Lists.newArrayList();
Multimap<String, DeleteFile> duplicates =
Multimaps.newListMultimap(Maps.newHashMap(), Lists::newArrayList);
Map<String, Pair<PartitionSpec, StructLike>> partitions = Maps.newHashMap();
for (Map.Entry<String, List<DeleteFile>> entry : dvsByFile.entrySet()) {
if (entry.getValue().size() > 1) {
duplicates.putAll(entry.getKey(), entry.getValue());
DeleteFile first = entry.getValue().get(0);
partitions.put(entry.getKey(), Pair.of(specs.get(first.specId()), first.partition()));
} else {
finalDVs.addAll(entry.getValue());
}
}
if (duplicates.isEmpty()) {
return finalDVs;
}
validateCanMerge(duplicates, partitions);
Map<String, PositionDeleteIndex> deletes =
readAndMergeDVs(duplicates.values().toArray(DeleteFile[]::new), fileIO, pool);
finalDVs.addAll(writeDVs(deletes, partitions, dvOutputFile));
return finalDVs;
}| cachedNewDeleteManifests.clear(); | ||
| // On cache invalidation of delete files, clear the whole summary. | ||
| // Since the summary contained both data files and DVs, add back the data files. | ||
| addedFilesSummary.clear(); |
There was a problem hiding this comment.
Is this an added files summary, or is it a commit summary? Maybe we should change the name to be more clear.
| // On cache invalidation of delete files, clear the whole summary. | ||
| // Since the summary contained both data files and DVs, add back the data files. | ||
| addedFilesSummary.clear(); | ||
| newDataFilesBySpec.forEach( |
There was a problem hiding this comment.
I think it would be better not to handle data files in newDeleteFilesAsManifests. It isn't obvious that this method is going to clear the data file summary and no one would look here for it. Why not keep a separate summary for data files and merge it with the deletes for the final summary? The data file one doesn't need to change.
| dvsByReferencedFile, | ||
| () -> { | ||
| String filename = FileFormat.PUFFIN.addExtension(String.valueOf(snapshotId())); | ||
| return fileIO.newOutputFile(ops().locationProvider().newDataLocation(filename)); |
There was a problem hiding this comment.
I think we should add a little more to this path. This doesn't show what the file is and will overwrite the same file name on each retry.
I would use an instance field so that we can embed a file counter:
private final AtomicInteger dvMergeAttempt = new AtomicInteger(0);
String filename = FileFormat.PUFFIN.addExtension(String.format("merged-dvs-%s-%s", snapshotId(), dvMergeAttempt.incrementAndGet()));| ThreadPools.getDeleteWorkerPool()); | ||
|
|
||
| return finalDVs.stream() | ||
| .map(file -> Delegates.pendingDeleteFile(file, file.dataSequenceNumber())) |
There was a problem hiding this comment.
As I noted, I'd prefer to remove the wrapper here.
|
|
||
| private PuffinWriter newWriter() { | ||
| EncryptedOutputFile outputFile = fileFactory.newOutputFile(); | ||
| OutputFile outputFile = dvOutputFile.get(); |
There was a problem hiding this comment.
I think there's an existing problem with this class. It doesn't need to be fixed right now but we should follow up.
The issue is that this doesn't handle encryption key metadata when it creates DeleteFile instances in createDV. I would expect this to pass the output file's encryption key metadata (if it is an EncryptedOutputFile) to the delete file builder via withEncryptionKeyMetadata. But that isn't called.
I think that this class will still create encrypted streams because it calls EncryptedOutputFile#encryptingOutputFile to produce output streams. It just doesn't get the key metadata and pass it on.
This isn't a blocker for this commit since it's an existing bug, but we shouldn't leave it this way because it could create DVs that are encrypted and unrecoverable.
@aokolnychyi, you may want to take a look.
afb63b6 to
eab1ca5
Compare
eab1ca5 to
380d09e
Compare
| protected boolean addsDeleteFiles() { | ||
| return !newDeleteFilesBySpec.isEmpty(); | ||
| return !v2Deletes.isEmpty() | ||
| || dvsByReferencedFile.values().stream().anyMatch(dvs -> !dvs.isEmpty()); |
There was a problem hiding this comment.
How do we get empty "dvs" here? Shouldn't it always have contents?
There was a problem hiding this comment.
Yes, it should. This is a defensive check.
| .collect(Collectors.toList())); | ||
| assertThat(committedEqDelete).isNotNull(); | ||
| assertThat(committedEqDelete.content()).isEqualTo(FileContent.EQUALITY_DELETES); | ||
| } |
There was a problem hiding this comment.
There are a few other validations we could test here as well, (since we spent the time to add them)
Same file name different specs
Sequence number mismatch
I forgot what else you added :)
There was a problem hiding this comment.
Asked claude,
Mismatched sequence numbers -- two DVs for the same data file have different dataSequenceNumber() values. The validation at line 110-115 should throw IllegalArgumentException but this is never exercised.
Mismatched spec IDs -- two DVs for the same data file reference different partition specs (line 117-122). Not tested.
Mismatched partition tuples -- same spec but different partition values (line 124-129). Not tested.
| private final SnapshotSummary.Builder addedFilesSummary = SnapshotSummary.builder(); | ||
| private final SnapshotSummary.Builder addedDataFilesSummary = SnapshotSummary.builder(); | ||
| private final SnapshotSummary.Builder addedDeleteFilesSummary = SnapshotSummary.builder(); | ||
|
|
There was a problem hiding this comment.
Nit: unnecessary whitespace change.
| // this triggers a rewrite of all delete manifests even if there is only one new delete file | ||
| // if there is a relevant use case in the future, the behavior can be optimized | ||
| cachedNewDeleteManifests.clear(); | ||
| // On cache invalidation of delete files, clear the summary. |
There was a problem hiding this comment.
I think this comment is obvious from the call below. I think the idea is right to have a comment here: why is it necessary to clear the summary? That's because the summary can't be generated until after DV merge happens and any new DV could require a merge. So the summary is built each time DV manifests are written.
rdblue
left a comment
There was a problem hiding this comment.
Looks good to me! Russell pointed out some test cases that should be added and I found a couple of minor things that are nice-to-have.
While generally, writers are expected to merge DVs for a given data file before attempting to commit, we probably want to have a safeguard in the commit path in case this assumption is violated. This has been observed when AQE is enabled in Spark and a data file is split across multiple tasks (really just depends on how files and deletes are split); then multiple DVs are produced for a given data file, and then committed. Currently, after that commit reads would fail since the DeleteFileIndex detects the duplicates and fails on read.
Arguably, there should be a safeguard on the commit path which detects duplicates and fixes them up to prevent any invalid table states. Doing this behind the API covers any engine integration using the library.
This change updates MergingSnapshotProducer to track duplicate DVs for a datafile, and then merge them and produces a Puffin file per DV. Note that since we generally expect duplicates to be rare, we don't expect there to be too many small Puffins produced, and we don't add the additional logic to coalesce into larger files. Furthermore, these can later be compacted. In case of large scale duplicates, then engines should arguably fix those up before handing off to the commit path.