diff --git a/core/src/main/java/org/apache/iceberg/BaseRowDelta.java b/core/src/main/java/org/apache/iceberg/BaseRowDelta.java index 4fa871811939..9032cd4624c1 100644 --- a/core/src/main/java/org/apache/iceberg/BaseRowDelta.java +++ b/core/src/main/java/org/apache/iceberg/BaseRowDelta.java @@ -174,13 +174,43 @@ protected void validate(TableMetadata base, Snapshot parent) { } /** - * Concurrently added DVs can be merged into this operation's DVs only when this operation does - * not add or remove data files and does not require concurrent delete files to fail via {@link - * #validateNoConflictingDeleteFiles()}. + * Concurrently added DVs can be merged into this operation's DVs when this operation does not + * remove data files and does not require concurrent delete files to fail via {@link + * #validateNoConflictingDeleteFiles()}. Removing data files makes concurrent positions on those + * files meaningless, so merging is disabled in that case. + * + *

Pure delete commits always merge concurrent DVs because deletes are commutative. Operations + * that also add data files (e.g. UPDATE or MERGE) can merge a concurrent DV only when its newly + * deleted positions are disjoint from this operation's own deleted positions (see {@link + * #mergedConcurrentDVsMustBeDisjoint()}), and only when {@link + * TableProperties#ROW_LEVEL_CONCURRENT_DV_MERGE_ENABLED} is enabled. */ @Override protected boolean canMergeConcurrentDVs() { - return !validateNewDeleteFiles && !addsDataFiles() && !deletesDataFiles(); + if (validateNewDeleteFiles || deletesDataFiles()) { + return false; + } + + if (addsDataFiles()) { + return ops() + .current() + .propertyAsBoolean( + TableProperties.ROW_LEVEL_CONCURRENT_DV_MERGE_ENABLED, + TableProperties.ROW_LEVEL_CONCURRENT_DV_MERGE_ENABLED_DEFAULT); + } + + return true; + } + + /** + * Deletes are commutative, so an operation that only adds DVs may union any concurrent DV. An + * operation that also rewrites rows into new data files (e.g. UPDATE or MERGE) can only union a + * concurrent DV when the newly deleted positions are disjoint; otherwise the same row would be + * both rewritten here and deleted concurrently. + */ + @Override + protected boolean mergedConcurrentDVsMustBeDisjoint() { + return addsDataFiles(); } /** diff --git a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java index 61514e5ea26a..27b352ad87e9 100644 --- a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java @@ -33,6 +33,7 @@ import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; +import org.apache.iceberg.deletes.PositionDeleteIndex; import org.apache.iceberg.encryption.EncryptedOutputFile; import org.apache.iceberg.encryption.EncryptingFileIO; import org.apache.iceberg.events.CreateSnapshotEvent; @@ -100,6 +101,10 @@ abstract class MergingSnapshotProducer extends SnapshotProducer { private final Map> dvsByReferencedFile = Maps.newLinkedHashMap(); private final Map mergedConcurrentDVsByFile = Maps.newHashMap(); private final List mergedDVFileLocations = Lists.newArrayList(); + // positions deleted at this operation's starting snapshot for a referenced data file (the DVs it + // supersedes); captured once because merging a concurrent DV mutates the pending removals + private final Map> basePositionsByReferencedFile = + Maps.newHashMap(); private final List appendManifests = Lists.newArrayList(); private final List rewrittenAppendManifests = Lists.newArrayList(); private final SnapshotSummary.Builder addedDataFilesSummary = SnapshotSummary.builder(); @@ -833,6 +838,16 @@ protected boolean canMergeConcurrentDVs() { return false; } + /** + * Whether a concurrently added DV merged into this operation's DVs must have deleted positions + * that are disjoint from this operation's own deleted positions. When true, a concurrent DV whose + * newly deleted positions overlap this operation's newly deleted positions fails validation + * instead of being merged. + */ + protected boolean mergedConcurrentDVsMustBeDisjoint() { + return false; + } + // validates there are no concurrently added DVs for referenced data files, or merges the // concurrent DVs into this operation's DVs if merging is allowed protected void validateAddedDVs( @@ -954,6 +969,10 @@ private void mergeConcurrentDVs(List concurrentDVs) { dvsByReferencedFile.get(referencedDataFile).remove(previouslyMerged); } + if (mergedConcurrentDVsMustBeDisjoint()) { + validateDisjointConcurrentDV(referencedDataFile, concurrentDV); + } + LOG.info( "Merging concurrently added DV {} for {} into the new DV in table {}", concurrentDV.location(), @@ -974,6 +993,77 @@ private void mergeConcurrentDVs(List concurrentDVs) { } } + /** + * Validates that a concurrently added DV deletes no position that this operation also newly + * deletes for the same data file. Positions already deleted at this operation's starting snapshot + * (the DVs this operation supersedes and removes) are shared by both DVs and are not a conflict; + * they are subtracted before comparing. + */ + private void validateDisjointConcurrentDV(String referencedDataFile, DeleteFile concurrentDV) { + FileIO io = EncryptingFileIO.combine(ops().io(), ops().encryption()); + PositionDeleteIndex concurrentPositions = DVUtil.readDV(concurrentDV, io); + List basePositions = basePositions(referencedDataFile, io); + + for (DeleteFile ownDV : dvsByReferencedFile.get(referencedDataFile)) { + long overlap = + overlappingPositionCount(DVUtil.readDV(ownDV, io), concurrentPositions, basePositions); + ValidationException.check( + overlap == 0, + "Found concurrently added DV for %s that deletes %s position(s) also deleted by this operation: %s", + referencedDataFile, + overlap, + ContentFileUtil.dvDesc(concurrentDV)); + } + } + + /** + * Returns the positions already deleted at this operation's starting snapshot for a referenced + * data file (the DVs this operation supersedes). These DVs are fixed when the operation is built, + * so their positions are read once and cached per referenced file. Merging a concurrent DV drops + * the superseded DVs from the pending removals and adds the merged DV in their place, so on a + * later commit attempt the pending removals no longer identify this operation's base. + */ + private List basePositions(String referencedDataFile, FileIO io) { + return basePositionsByReferencedFile.computeIfAbsent( + referencedDataFile, + file -> { + List positions = Lists.newArrayList(); + for (DeleteFile removed : deleteFilterManager.filesToBeDeleted()) { + if (file.equals(removed.referencedDataFile()) && ContentFileUtil.isDV(removed)) { + positions.add(DVUtil.readDV(removed, io)); + } + } + return positions; + }); + } + + // counts positions deleted by both DVs that were not already deleted at the starting snapshot + private static long overlappingPositionCount( + PositionDeleteIndex self, PositionDeleteIndex concurrent, List base) { + // iterate the smaller index and probe the larger to keep the scan proportional to the smaller + // DV + PositionDeleteIndex smaller = + self.cardinality() <= concurrent.cardinality() ? self : concurrent; + PositionDeleteIndex larger = smaller == self ? concurrent : self; + long[] overlap = {0L}; + smaller.forEach( + position -> { + if (larger.isDeleted(position) && !isDeletedIn(base, position)) { + overlap[0] += 1L; + } + }); + return overlap[0]; + } + + private static boolean isDeletedIn(List indexes, long position) { + for (PositionDeleteIndex index : indexes) { + if (index.isDeleted(position)) { + return true; + } + } + return false; + } + private static boolean isSameDV(DeleteFile dv1, DeleteFile dv2) { return dv1.location().equals(dv2.location()) && Objects.equals(dv1.contentOffset(), dv2.contentOffset()); diff --git a/core/src/main/java/org/apache/iceberg/TableProperties.java b/core/src/main/java/org/apache/iceberg/TableProperties.java index ad807d593f71..924e4b97fcbc 100644 --- a/core/src/main/java/org/apache/iceberg/TableProperties.java +++ b/core/src/main/java/org/apache/iceberg/TableProperties.java @@ -121,6 +121,16 @@ private TableProperties() {} public static final String MANIFEST_MERGE_ENABLED = "commit.manifest-merge.enabled"; public static final boolean MANIFEST_MERGE_ENABLED_DEFAULT = true; + // Enables merging a concurrently added DV into a row-level operation that also adds data files + // (e.g. UPDATE or MERGE) when the newly deleted positions are disjoint, instead of failing the + // commit. The check reads DV bitmaps on the driver only for data files this operation and a + // concurrent commit both added a DV to, so it is skipped entirely when the commits touch + // disjoint data files. Pure delete commits always merge concurrent DVs regardless of this + // property. + public static final String ROW_LEVEL_CONCURRENT_DV_MERGE_ENABLED = + "commit.row-level.concurrent-dv-merge.enabled"; + public static final boolean ROW_LEVEL_CONCURRENT_DV_MERGE_ENABLED_DEFAULT = true; + public static final String DEFAULT_FILE_FORMAT = "write.format.default"; public static final String DELETE_DEFAULT_FILE_FORMAT = "write.delete.format.default"; public static final String DEFAULT_FILE_FORMAT_DEFAULT = "parquet"; diff --git a/core/src/test/java/org/apache/iceberg/TestRowDelta.java b/core/src/test/java/org/apache/iceberg/TestRowDelta.java index b85c3cde9d1d..d7f46e209ccb 100644 --- a/core/src/test/java/org/apache/iceberg/TestRowDelta.java +++ b/core/src/test/java/org/apache/iceberg/TestRowDelta.java @@ -2149,24 +2149,239 @@ private List mergedDVFileNames() throws IOException { } @TestTemplate - public void testConcurrentDVNotMergedWhenAddingDataFiles() { + public void testConcurrentDisjointDVMergedWhenAddingDataFiles() throws IOException { assumeThat(formatVersion).isGreaterThanOrEqualTo(3); DataFile dataFile = newDataFile("data_bucket=0"); commit(table, table.newRowDelta().addRows(dataFile), branch); - DeleteFile deleteFile1 = newDeletes(dataFile); - RowDelta rowDelta1 = table.newRowDelta().addDeletes(deleteFile1); + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); - DeleteFile deleteFile2 = newDeletes(dataFile); - RowDelta rowDelta2 = - table.newRowDelta().addRows(newDataFile("data_bucket=1")).addDeletes(deleteFile2); + // a concurrent delete removes rows 0-1 + DeleteFile concurrentDV = dvWithPositions(dataFile, fileFactory, 0, 2); + RowDelta concurrent = table.newRowDelta().addDeletes(concurrentDV); - commit(table, rowDelta1, branch); + // this operation rewrites rows 2-3 into a new data file; its deletes are disjoint from the + // concurrent delete, so the concurrent DV must be merged instead of failing + DeleteFile updateDV = dvWithPositions(dataFile, fileFactory, 2, 4); + RowDelta update = + table.newRowDelta().addRows(newDataFile("data_bucket=1")).addDeletes(updateDV); - assertThatThrownBy(() -> commit(table, rowDelta2, branch)) + commit(table, concurrent, branch); + commit(table, update, branch); + + SnapshotChanges changes = + SnapshotChanges.builderFor(table).snapshot(latestSnapshot(table, branch)).build(); + + DeleteFile mergedDV = Iterables.getOnlyElement(changes.addedDeleteFiles()); + assertDVHasDeletedPositions(mergedDV, LongStream.range(0, 4).boxed()::iterator); + assertThat(changes.removedDeleteFiles()) + .extracting(ContentFile::location) + .containsExactly(concurrentDV.location()); + } + + @TestTemplate + public void testConcurrentOverlappingDVFailsWhenAddingDataFiles() throws IOException { + assumeThat(formatVersion).isGreaterThanOrEqualTo(3); + + DataFile dataFile = newDataFile("data_bucket=0"); + commit(table, table.newRowDelta().addRows(dataFile), branch); + + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + + // a concurrent delete removes rows 0-2 + DeleteFile concurrentDV = dvWithPositions(dataFile, fileFactory, 0, 3); + RowDelta concurrent = table.newRowDelta().addDeletes(concurrentDV); + + // this operation rewrites rows 2-3 into a new data file; row 2 is also deleted concurrently, so + // merging would let the row be both rewritten here and deleted concurrently + DeleteFile updateDV = dvWithPositions(dataFile, fileFactory, 2, 4); + RowDelta update = + table.newRowDelta().addRows(newDataFile("data_bucket=1")).addDeletes(updateDV); + + commit(table, concurrent, branch); + + assertThatThrownBy(() -> commit(table, update, branch)) .isInstanceOf(ValidationException.class) - .hasMessageContaining("Found concurrently added DV for %s", dataFile.location()); + .hasMessageContaining("Found concurrently added DV for %s", dataFile.location()) + .hasMessageContaining("also deleted by this operation"); + } + + @TestTemplate + public void testConcurrentDisjointDVNotMergedWhenRowLevelMergeDisabled() throws IOException { + assumeThat(formatVersion).isGreaterThanOrEqualTo(3); + + table + .updateProperties() + .set(TableProperties.ROW_LEVEL_CONCURRENT_DV_MERGE_ENABLED, "false") + .commit(); + + DataFile dataFile = newDataFile("data_bucket=0"); + commit(table, table.newRowDelta().addRows(dataFile), branch); + + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + + // the deleted positions are disjoint, so this would merge when the property is enabled + DeleteFile concurrentDV = dvWithPositions(dataFile, fileFactory, 0, 2); + RowDelta concurrent = table.newRowDelta().addDeletes(concurrentDV); + + DeleteFile updateDV = dvWithPositions(dataFile, fileFactory, 2, 4); + RowDelta update = + table.newRowDelta().addRows(newDataFile("data_bucket=1")).addDeletes(updateDV); + + commit(table, concurrent, branch); + + // with the merge disabled, an operation that adds data files falls back to failing on any + // concurrent DV for a data file it also added a DV to, without comparing positions + assertThatThrownBy(() -> commit(table, update, branch)) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("Found concurrently added DV for %s", dataFile.location()) + .hasMessageNotContaining("also deleted by this operation"); + } + + @TestTemplate + public void testConcurrentDVSharingOnlyBaseDeletionMergedWhenAddingDataFiles() + throws IOException { + assumeThat(formatVersion).isGreaterThanOrEqualTo(3); + + DataFile dataFile = newDataFile("data_bucket=0"); + commit(table, table.newRowDelta().addRows(dataFile), branch); + + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + + // row 0 is already deleted at the base snapshot both operations start from + DeleteFile existingDV = dvWithPositions(dataFile, fileFactory, 0, 1); + commit(table, table.newRowDelta().addDeletes(existingDV), branch); + Snapshot base = latestSnapshot(table, branch); + + // a concurrent delete supersedes the base DV and additionally removes row 1 + DeleteFile concurrentDV = dvWithExplicitPositions(dataFile, fileFactory, 0L, 1L); + RowDelta concurrent = + table + .newRowDelta() + .addDeletes(concurrentDV) + .removeDeletes(existingDV) + .validateFromSnapshot(base.snapshotId()); + + // this operation rewrites row 2 into a new data file; its DV supersedes the base DV. + // row 0 is shared with the concurrent DV only because both inherited it from the base DV. + // after subtracting the base, the newly deleted positions (1 and 2) are disjoint, so it merges + DeleteFile updateDV = dvWithExplicitPositions(dataFile, fileFactory, 0L, 2L); + RowDelta update = + table + .newRowDelta() + .addRows(newDataFile("data_bucket=1")) + .addDeletes(updateDV) + .removeDeletes(existingDV) + .validateFromSnapshot(base.snapshotId()); + + commit(table, concurrent, branch); + commit(table, update, branch); + + SnapshotChanges changes = + SnapshotChanges.builderFor(table).snapshot(latestSnapshot(table, branch)).build(); + + DeleteFile mergedDV = Iterables.getOnlyElement(changes.addedDeleteFiles()); + assertDVHasDeletedPositions(mergedDV, LongStream.range(0, 3).boxed()::iterator); + assertThat(changes.removedDeleteFiles()) + .extracting(ContentFile::location) + .containsExactly(concurrentDV.location()); + } + + @TestTemplate + public void testConcurrentDVConflictingOnNewPositionFailsWithBaseWhenAddingDataFiles() + throws IOException { + assumeThat(formatVersion).isGreaterThanOrEqualTo(3); + + DataFile dataFile = newDataFile("data_bucket=0"); + commit(table, table.newRowDelta().addRows(dataFile), branch); + + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + + // row 0 is already deleted at the base snapshot both operations start from + DeleteFile existingDV = dvWithPositions(dataFile, fileFactory, 0, 1); + commit(table, table.newRowDelta().addDeletes(existingDV), branch); + Snapshot base = latestSnapshot(table, branch); + + // a concurrent delete supersedes the base DV and additionally removes row 1 + DeleteFile concurrentDV = dvWithExplicitPositions(dataFile, fileFactory, 0L, 1L); + RowDelta concurrent = + table + .newRowDelta() + .addDeletes(concurrentDV) + .removeDeletes(existingDV) + .validateFromSnapshot(base.snapshotId()); + + // this operation rewrites row 1 into a new data file; its DV supersedes the base DV. + // row 1 is newly deleted by both operations and is not part of the base, so subtracting the + // base still leaves a real conflict and the merge must fail + DeleteFile updateDV = dvWithExplicitPositions(dataFile, fileFactory, 0L, 1L); + RowDelta update = + table + .newRowDelta() + .addRows(newDataFile("data_bucket=1")) + .addDeletes(updateDV) + .removeDeletes(existingDV) + .validateFromSnapshot(base.snapshotId()); + + commit(table, concurrent, branch); + + assertThatThrownBy(() -> commit(table, update, branch)) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("Found concurrently added DV for %s", dataFile.location()) + .hasMessageContaining("also deleted by this operation"); + } + + @TestTemplate + public void testConcurrentDisjointDVWithBaseMergedOnRetryWhenAddingDataFiles() + throws IOException { + assumeThat(formatVersion).isGreaterThanOrEqualTo(3); + + DataFile dataFile = newDataFile("data_bucket=0"); + commit(table, table.newRowDelta().addRows(dataFile), branch); + + OutputFileFactory fileFactory = + OutputFileFactory.builderFor(table, 1, 1).format(FileFormat.PUFFIN).build(); + + // row 0 is already deleted at the base snapshot both operations start from + DeleteFile existingDV = dvWithPositions(dataFile, fileFactory, 0, 1); + commit(table, table.newRowDelta().addDeletes(existingDV), branch); + Snapshot base = latestSnapshot(table, branch); + + DeleteFile concurrentDV = dvWithExplicitPositions(dataFile, fileFactory, 0L, 1L); + RowDelta concurrent = + table + .newRowDelta() + .addDeletes(concurrentDV) + .removeDeletes(existingDV) + .validateFromSnapshot(base.snapshotId()); + + DeleteFile updateDV = dvWithExplicitPositions(dataFile, fileFactory, 0L, 2L); + RowDelta update = + table + .newRowDelta() + .addRows(newDataFile("data_bucket=1")) + .addDeletes(updateDV) + .removeDeletes(existingDV) + .validateFromSnapshot(base.snapshotId()); + + commit(table, concurrent, branch); + + // on retry the same concurrent DV is recognized as already merged, so the disjoint update's + // merged result is committed once without being validated or applied a second time + table.ops().failCommits(1); + commit(table, update, branch); + + SnapshotChanges changes = + SnapshotChanges.builderFor(table).snapshot(latestSnapshot(table, branch)).build(); + + DeleteFile mergedDV = Iterables.getOnlyElement(changes.addedDeleteFiles()); + assertDVHasDeletedPositions(mergedDV, LongStream.range(0, 3).boxed()::iterator); } @TestTemplate @@ -2932,6 +3147,16 @@ private DeleteFile dvWithPositions( return writeDV(targetTable, deletes, dataFile.specId(), dataFile.partition(), fileFactory); } + private DeleteFile dvWithExplicitPositions( + DataFile dataFile, OutputFileFactory fileFactory, long... positions) throws IOException { + List> deletes = Lists.newArrayList(); + for (long pos : positions) { + deletes.add(PositionDelete.create().set(dataFile.location(), pos)); + } + + return writeDV(table, deletes, dataFile.specId(), dataFile.partition(), fileFactory); + } + private void assertDVHasDeletedPositions(DeleteFile dv, Iterable positions) { assertDVHasDeletedPositions(table, dv, positions); }