Core: support rewrite data files with starting sequence number

api/src/main/java/org/apache/iceberg/RewriteFiles.java

@@ -51,6 +51,17 @@ default RewriteFiles rewriteFiles(Set<DataFile> filesToDelete, Set<DataFile> fil
     );
   }
 
+  /**
+   * Add a rewrite that replaces one set of data files with another set that contains the same data.
+   * The sequence number provided will be used for all the data files added.
+   *
+   * @param filesToDelete  files that will be replaced (deleted), cannot be null or empty.
+   * @param filesToAdd     files that will be added, cannot be null or empty.
+   * @param sequenceNumber sequence number to use for all data files added
+   * @return this for method chaining
+   */
+  RewriteFiles rewriteFiles(Set<DataFile> filesToDelete, Set<DataFile> filesToAdd, long sequenceNumber);
+
   /**
    * Add a rewrite that replaces one set of files with another set that contains the same data.
    *

api/src/main/java/org/apache/iceberg/actions/RewriteDataFiles.java

@@ -77,6 +77,17 @@ public interface RewriteDataFiles extends SnapshotUpdate<RewriteDataFiles, Rewri
    */
   String TARGET_FILE_SIZE_BYTES = "target-file-size-bytes";
 
+  /**
+   * If the compaction should use the sequence number of the snapshot at compaction start time for new data files,
+   * instead of using the sequence number of the newly produced snapshot.
+   * <p>
+   * This avoids commit conflicts with updates that add newer equality deletes at a higher sequence number.
+   * <p>
+   * Defaults to true.
+   */
+  String USE_STARTING_SEQUENCE_NUMBER = "use-starting-sequence-number";
+  boolean USE_STARTING_SEQUENCE_NUMBER_DEFAULT = true;
+
   /**
    * Choose BINPACK as a strategy for this rewrite operation
    * @return this for method chaining

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

@@ -21,6 +21,7 @@
 
 import java.util.Set;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 
 class BaseRewriteFiles extends MergingSnapshotProducer<RewriteFiles> implements RewriteFiles {
@@ -66,6 +67,12 @@ private void verifyInputAndOutputFiles(Set<DataFile> dataFilesToDelete, Set<Dele
     }
   }
 
+  @Override
+  public RewriteFiles rewriteFiles(Set<DataFile> filesToDelete, Set<DataFile> filesToAdd, long sequenceNumber) {
+    setNewFilesSequenceNumber(sequenceNumber);
+    return rewriteFiles(filesToDelete, ImmutableSet.of(), filesToAdd, ImmutableSet.of());
+  }
+
   @Override
   public RewriteFiles rewriteFiles(Set<DataFile> dataFilesToReplace, Set<DeleteFile> deleteFilesToReplace,
                                    Set<DataFile> dataFilesToAdd, Set<DeleteFile> deleteFilesToAdd) {

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

@@ -62,9 +62,13 @@ ManifestEntry<F> wrapExisting(Long newSnapshotId, Long newSequenceNumber, F newF
   }
 
   ManifestEntry<F> wrapAppend(Long newSnapshotId, F newFile) {
+    return wrapAppend(newSnapshotId, null, newFile);
+  }
+
+  ManifestEntry<F> wrapAppend(Long newSnapshotId, Long newSequenceNumber, F newFile) {
     this.status = Status.ADDED;
     this.snapshotId = newSnapshotId;
-    this.sequenceNumber = null;
+    this.sequenceNumber = newSequenceNumber;
     this.file = newFile;
     return this;
   }

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

@@ -103,6 +103,19 @@ public void add(F addedFile) {
     addEntry(reused.wrapAppend(snapshotId, addedFile));
   }
 
+  /**
+   * Add an added entry for a file with a specific sequence number.
+   * <p>
+   * The entry's snapshot ID will be this manifest's snapshot ID.
+   * The entry's sequence number will be the provided sequence number.
+   *
+   * @param addedFile a data file
+   * @param sequenceNumber sequence number for the data file
+   */
+  public void add(F addedFile, long sequenceNumber) {
+    addEntry(reused.wrapAppend(snapshotId, sequenceNumber, addedFile));
+  }
+
   void add(ManifestEntry<F> entry) {
     addEntry(reused.wrapAppend(snapshotId, entry.file()));
   }

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

@@ -21,6 +21,7 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.ListIterator;
 import java.util.Map;
