Flink: Backport: Remove converted equality deletes for same-branch conversion (#17189)

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertCommitter.java

@@ -81,6 +81,7 @@ public class EqualityConvertCommitter extends AbstractStreamOperator<Trigger>
       "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<Trigger>
   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,13 +250,19 @@ private void commitIfNeeded() {
     }
 
     addedDvNumCounter.inc(allDvFiles.size());
+    removedEqDeleteNumCounter.inc(removedEqDeletes);
   }
 
   @VisibleForTesting
   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());
     }
 

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<DataFile> dataFiles,
     List<DeleteFile> stagingDVFiles,
+    List<DeleteFile> 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,

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,

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<DeleteFile> reader =
+          ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+        for (DeleteFile file : reader) {
+          if (file.content() == FileContent.EQUALITY_DELETES) {
+            count++;
+          }
+        }
+      }
+    }
+
+    return count;
+  }
 }

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertCommitter.java

@@ -20,18 +20,24 @@
 
 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;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 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<DVWriteResult, EqualityConvertPlan, Trigger> 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<DeleteFile> dvs = deletesForDataFile(table, dataFile.location());
+      assertThat(dvs).hasSize(1);
+      assertThat(dvs.get(0).location()).isEqualTo(convertedDv.location());
+
+      assertThat(committer.removedEqDeleteNum()).isEqualTo(1);
+    }
+  }
+
   private TwoInputStreamOperatorTestHarness<DVWriteResult, EqualityConvertPlan, Trigger>
       createHarness() throws Exception {
     return new TwoInputStreamOperatorTestHarness<>(
@@ -539,6 +619,35 @@ private static List<DeleteFile> deletesForDataFile(Table table, String dataFileP
     return deletes;
   }
 
+  private static List<DeleteFile> equalityDeletes(Table table) {
+    List<DeleteFile> deletes = Lists.newArrayList();
+    for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) {
+      try (ManifestReader<DeleteFile> 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<PositionDelete<?>> deletes = Lists.newArrayList();