Core/RewriteFiles: Fixed dropping delete files that are still required

This fixes a bug where delete files that are still required by the
re-written file could be deleted prematurely.

This happened because `MergingSnapshotProducer` calculated `minDataSequenceNumber` in a way that
could skip data sequence numbers that do exist in the final snapshot.

The filtered manifest list could exclude manifests of
a certain sequence number because their data files were re-written.
With specific orders of operations it was possible to skip the sequence
number of manifests that were re-written.

Author: jasonf20

core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java

@@ -915,16 +915,9 @@ public List<ManifestFile> apply(TableMetadata base, Snapshot snapshot) {
         filterManager.filterManifests(
             SnapshotUtil.schemaFor(base, targetBranch()),
             snapshot != null ? snapshot.dataManifests(ops().io()) : null);
-    long minDataSequenceNumber =
-        filtered.stream()
-            .map(ManifestFile::minSequenceNumber)
-            .filter(
-                seq ->
-                    seq
-                        != ManifestWriter
-                            .UNASSIGNED_SEQ) // filter out unassigned in rewritten manifests
-            .reduce(base.lastSequenceNumber(), Math::min);
-    deleteFilterManager.dropDeleteFilesOlderThan(minDataSequenceNumber);
+
+    long newMinSequenceNumber = minDataSequenceNumber(base, filtered);
+    deleteFilterManager.dropDeleteFilesOlderThan(newMinSequenceNumber);
     List<ManifestFile> filteredDeletes =
         deleteFilterManager.filterManifests(
             SnapshotUtil.schemaFor(base, targetBranch()),
@@ -955,6 +948,31 @@ public List<ManifestFile> apply(TableMetadata base, Snapshot snapshot) {
     return manifests;
   }
 
+  private long minDataSequenceNumber(TableMetadata base, List<ManifestFile> keptManifests) {
+    long minAddedDataSequenceNumber =
+        addedDataFiles().stream()
+            .map(ContentFile::dataSequenceNumber)
+            .filter(Objects::nonNull)
+            .filter(seq -> seq >= 0)
+            .reduce(base.nextSequenceNumber(), Math::min);
+
+    long minExistingDataSequenceNumber =
+        keptManifests.stream()
+            .map(ManifestFile::minSequenceNumber)
+            .filter(
+                seq ->
+                    seq
+                        != ManifestWriter
+                            .UNASSIGNED_SEQ) // filter out unassigned in rewritten manifests
+            .reduce(base.lastSequenceNumber(), Math::min);
+
+    return Math.min(
+        Math.min(Integer.MAX_VALUE, minExistingDataSequenceNumber),
+        newDataFilesDataSequenceNumber != null
+            ? newDataFilesDataSequenceNumber
+            : Integer.MAX_VALUE);
+  }
+
   @Override
   public Object updateEvent() {
     long snapshotId = snapshotId();

core/src/test/java/org/apache/iceberg/TestBase.java

@@ -122,6 +122,14 @@ public class TestBase {
           .withPartitionPath("data_bucket=1") // easy way to set partition data for now
           .withRecordCount(1)
           .build();
+  static final DeleteFile FILE_B_EQ_DELETES =
+      FileMetadata.deleteFileBuilder(SPEC)
+          .ofEqualityDeletes(1)
+          .withPath("/path/to/data-b2-deletes.parquet")
+          .withFileSizeInBytes(10)
+          .withPartitionPath("data_bucket=0")
+          .withRecordCount(1)
+          .build();
   static final DeleteFile FILE_B_DV =
       FileMetadata.deleteFileBuilder(SPEC)
           .ofPositionDeletes()

core/src/test/java/org/apache/iceberg/TestRewriteFiles.java

@@ -29,7 +29,9 @@
 import java.io.File;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
+import java.util.stream.Collectors;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
@@ -387,6 +389,83 @@ public void testRewriteDataAndAssignOldSequenceNumber() {
     assertThat(listManifestFiles()).hasSize(4);
   }
 
+  @TestTemplate
+  public void testRewriteDataAndAssignOldSequenceNumbersShouldNotDropDeleteFiles() {
+    assumeThat(formatVersion)
+        .as("Sequence number is only supported in iceberg format v2 or later")
+        .isGreaterThan(1);
+    assertThat(listManifestFiles()).isEmpty();
+
+    commit(table, table.newRowDelta().addRows(FILE_A).addDeletes(FILE_A2_DELETES), branch);
+
+    long firstCommitSequenceNumber = latestSnapshot(table, branch).sequenceNumber();
+
+    commit(table, table.newRowDelta().addRows(FILE_B).addDeletes(FILE_B_EQ_DELETES), branch);
+
+    long secondCommitSequenceNumber = latestSnapshot(table, branch).sequenceNumber();
+
+    TableMetadata base = readMetadata();
+    Snapshot baseSnap = latestSnapshot(base, branch);
+    long baseSnapshotId = baseSnap.snapshotId();
+
+    // FILE_B_EQ_DELETES and FILE_A2_DELETES should not be removed as the rewrite specifies
+    // `firstRewriteSequenceNumber` explicitly which is the same as that of FILE_A2_DELETES and before
+    // FILE_B_EQ_DELETES
+
+    // Technically FILE_A2_DELETES could be removed since it's an equality delete and should apply
+    // on data sequences strictly smaller, so it's no longer needed.
+    // However, MergingSnapshotProducer calls dropDeleteFilesOlderThan which doesn't consider if the
+    // file is an equality delete, if that API is changed the equality delete file could be dropped
+    // sooner
+    Snapshot pending =
+        apply(
+            table
+                .newRewrite()
+                .addFile(FILE_A2)
+                .deleteFile(FILE_A)
+                .dataSequenceNumber(firstCommitSequenceNumber),
+            branch);
+
+    assertThat(pending.allManifests(table.io())).hasSize(5);
+
+    long pendingId = pending.snapshotId();
+    List<ManifestFile> manifestFiles =
+        pending.allManifests(table.io()).stream()
+            .sorted(Comparator.comparingLong(ManifestFile::sequenceNumber).reversed())
+            .collect(Collectors.toList());
+
+    validateManifest(
+        manifestFiles.get(0),
+        dataSeqs(1L),
+        fileSeqs(secondCommitSequenceNumber + 1),
+        ids(pendingId),
+        files(FILE_A2),
+        statuses(ADDED));
+
+    validateManifestEntries(manifestFiles.get(1), ids(pendingId), files(FILE_A), statuses(DELETED));
+
+    validateManifestEntries(
+        manifestFiles.get(2), ids(baseSnapshotId), files(FILE_B), statuses(ADDED));
+
+    validateDeleteManifest(
+        manifestFiles.get(3),
+        dataSeqs(2L),
+        fileSeqs(2L),
+        ids(baseSnapshotId),
+        files(FILE_B_EQ_DELETES),
+        statuses(ADDED));
+
+    validateDeleteManifest(
+        manifestFiles.get(4),
+        dataSeqs(1L),
+        fileSeqs(1L),
+        ids(baseSnapshotId - 1),
+        files(FILE_A2_DELETES),
+        statuses(ADDED));
+
+    assertThat(listManifestFiles()).hasSize(6);
+  }
+
   @TestTemplate
   public void testFailure() {
     commit(table, table.newAppend().appendFile(FILE_A), branch);