@@ -80,6 +81,7 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
 
   // update data
   private final List<DataFile> newFiles = Lists.newArrayList();
+  private Long newFilesSequenceNumber;
   private final Map<Integer, List<DeleteFile>> newDeleteFilesBySpec = Maps.newHashMap();
   private final List<ManifestFile> appendManifests = Lists.newArrayList();
   private final List<ManifestFile> rewrittenAppendManifests = Lists.newArrayList();
@@ -297,7 +299,8 @@ protected void validateAddedDataFiles(TableMetadata base, Long startingSnapshotI
    */
   protected void validateNoNewDeletesForDataFiles(TableMetadata base, Long startingSnapshotId,
                                                   Iterable<DataFile> dataFiles) {
-    validateNoNewDeletesForDataFiles(base, startingSnapshotId, null, dataFiles, true);
+    validateNoNewDeletesForDataFiles(base, startingSnapshotId, null, dataFiles, true,
+        newFilesSequenceNumber != null);
   }
 
   /**
@@ -313,6 +316,28 @@ protected void validateNoNewDeletesForDataFiles(TableMetadata base, Long startin
   protected void validateNoNewDeletesForDataFiles(TableMetadata base, Long startingSnapshotId,
                                                   Expression dataFilter, Iterable<DataFile> dataFiles,
                                                   boolean caseSensitive) {
+    validateNoNewDeletesForDataFiles(base, startingSnapshotId, dataFilter, dataFiles, caseSensitive, false);
+  }
+
+  /**
+   * Validates that no new delete files that must be applied to the given data files have been added to the table since
+   * a starting snapshot, with the option to ignore equality deletes during the validation.
+   * <p>
+   * For example, in the case of rewriting data files, if the added data files have the same sequence number as the
+   * replaced data files, equality deletes added at a higher sequence number are still effective against the added
+   * data files, so there is no risk of commit conflict between RewriteFiles and RowDelta. In cases like this,
+   * validation against equality delete files can be omitted.
+   *
+   * @param base table metadata to validate
+   * @param startingSnapshotId id of the snapshot current at the start of the operation
+   * @param dataFilter a data filter
+   * @param dataFiles data files to validate have no new row deletes
+   * @param caseSensitive whether expression binding should be case-sensitive
+   * @param ignoreEqualityDeletes whether equality deletes should be ignored in validation
+   */
+  private void validateNoNewDeletesForDataFiles(TableMetadata base, Long startingSnapshotId,
+                                                Expression dataFilter, Iterable<DataFile> dataFiles,
+                                                boolean caseSensitive, boolean ignoreEqualityDeletes) {
     // if there is no current table state, no files have been added
     if (base.currentSnapshot() == null || base.formatVersion() < 2) {
       return;
@@ -327,8 +352,14 @@ protected void validateNoNewDeletesForDataFiles(TableMetadata base, Long startin
 
     for (DataFile dataFile : dataFiles) {
       // if any delete is found that applies to files written in or before the starting snapshot, fail
-      if (deletes.forDataFile(startingSequenceNumber, dataFile).length > 0) {
-        throw new ValidationException("Cannot commit, found new delete for replaced data file: %s", dataFile);
+      DeleteFile[] deleteFiles = deletes.forDataFile(startingSequenceNumber, dataFile);
+      if (ignoreEqualityDeletes) {
+        ValidationException.check(
+            Arrays.stream(deleteFiles).noneMatch(deleteFile -> deleteFile.content() == FileContent.POSITION_DELETES),
+            "Cannot commit, found new position delete for replaced data file: %s", dataFile);
+      } else {
+        ValidationException.check(deleteFiles.length == 0,
+            "Cannot commit, found new delete for replaced data file: %s", dataFile);
       }
     }
   }
@@ -360,6 +391,10 @@ protected void validateNoNewDeleteFiles(TableMetadata base, Long startingSnapsho
         dataFilter, Iterables.transform(deletes.referencedDeleteFiles(), ContentFile::path));
   }
 
+  protected void setNewFilesSequenceNumber(long sequenceNumber) {
+    this.newFilesSequenceNumber = sequenceNumber;
+  }
+
   private long startingSequenceNumber(TableMetadata metadata, Long staringSnapshotId) {
     if (staringSnapshotId != null && metadata.snapshot(staringSnapshotId) != null) {
       Snapshot startingSnapshot = metadata.snapshot(staringSnapshotId);
@@ -591,7 +626,11 @@ private ManifestFile newFilesAsManifest() {
       try {
         ManifestWriter<DataFile> writer = newManifestWriter(dataSpec());
         try {
-          writer.addAll(newFiles);
+          if (newFilesSequenceNumber == null) {
+            writer.addAll(newFiles);
+          } else {
+            newFiles.forEach(f -> writer.add(f, newFilesSequenceNumber));
+          }
         } finally {
           writer.close();
         }

core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java

@@ -48,15 +48,21 @@ public class RewriteDataFilesCommitManager {
 
   private final Table table;
   private final long startingSnapshotId;
+  private final boolean useStartingSequenceNumber;
 
   // constructor used for testing
   public RewriteDataFilesCommitManager(Table table) {
     this(table, table.currentSnapshot().snapshotId());
   }
 
   public RewriteDataFilesCommitManager(Table table, long startingSnapshotId) {
+    this(table, startingSnapshotId, RewriteDataFiles.USE_STARTING_SEQUENCE_NUMBER_DEFAULT);
+  }
+
+  public RewriteDataFilesCommitManager(Table table, long startingSnapshotId, boolean useStartingSequenceNumber) {
     this.table = table;
     this.startingSnapshotId = startingSnapshotId;
+    this.useStartingSequenceNumber = useStartingSequenceNumber;
   }
 
   /**
@@ -72,9 +78,14 @@ public void commitFileGroups(Set<RewriteFileGroup> fileGroups) {
       addedDataFiles = Sets.union(addedDataFiles, group.addedFiles());
     }
 
-    RewriteFiles rewrite = table.newRewrite()
-        .validateFromSnapshot(startingSnapshotId)
-        .rewriteFiles(rewrittenDataFiles, addedDataFiles);
+    RewriteFiles rewrite = table.newRewrite().validateFromSnapshot(startingSnapshotId);
+    if (useStartingSequenceNumber) {
+      long sequenceNumber = table.snapshot(startingSnapshotId).sequenceNumber();
+      rewrite.rewriteFiles(rewrittenDataFiles, addedDataFiles, sequenceNumber);
+    } else {
+      rewrite.rewriteFiles(rewrittenDataFiles, addedDataFiles);
+    }
+
     rewrite.commit();
   }
 

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

@@ -469,6 +469,17 @@ protected DeleteFile newDeleteFile(int specId, String partitionPath) {
         .build();
   }
 
+  protected DeleteFile newEqualityDeleteFile(int specId, String partitionPath, int... fieldIds) {
+    PartitionSpec spec = table.specs().get(specId);
+    return FileMetadata.deleteFileBuilder(spec)
+        .ofEqualityDeletes(fieldIds)
+        .withPath("/path/to/delete-" + UUID.randomUUID() + ".parquet")
+        .withFileSizeInBytes(10)
+        .withPartitionPath(partitionPath)
+        .withRecordCount(1)
+        .build();
+  }
+
   protected <T> PositionDelete<T> positionDelete(CharSequence path, long pos, T row) {
     PositionDelete<T> positionDelete = PositionDelete.create();
     return positionDelete.set(path, pos, row);

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

@@ -19,13 +19,16 @@
 
 package org.apache.iceberg;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.List;
 import java.util.UUID;
 import org.apache.iceberg.ManifestEntry.Status;
+import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.types.Conversions;
 import org.apache.iceberg.types.Types;
 import org.junit.Assert;
+import org.junit.Assume;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -87,6 +90,24 @@ public void testManifestPartitionStats() throws IOException {
         Conversions.fromByteBuffer(Types.IntegerType.get(), partitionFieldSummary.upperBound()));
   }
 
+  @Test
+  public void testWriteManifestWithSequenceNumber() throws IOException {
+    Assume.assumeTrue("sequence number is only valid for format version > 1", formatVersion > 1);
+    File manifestFile = temp.newFile("manifest.avro");
+    Assert.assertTrue(manifestFile.delete());
+    OutputFile outputFile = table.ops().io().newOutputFile(manifestFile.getCanonicalPath());
+    ManifestWriter<DataFile> writer = ManifestFiles.write(formatVersion, table.spec(), outputFile, 1L);
+    writer.add(newFile(10, TestHelpers.Row.of(1)), 1000L);
+    writer.close();
+    ManifestFile manifest = writer.toManifestFile();
+    Assert.assertEquals("Manifest should have no sequence number", -1L, manifest.sequenceNumber());
+    ManifestReader<DataFile> manifestReader = ManifestFiles.read(manifest, table.io());
+    for (ManifestEntry<DataFile> entry : manifestReader.entries()) {
+      Assert.assertEquals("Custom sequence number should be used for all manifest entries",
+          1000L, (long) entry.sequenceNumber());
+    }
+  }
+
   private DataFile newFile(long recordCount) {
     return newFile(recordCount, null);
   }

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

@@ -264,10 +264,75 @@ public void testRewriteDataAndDeleteFiles() {
         files(FILE_A_DELETES, FILE_B_DELETES),
         statuses(DELETED, EXISTING));
 
-    // We should only get the 3 manifests that this test is expected to add.
+    // We should only get the 5 manifests that this test is expected to add.
     Assert.assertEquals("Only 5 manifests should exist", 5, listManifestFiles().size());
   }
 
+  @Test
+  public void testRewriteDataAndAssignOldSequenceNumber() {
+    Assume.assumeTrue("Sequence number is only supported in iceberg format v2. ", formatVersion > 1);
+    Assert.assertEquals("Table should start empty", 0, listManifestFiles().size());
+
+    table.newRowDelta()
+        .addRows(FILE_A)
+        .addRows(FILE_B)
+        .addRows(FILE_C)
+        .addDeletes(FILE_A_DELETES)
+        .addDeletes(FILE_B_DELETES)
+        .commit();
+
+    TableMetadata base = readMetadata();
+    Snapshot baseSnap = base.currentSnapshot();
+    long baseSnapshotId = baseSnap.snapshotId();
+    Assert.assertEquals("Should create 2 manifests for initial write", 2, baseSnap.allManifests().size());
+    List<ManifestFile> initialManifests = baseSnap.allManifests();
+
+    validateManifestEntries(initialManifests.get(0),
+        ids(baseSnapshotId, baseSnapshotId, baseSnapshotId),
+        files(FILE_A, FILE_B, FILE_C),
+        statuses(ADDED, ADDED, ADDED));
+    validateDeleteManifest(initialManifests.get(1),
+        seqs(1, 1),
+        ids(baseSnapshotId, baseSnapshotId),
+        files(FILE_A_DELETES, FILE_B_DELETES),
+        statuses(ADDED, ADDED));
+
+    // Rewrite the files.
+    long oldSequenceNumber = table.currentSnapshot().sequenceNumber();
+    Snapshot pending = table.newRewrite()
+        .validateFromSnapshot(table.currentSnapshot().snapshotId())
+        .rewriteFiles(ImmutableSet.of(FILE_A), ImmutableSet.of(FILE_D), oldSequenceNumber)
+        .apply();
+
+    Assert.assertEquals("Should contain 3 manifest", 3, pending.allManifests().size());
+    Assert.assertFalse("Should not contain data manifest from initial write",
+        pending.dataManifests().stream().anyMatch(initialManifests::contains));
+
+    long pendingId = pending.snapshotId();
+    ManifestFile newManifest = pending.allManifests().get(0);
+    validateManifestEntries(newManifest, ids(pendingId), files(FILE_D), statuses(ADDED));
+    for (ManifestEntry<DataFile> entry : ManifestFiles.read(newManifest, FILE_IO).entries()) {
+      Assert.assertEquals("Should have old sequence number for manifest entries",
+          oldSequenceNumber, (long) entry.sequenceNumber());
+    }
+    Assert.assertEquals("Should use new sequence number for the manifest file",
+        oldSequenceNumber + 1, newManifest.sequenceNumber());
+
+    validateManifestEntries(pending.allManifests().get(1),
+        ids(pendingId, baseSnapshotId, baseSnapshotId),
+        files(FILE_A, FILE_B, FILE_C),
+        statuses(DELETED, EXISTING, EXISTING));
+
+    validateDeleteManifest(pending.allManifests().get(2),
+        seqs(1, 1),
+        ids(baseSnapshotId, baseSnapshotId),
+        files(FILE_A_DELETES, FILE_B_DELETES),
+        statuses(ADDED, ADDED));
+
+    // We should only get the 4 manifests that this test is expected to add.
+    Assert.assertEquals("Only 4 manifests should exist", 4, listManifestFiles().size());
+  }
+
   @Test
   public void testFailure() {
     table.newAppend()