From b585203950a7a6acb412ccf918e922f32bac99ac Mon Sep 17 00:00:00 2001 From: Maximilian Michels Date: Mon, 13 Jul 2026 17:45:51 +0200 Subject: [PATCH] Flink: Backport: Remove converted equality deletes for same-branch conversion (#17189) --- .../operator/EqualityConvertCommitter.java | 26 ++++- .../operator/EqualityConvertPlan.java | 5 + .../operator/EqualityConvertPlanner.java | 1 + .../api/TestConvertEqualityDeletesE2E.java | 26 +++++ .../TestEqualityConvertCommitter.java | 109 ++++++++++++++++++ .../operator/TestEqualityConvertDVWriter.java | 19 ++- .../operator/EqualityConvertCommitter.java | 26 ++++- .../operator/EqualityConvertPlan.java | 5 + .../operator/EqualityConvertPlanner.java | 1 + .../api/TestConvertEqualityDeletesE2E.java | 26 +++++ .../TestEqualityConvertCommitter.java | 109 ++++++++++++++++++ .../operator/TestEqualityConvertDVWriter.java | 19 ++- 12 files changed, 360 insertions(+), 12 deletions(-) diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index 95c9ccd5de1c..ada16ec09384 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -81,6 +81,7 @@ public class EqualityConvertCommitter extends AbstractStreamOperator "equality-convert-staging-snapshot"; private static final String ADDED_DV_NUM_METRIC = "addedDvNum"; + private static final String REMOVED_EQ_DELETE_NUM_METRIC = "removedEqDeleteNum"; private static final String COMMIT_DURATION_MS_METRIC = "commitDurationMs"; private final String tableName; @@ -97,6 +98,7 @@ public class EqualityConvertCommitter extends AbstractStreamOperator private transient Counter addedDataFileNumCounter; private transient Counter addedDataFileSizeCounter; private transient Counter addedDvNumCounter; + private transient Counter removedEqDeleteNumCounter; private transient Counter commitDurationMsCounter; public EqualityConvertCommitter( @@ -130,6 +132,7 @@ public void open() throws Exception { this.addedDataFileSizeCounter = taskMetricGroup.counter(TableMaintenanceMetrics.ADDED_DATA_FILE_SIZE_METRIC); this.addedDvNumCounter = taskMetricGroup.counter(ADDED_DV_NUM_METRIC); + this.removedEqDeleteNumCounter = taskMetricGroup.counter(REMOVED_EQ_DELETE_NUM_METRIC); this.commitDurationMsCounter = taskMetricGroup.counter(COMMIT_DURATION_MS_METRIC); } @@ -225,11 +228,13 @@ private void commitIfNeeded() { long durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano); commitDurationMsCounter.inc(durationMs); + int removedEqDeletes = stagingOnTargetBranch ? planResult.eqDeleteFiles().size() : 0; LOG.info( - "Committed {} data files and {} DV files to branch '{}' for table {} in {} ms. " - + "Processed staging snapshot {}.", + "Committed {} data files and {} DV files, removed {} equality delete files on branch '{}' " + + "for table {} in {} ms. Processed staging snapshot {}.", planResult.dataFiles().size(), allDvFiles.size(), + removedEqDeletes, targetBranch, tableName, durationMs, @@ -245,6 +250,7 @@ private void commitIfNeeded() { } addedDvNumCounter.inc(allDvFiles.size()); + removedEqDeleteNumCounter.inc(removedEqDeletes); } @VisibleForTesting @@ -252,6 +258,11 @@ void commit(RowDelta rowDelta) { rowDelta.commit(); } + @VisibleForTesting + long removedEqDeleteNum() { + return removedEqDeleteNumCounter.getCount(); + } + /** * Deletes the DVs this cycle wrote but did not commit (abort or definite commit failure). Only * the newly written DVs are removed; rewritten DVs remain referenced on the target branch. Best @@ -315,7 +326,16 @@ private RowDelta buildRowDelta( rowDelta.addDeletes(dvFile); } - if (!stagingOnTargetBranch) { + if (stagingOnTargetBranch) { + // In-place conversion: the equality deletes are already on the target branch. Their rows + // are now covered by DVs, so remove them; readers then stop loading and applying equality + // deletes. + for (DeleteFile eqDeleteFile : planResult.eqDeleteFiles()) { + rowDelta.removeDeletes(eqDeleteFile); + } + } else { + // Separate target branch: the writer's DVs exist only on staging, so add them to the target + // (addStagingDeletes skips any superseded DVs). addStagingDeletes(rowDelta, referencedDataFiles, planResult.stagingDVFiles()); } diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java index 93ae956b2b37..191bf8322039 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java @@ -37,6 +37,9 @@ * @param dataFiles new staging data files committed in the cycle * @param stagingDVFiles staging DVs passed through to main, used by DVWriter to merge with * newly-created DVs + * @param eqDeleteFiles equality delete files resolved this cycle. Removed by the committer when + * staging and target are the same branch, so readers stop applying them once the equivalent DVs + * commit. Empty on a separate target branch, where the eq deletes remain on staging. * @param stagingSnapshotId staging snapshot the cycle resolved against, or {@link * #NO_OP_STAGING_SNAPSHOT_ID} for a no-op cycle * @param mainSnapshotId main branch snapshot ID the index was resolved against, used for commit @@ -49,6 +52,7 @@ public record EqualityConvertPlan( List dataFiles, List stagingDVFiles, + List eqDeleteFiles, long stagingSnapshotId, Long mainSnapshotId, long triggerTimestamp, @@ -66,6 +70,7 @@ public boolean noOp() { public static EqualityConvertPlan noOp( Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) { return new EqualityConvertPlan( + ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), NO_OP_STAGING_SNAPSHOT_ID, diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java index 89e5510dd848..8499f689b27c 100644 --- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java +++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java @@ -461,6 +461,7 @@ private void processStagingSnapshot( new EqualityConvertPlan( inputs.newDataFiles(), inputs.stagingDVFiles(), + inputs.eqDeleteFiles(), stagingSnapshot.snapshotId(), currentMainSnapshotId, triggerTs, diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java index f6d1fd9464e2..f3bdaa8fe1c6 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java @@ -29,6 +29,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; import org.apache.iceberg.Files; import org.apache.iceberg.ManifestFile; @@ -126,6 +127,10 @@ void testConvertEqualityDeletesE2E(String stagingBranch) throws Exception { .pollInterval(Duration.ofMillis(200)) .untilAsserted(() -> assertThat(dvCountOnMain(table)).isEqualTo(2)); SimpleDataUtil.assertTableRecords(table, ImmutableList.of(createRecord(3, "c"))); + + // In-place conversion (staging == main) removes the converted equality deletes so reads no + // longer apply them. On a separate staging branch, main never holds equality deletes. + assertThat(eqDeleteCountOnMain(table)).isZero(); } finally { closeJobClient(jobClient); } @@ -167,4 +172,25 @@ private static long dvCountOnMain(Table table) throws IOException { return count; } + + private static long eqDeleteCountOnMain(Table table) throws IOException { + table.refresh(); + if (table.currentSnapshot() == null) { + return 0; + } + + long count = 0; + for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) { + for (DeleteFile file : reader) { + if (file.content() == FileContent.EQUALITY_DELETES) { + count++; + } + } + } + } + + return count; + } } diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java index 54f0b1758d90..84c0c2b67969 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java @@ -20,7 +20,9 @@ import static org.assertj.core.api.Assertions.assertThat; +import java.io.File; import java.io.IOException; +import java.nio.file.Path; import java.util.List; import java.util.stream.Collectors; import org.apache.flink.streaming.api.watermark.Watermark; @@ -28,10 +30,14 @@ import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; +import org.apache.iceberg.Files; import org.apache.iceberg.ManifestFile; import org.apache.iceberg.ManifestFiles; import org.apache.iceberg.ManifestReader; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.RowDelta; import org.apache.iceberg.SnapshotRef; import org.apache.iceberg.Table; @@ -40,12 +46,16 @@ import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.exceptions.CommitStateUnknownException; import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.flink.SimpleDataUtil; import org.apache.iceberg.flink.maintenance.api.Trigger; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; class TestEqualityConvertCommitter extends OperatorTestBase { + @TempDir private Path tempDir; + @Test void commitsDataFilesToMainBranch() throws Exception { Table table = createTable(3, FileFormat.PARQUET); @@ -64,6 +74,7 @@ void commitsDataFilesToMainBranch() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), stagingSnapshotId, snapshotIdBefore, doneTs - 1, @@ -101,6 +112,7 @@ void holdsBackWatermarkUntilCommit() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 42L, snapshotIdBefore, doneTs - 2, @@ -158,6 +170,7 @@ void abortsCommitWhenDVWriterFailed() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 42L, snapshotIdBefore, doneTs - 1, @@ -194,6 +207,7 @@ void failsCommitWhenExternalCommitLandsAfterPlan() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), stagingSnapshotId, mainSnapshotIdAtPlan, doneTs - 1, @@ -251,6 +265,7 @@ void failsReplayOfCommittedPlanFromOlderState() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), stagingSnapshotId, mainSnapshotIdAtPlan, doneTs - 1, @@ -273,6 +288,7 @@ void failsReplayOfCommittedPlanFromOlderState() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), stagingSnapshotId, mainSnapshotIdAtPlan, doneTs2 - 1, @@ -318,6 +334,7 @@ void deletesWrittenDvsWhenCommitFails() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 777L, mainSnapshotIdAtPlan, doneTs - 1, @@ -368,6 +385,7 @@ void deletesSiblingDvsOnAbortButKeepsRewrittenDvs() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 42L, snapshotIdBefore, doneTs - 1, @@ -422,6 +440,7 @@ void commit(RowDelta rowDelta) { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 888L, mainSnapshotIdAtPlan, doneTs - 1, @@ -473,6 +492,7 @@ void removesRewrittenStagingDvOnSharedBranch() throws Exception { new EqualityConvertPlan( Lists.newArrayList(), Lists.newArrayList(stagingDv), + Lists.newArrayList(), 123L, mainSnapshotId, doneTs - 1, @@ -496,6 +516,66 @@ void removesRewrittenStagingDvOnSharedBranch() throws Exception { } } + @Test + void removesConvertedEqualityDeletesOnSharedBranch() throws Exception { + // Shared branch: the writer's equality deletes are already on the target branch. Once the cycle + // resolves them to DVs, the committer removes the equality delete files in the same commit so + // readers stop applying them. + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + DataFile dataFile = getFirstDataFile(table); + + DeleteFile eqDelete = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete).commit(); + table.refresh(); + long mainSnapshotId = table.currentSnapshot().snapshotId(); + assertThat(equalityDeletes(table)).hasSize(1); + + // The cycle resolved the eq delete to a DV covering row position 0 of the data file. + DeleteFile convertedDv = writeDV(table, dataFile.location(), 0L); + + EqualityConvertCommitter committer = + new EqualityConvertCommitter( + DUMMY_TABLE_NAME, + DUMMY_TASK_NAME, + tableLoader(), + SnapshotRef.MAIN_BRANCH, + SnapshotRef.MAIN_BRANCH); + try (TwoInputStreamOperatorTestHarness harness = + new TwoInputStreamOperatorTestHarness<>(committer)) { + harness.open(); + + long doneTs = System.currentTimeMillis(); + EqualityConvertPlan planResult = + new EqualityConvertPlan( + Lists.newArrayList(), + Lists.newArrayList(), + Lists.newArrayList(eqDelete), + 123L, + mainSnapshotId, + doneTs - 1, + doneTs); + + harness.processElement1( + new StreamRecord<>( + new DVWriteResult(Lists.newArrayList(convertedDv), Lists.newArrayList()), doneTs)); + harness.processElement2(new StreamRecord<>(planResult, doneTs - 1)); + harness.processBothWatermarks(new Watermark(doneTs)); + + assertThat(harness.extractOutputValues()).hasSize(1); + + table.refresh(); + // The equality delete file is gone; only the converted DV remains for the data file. + assertThat(equalityDeletes(table)).isEmpty(); + List dvs = deletesForDataFile(table, dataFile.location()); + assertThat(dvs).hasSize(1); + assertThat(dvs.get(0).location()).isEqualTo(convertedDv.location()); + + assertThat(committer.removedEqDeleteNum()).isEqualTo(1); + } + } + private TwoInputStreamOperatorTestHarness createHarness() throws Exception { return new TwoInputStreamOperatorTestHarness<>( @@ -539,6 +619,35 @@ private static List deletesForDataFile(Table table, String dataFileP return deletes; } + private static List equalityDeletes(Table table) { + List deletes = Lists.newArrayList(); + for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) { + for (DeleteFile file : reader) { + if (file.content() == FileContent.EQUALITY_DELETES) { + deletes.add(file.copy()); + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + return deletes; + } + + private DeleteFile writeEqualityDelete(Table table, Integer id, String data) throws IOException { + File file = File.createTempFile("junit", null, tempDir.toFile()); + assertThat(file.delete()).isTrue(); + return FileHelpers.writeDeleteFile( + table, + Files.localOutput(file), + new PartitionData(PartitionSpec.unpartitioned().partitionType()), + Lists.newArrayList(SimpleDataUtil.createRecord(id, data)), + table.schema()); + } + private static DeleteFile writeDV(Table table, String dataFilePath, long... positions) throws IOException { List> deletes = Lists.newArrayList(); diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java index a0c2f3568baf..56c6a2f98b33 100644 --- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java +++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java @@ -210,7 +210,13 @@ void mergesStagingDVIntoWrittenDV() throws Exception { long time = System.currentTimeMillis(); EqualityConvertPlan planResult = new EqualityConvertPlan( - Lists.newArrayList(), Lists.newArrayList(stagingDV), 1L, null, 0L, 0L); + Lists.newArrayList(), + Lists.newArrayList(stagingDV), + Lists.newArrayList(), + 1L, + null, + 0L, + 0L); // New position 1 in the same file. Must merge it with the staged position 0. harness.processElement1( @@ -245,7 +251,13 @@ void abortsWhenMainSnapshotChangedSincePlanning() throws Exception { long time = System.currentTimeMillis(); EqualityConvertPlan planResult = new EqualityConvertPlan( - Lists.newArrayList(), Lists.newArrayList(), 1L, plannedSnapshotId, 0L, 0L); + Lists.newArrayList(), + Lists.newArrayList(), + Lists.newArrayList(), + 1L, + plannedSnapshotId, + 0L, + 0L); harness.processElement1( new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time)); @@ -447,7 +459,8 @@ private EqualityConvertDVWriter newResolver() { } private static EqualityConvertPlan emptyEqualityConvertPlan() { - return new EqualityConvertPlan(Lists.newArrayList(), Lists.newArrayList(), 1L, null, 0L, 0L); + return new EqualityConvertPlan( + Lists.newArrayList(), Lists.newArrayList(), Lists.newArrayList(), 1L, null, 0L, 0L); } private static String getFirstDataFilePath(Table table) { diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java index 95c9ccd5de1c..ada16ec09384 100644 --- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java +++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java @@ -81,6 +81,7 @@ public class EqualityConvertCommitter extends AbstractStreamOperator "equality-convert-staging-snapshot"; private static final String ADDED_DV_NUM_METRIC = "addedDvNum"; + private static final String REMOVED_EQ_DELETE_NUM_METRIC = "removedEqDeleteNum"; private static final String COMMIT_DURATION_MS_METRIC = "commitDurationMs"; private final String tableName; @@ -97,6 +98,7 @@ public class EqualityConvertCommitter extends AbstractStreamOperator private transient Counter addedDataFileNumCounter; private transient Counter addedDataFileSizeCounter; private transient Counter addedDvNumCounter; + private transient Counter removedEqDeleteNumCounter; private transient Counter commitDurationMsCounter; public EqualityConvertCommitter( @@ -130,6 +132,7 @@ public void open() throws Exception { this.addedDataFileSizeCounter = taskMetricGroup.counter(TableMaintenanceMetrics.ADDED_DATA_FILE_SIZE_METRIC); this.addedDvNumCounter = taskMetricGroup.counter(ADDED_DV_NUM_METRIC); + this.removedEqDeleteNumCounter = taskMetricGroup.counter(REMOVED_EQ_DELETE_NUM_METRIC); this.commitDurationMsCounter = taskMetricGroup.counter(COMMIT_DURATION_MS_METRIC); } @@ -225,11 +228,13 @@ private void commitIfNeeded() { long durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNano); commitDurationMsCounter.inc(durationMs); + int removedEqDeletes = stagingOnTargetBranch ? planResult.eqDeleteFiles().size() : 0; LOG.info( - "Committed {} data files and {} DV files to branch '{}' for table {} in {} ms. " - + "Processed staging snapshot {}.", + "Committed {} data files and {} DV files, removed {} equality delete files on branch '{}' " + + "for table {} in {} ms. Processed staging snapshot {}.", planResult.dataFiles().size(), allDvFiles.size(), + removedEqDeletes, targetBranch, tableName, durationMs, @@ -245,6 +250,7 @@ private void commitIfNeeded() { } addedDvNumCounter.inc(allDvFiles.size()); + removedEqDeleteNumCounter.inc(removedEqDeletes); } @VisibleForTesting @@ -252,6 +258,11 @@ void commit(RowDelta rowDelta) { rowDelta.commit(); } + @VisibleForTesting + long removedEqDeleteNum() { + return removedEqDeleteNumCounter.getCount(); + } + /** * Deletes the DVs this cycle wrote but did not commit (abort or definite commit failure). Only * the newly written DVs are removed; rewritten DVs remain referenced on the target branch. Best @@ -315,7 +326,16 @@ private RowDelta buildRowDelta( rowDelta.addDeletes(dvFile); } - if (!stagingOnTargetBranch) { + if (stagingOnTargetBranch) { + // In-place conversion: the equality deletes are already on the target branch. Their rows + // are now covered by DVs, so remove them; readers then stop loading and applying equality + // deletes. + for (DeleteFile eqDeleteFile : planResult.eqDeleteFiles()) { + rowDelta.removeDeletes(eqDeleteFile); + } + } else { + // Separate target branch: the writer's DVs exist only on staging, so add them to the target + // (addStagingDeletes skips any superseded DVs). addStagingDeletes(rowDelta, referencedDataFiles, planResult.stagingDVFiles()); } diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java index 93ae956b2b37..191bf8322039 100644 --- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java +++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlan.java @@ -37,6 +37,9 @@ * @param dataFiles new staging data files committed in the cycle * @param stagingDVFiles staging DVs passed through to main, used by DVWriter to merge with * newly-created DVs + * @param eqDeleteFiles equality delete files resolved this cycle. Removed by the committer when + * staging and target are the same branch, so readers stop applying them once the equivalent DVs + * commit. Empty on a separate target branch, where the eq deletes remain on staging. * @param stagingSnapshotId staging snapshot the cycle resolved against, or {@link * #NO_OP_STAGING_SNAPSHOT_ID} for a no-op cycle * @param mainSnapshotId main branch snapshot ID the index was resolved against, used for commit @@ -49,6 +52,7 @@ public record EqualityConvertPlan( List dataFiles, List stagingDVFiles, + List eqDeleteFiles, long stagingSnapshotId, Long mainSnapshotId, long triggerTimestamp, @@ -66,6 +70,7 @@ public boolean noOp() { public static EqualityConvertPlan noOp( Long mainSnapshotId, long triggerTimestamp, long doneTimestamp) { return new EqualityConvertPlan( + ImmutableList.of(), ImmutableList.of(), ImmutableList.of(), NO_OP_STAGING_SNAPSHOT_ID, diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java index 89e5510dd848..8499f689b27c 100644 --- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java +++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPlanner.java @@ -461,6 +461,7 @@ private void processStagingSnapshot( new EqualityConvertPlan( inputs.newDataFiles(), inputs.stagingDVFiles(), + inputs.eqDeleteFiles(), stagingSnapshot.snapshotId(), currentMainSnapshotId, triggerTs, diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java index f6d1fd9464e2..f3bdaa8fe1c6 100644 --- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java +++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesE2E.java @@ -29,6 +29,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; import org.apache.iceberg.Files; import org.apache.iceberg.ManifestFile; @@ -126,6 +127,10 @@ void testConvertEqualityDeletesE2E(String stagingBranch) throws Exception { .pollInterval(Duration.ofMillis(200)) .untilAsserted(() -> assertThat(dvCountOnMain(table)).isEqualTo(2)); SimpleDataUtil.assertTableRecords(table, ImmutableList.of(createRecord(3, "c"))); + + // In-place conversion (staging == main) removes the converted equality deletes so reads no + // longer apply them. On a separate staging branch, main never holds equality deletes. + assertThat(eqDeleteCountOnMain(table)).isZero(); } finally { closeJobClient(jobClient); } @@ -167,4 +172,25 @@ private static long dvCountOnMain(Table table) throws IOException { return count; } + + private static long eqDeleteCountOnMain(Table table) throws IOException { + table.refresh(); + if (table.currentSnapshot() == null) { + return 0; + } + + long count = 0; + for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) { + for (DeleteFile file : reader) { + if (file.content() == FileContent.EQUALITY_DELETES) { + count++; + } + } + } + } + + return count; + } } diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java index 54f0b1758d90..84c0c2b67969 100644 --- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java +++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java @@ -20,7 +20,9 @@ import static org.assertj.core.api.Assertions.assertThat; +import java.io.File; import java.io.IOException; +import java.nio.file.Path; import java.util.List; import java.util.stream.Collectors; import org.apache.flink.streaming.api.watermark.Watermark; @@ -28,10 +30,14 @@ import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness; import org.apache.iceberg.DataFile; import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; import org.apache.iceberg.FileFormat; +import org.apache.iceberg.Files; import org.apache.iceberg.ManifestFile; import org.apache.iceberg.ManifestFiles; import org.apache.iceberg.ManifestReader; +import org.apache.iceberg.PartitionData; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.RowDelta; import org.apache.iceberg.SnapshotRef; import org.apache.iceberg.Table; @@ -40,12 +46,16 @@ import org.apache.iceberg.deletes.PositionDelete; import org.apache.iceberg.exceptions.CommitStateUnknownException; import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.flink.SimpleDataUtil; import org.apache.iceberg.flink.maintenance.api.Trigger; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; class TestEqualityConvertCommitter extends OperatorTestBase { + @TempDir private Path tempDir; + @Test void commitsDataFilesToMainBranch() throws Exception { Table table = createTable(3, FileFormat.PARQUET); @@ -64,6 +74,7 @@ void commitsDataFilesToMainBranch() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), stagingSnapshotId, snapshotIdBefore, doneTs - 1, @@ -101,6 +112,7 @@ void holdsBackWatermarkUntilCommit() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 42L, snapshotIdBefore, doneTs - 2, @@ -158,6 +170,7 @@ void abortsCommitWhenDVWriterFailed() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 42L, snapshotIdBefore, doneTs - 1, @@ -194,6 +207,7 @@ void failsCommitWhenExternalCommitLandsAfterPlan() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), stagingSnapshotId, mainSnapshotIdAtPlan, doneTs - 1, @@ -251,6 +265,7 @@ void failsReplayOfCommittedPlanFromOlderState() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), stagingSnapshotId, mainSnapshotIdAtPlan, doneTs - 1, @@ -273,6 +288,7 @@ void failsReplayOfCommittedPlanFromOlderState() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), stagingSnapshotId, mainSnapshotIdAtPlan, doneTs2 - 1, @@ -318,6 +334,7 @@ void deletesWrittenDvsWhenCommitFails() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 777L, mainSnapshotIdAtPlan, doneTs - 1, @@ -368,6 +385,7 @@ void deletesSiblingDvsOnAbortButKeepsRewrittenDvs() throws Exception { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 42L, snapshotIdBefore, doneTs - 1, @@ -422,6 +440,7 @@ void commit(RowDelta rowDelta) { new EqualityConvertPlan( Lists.newArrayList(stagingDataFile), Lists.newArrayList(), + Lists.newArrayList(), 888L, mainSnapshotIdAtPlan, doneTs - 1, @@ -473,6 +492,7 @@ void removesRewrittenStagingDvOnSharedBranch() throws Exception { new EqualityConvertPlan( Lists.newArrayList(), Lists.newArrayList(stagingDv), + Lists.newArrayList(), 123L, mainSnapshotId, doneTs - 1, @@ -496,6 +516,66 @@ void removesRewrittenStagingDvOnSharedBranch() throws Exception { } } + @Test + void removesConvertedEqualityDeletesOnSharedBranch() throws Exception { + // Shared branch: the writer's equality deletes are already on the target branch. Once the cycle + // resolves them to DVs, the committer removes the equality delete files in the same commit so + // readers stop applying them. + Table table = createTableWithDelete(3); + insert(table, 1, "a"); + + DataFile dataFile = getFirstDataFile(table); + + DeleteFile eqDelete = writeEqualityDelete(table, 1, "a"); + table.newRowDelta().addDeletes(eqDelete).commit(); + table.refresh(); + long mainSnapshotId = table.currentSnapshot().snapshotId(); + assertThat(equalityDeletes(table)).hasSize(1); + + // The cycle resolved the eq delete to a DV covering row position 0 of the data file. + DeleteFile convertedDv = writeDV(table, dataFile.location(), 0L); + + EqualityConvertCommitter committer = + new EqualityConvertCommitter( + DUMMY_TABLE_NAME, + DUMMY_TASK_NAME, + tableLoader(), + SnapshotRef.MAIN_BRANCH, + SnapshotRef.MAIN_BRANCH); + try (TwoInputStreamOperatorTestHarness harness = + new TwoInputStreamOperatorTestHarness<>(committer)) { + harness.open(); + + long doneTs = System.currentTimeMillis(); + EqualityConvertPlan planResult = + new EqualityConvertPlan( + Lists.newArrayList(), + Lists.newArrayList(), + Lists.newArrayList(eqDelete), + 123L, + mainSnapshotId, + doneTs - 1, + doneTs); + + harness.processElement1( + new StreamRecord<>( + new DVWriteResult(Lists.newArrayList(convertedDv), Lists.newArrayList()), doneTs)); + harness.processElement2(new StreamRecord<>(planResult, doneTs - 1)); + harness.processBothWatermarks(new Watermark(doneTs)); + + assertThat(harness.extractOutputValues()).hasSize(1); + + table.refresh(); + // The equality delete file is gone; only the converted DV remains for the data file. + assertThat(equalityDeletes(table)).isEmpty(); + List dvs = deletesForDataFile(table, dataFile.location()); + assertThat(dvs).hasSize(1); + assertThat(dvs.get(0).location()).isEqualTo(convertedDv.location()); + + assertThat(committer.removedEqDeleteNum()).isEqualTo(1); + } + } + private TwoInputStreamOperatorTestHarness createHarness() throws Exception { return new TwoInputStreamOperatorTestHarness<>( @@ -539,6 +619,35 @@ private static List deletesForDataFile(Table table, String dataFileP return deletes; } + private static List equalityDeletes(Table table) { + List deletes = Lists.newArrayList(); + for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) { + for (DeleteFile file : reader) { + if (file.content() == FileContent.EQUALITY_DELETES) { + deletes.add(file.copy()); + } + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + return deletes; + } + + private DeleteFile writeEqualityDelete(Table table, Integer id, String data) throws IOException { + File file = File.createTempFile("junit", null, tempDir.toFile()); + assertThat(file.delete()).isTrue(); + return FileHelpers.writeDeleteFile( + table, + Files.localOutput(file), + new PartitionData(PartitionSpec.unpartitioned().partitionType()), + Lists.newArrayList(SimpleDataUtil.createRecord(id, data)), + table.schema()); + } + private static DeleteFile writeDV(Table table, String dataFilePath, long... positions) throws IOException { List> deletes = Lists.newArrayList(); diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java index a0c2f3568baf..56c6a2f98b33 100644 --- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java +++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertDVWriter.java @@ -210,7 +210,13 @@ void mergesStagingDVIntoWrittenDV() throws Exception { long time = System.currentTimeMillis(); EqualityConvertPlan planResult = new EqualityConvertPlan( - Lists.newArrayList(), Lists.newArrayList(stagingDV), 1L, null, 0L, 0L); + Lists.newArrayList(), + Lists.newArrayList(stagingDV), + Lists.newArrayList(), + 1L, + null, + 0L, + 0L); // New position 1 in the same file. Must merge it with the staged position 0. harness.processElement1( @@ -245,7 +251,13 @@ void abortsWhenMainSnapshotChangedSincePlanning() throws Exception { long time = System.currentTimeMillis(); EqualityConvertPlan planResult = new EqualityConvertPlan( - Lists.newArrayList(), Lists.newArrayList(), 1L, plannedSnapshotId, 0L, 0L); + Lists.newArrayList(), + Lists.newArrayList(), + Lists.newArrayList(), + 1L, + plannedSnapshotId, + 0L, + 0L); harness.processElement1( new StreamRecord<>(new DVPosition(dataFilePath, 0, 0, EMPTY_PARTITION, 0L), time)); @@ -447,7 +459,8 @@ private EqualityConvertDVWriter newResolver() { } private static EqualityConvertPlan emptyEqualityConvertPlan() { - return new EqualityConvertPlan(Lists.newArrayList(), Lists.newArrayList(), 1L, null, 0L, 0L); + return new EqualityConvertPlan( + Lists.newArrayList(), Lists.newArrayList(), Lists.newArrayList(), 1L, null, 0L, 0L); } private static String getFirstDataFilePath(Table table) {