Core: Write rewrite manifests concurrently but commit one at a time

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

@@ -293,7 +293,7 @@
   }
 
   private static void validateDeleteFileForVersion(DeleteFile file, int formatVersion) {
     switch (formatVersion) {
       case 1:
         throw new IllegalArgumentException("Deletes are supported in V2 and above");
       case 2:
@@ -1180,13 +1180,13 @@
       // this triggers a rewrite of all delete manifests even if there is only one new delete file
       // if there is a relevant use case in the future, the behavior can be optimized
       cachedNewDeleteManifests.clear();
-      // On cache invalidation of delete files, clear the summary because any new DV could require a
-      // merge,
-      // and the summary cannot be generated until after merging is complete.
-      addedDeleteFilesSummary.clear();
     }
 
     if (cachedNewDeleteManifests.isEmpty()) {
+      // Rebuild the summary with the manifests because any new DV could require a merge, and the
+      // summary cannot be generated until after merging is complete. This also discards entries
+      // left by a previous attempt that failed to write its manifests.
+      addedDeleteFilesSummary.clear();
       List<DeleteFile> mergedDVs = mergeDVs();
       Map<Integer, List<DeleteFile>> newDeleteFilesBySpec =
           Streams.stream(Iterables.concat(mergedDVs, DeleteFileSet.of(v2Deletes)))

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

@@ -41,6 +41,7 @@ public class RewriteDataFilesCommitManager {
   private final boolean useStartingSequenceNumber;
   private final Map<String, String> snapshotProperties;
   private final String branch;
+  private final Object commitLock = new Object();
 
   // constructor used for testing
   public RewriteDataFilesCommitManager(Table table) {
@@ -109,7 +110,18 @@ public void commitFileGroups(Set<RewriteFileGroup> fileGroups) {
       rewrite.toBranch(branch);
     }
 
-    rewrite.commit();
+    // Commits through this manager conflict with each other, so they are serialized. Calling
+    // apply() first writes manifests outside the lock for commit() to reuse. A failure is left to
+    // commit(), which applies again and either succeeds or fails and cleans up.
+    try {
+      rewrite.apply();
+    } catch (RuntimeException e) {
+      LOG.debug("Failed to apply rewrite before committing, applying again in commit", e);
+    }
+
+    synchronized (commitLock) {
+      rewrite.commit();
+    }
   }
 
   /**

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

@@ -44,6 +44,7 @@ public class RewritePositionDeletesCommitManager {
   private final Table table;
   private final long startingSnapshotId;
   private final Map<String, String> snapshotProperties;
+  private final Object commitLock = new Object();
 
   public RewritePositionDeletesCommitManager(Table table) {
     this(table, ImmutableMap.of());
@@ -76,7 +77,18 @@ public void commit(Set<RewritePositionDeletesGroup> fileGroups) {
 
     snapshotProperties.forEach(rewriteFiles::set);
 
-    rewriteFiles.commit();
+    // Commits through this manager conflict with each other, so they are serialized. Calling
+    // apply() first writes manifests outside the lock for commit() to reuse. A failure is left to
+    // commit(), which applies again and either succeeds or fails and cleans up.
+    try {
+      rewriteFiles.apply();
+    } catch (RuntimeException e) {
+      LOG.debug("Failed to apply rewrite before committing, applying again in commit", e);
+    }
+
+    synchronized (commitLock) {
+      rewriteFiles.commit();
+    }
   }
 
   /**

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

@@ -31,12 +31,14 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.Assumptions.assumeThat;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
@@ -48,12 +50,14 @@
 import org.apache.iceberg.encryption.EncryptingFileIO;
 import org.apache.iceberg.encryption.EncryptionManager;
 import org.apache.iceberg.encryption.EncryptionTestHelpers;
+import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.expressions.ManifestEvaluator;
 import org.apache.iceberg.expressions.Projections;
 import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.io.OutputFileFactory;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
@@ -96,6 +100,45 @@ public void addOnlyDeleteFilesProducesDeleteOperation() {
         .containsEntry(SnapshotSummary.REPLACED_MANIFESTS_COUNT, "0");
   }
 
+  @TestTemplate
+  void addedDeletesCountedOnceAfterFailedManifestWrite() {
+    AtomicBoolean failNextManifest = new AtomicBoolean(false);
+    TestTables.LocalFileIO io =
+        new TestTables.LocalFileIO() {
+          @Override
+          public OutputFile newOutputFile(String path) {
+            boolean isManifestList = new File(path).getName().startsWith("snap-");
+            if (!isManifestList && failNextManifest.compareAndSet(true, false)) {
+              throw new RuntimeIOException("Injected failure writing %s", path);
+            }
+
+            return super.newOutputFile(path);
+          }
+        };
+    String tableName = "failing-io-" + branch;
+    File location = temp.resolve(tableName).toFile();
+    TestTables.TestTable failingTable =
+        TestTables.create(
+            location,
+            tableName,
+            SCHEMA,
+            SPEC,
+            SortOrder.unsorted(),
+            formatVersion,
+            new TestTables.TestTableOperations(tableName, location, io));
+
+    RowDelta rowDelta = failingTable.newRowDelta().addDeletes(fileADeletes());
+    failNextManifest.set(true);
+    assertThatThrownBy(() -> apply(rowDelta, branch))
+        .isInstanceOf(RuntimeIOException.class)
+        .hasMessageContaining("Injected failure writing");
+
+    Snapshot snap = commit(failingTable, rowDelta, branch);
+    assertThat(snap.summary())
+        .containsEntry(ADDED_DELETE_FILES_PROP, "1")
+        .containsEntry(TOTAL_DELETE_FILES_PROP, "1");
+  }
+
   @TestTemplate
   public void addOnlyDataFilesProducesAppendOperation() {
     SnapshotUpdate<?> rowDelta = table.newRowDelta().addRows(FILE_A).addRows(FILE_B);

core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java

@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.actions;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TestBase;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+class TestRewriteDataFilesCommitManager extends TestBase {
+
+  @TestTemplate
+  void concurrentCommitsDoNotConflict() throws Exception {
+    int numGroups = 8;
+    table
+        .updateProperties()
+        // fail on the first conflict instead of hiding it behind a retry
+        .set(TableProperties.COMMIT_NUM_RETRIES, "0")
+        // test tables otherwise assign sequential snapshot IDs, and a reused ID drops a commit
+        .set("random-snapshot-ids", "true")
+        .commit();
+    AppendFiles append = table.newFastAppend();
+    for (int i = 0; i < numGroups; i++) {
+      append.appendFile(copy(FILE_A, "/path/to/data-" + i + ".parquet"));
+    }
+
+    append.commit();
+
+    List<RewriteFileGroup> groups = Lists.newArrayList();
+    try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
+      for (FileScanTask task : tasks) {
+        DataFile output = copy(task.file(), task.file().location() + ".rewritten");
+        groups.add(group(List.of(task), output));
+      }
+    }
+
+    RewriteDataFilesCommitManager commitManager = new RewriteDataFilesCommitManager(table);
+    CyclicBarrier start = new CyclicBarrier(numGroups);
+    ExecutorService executorService = Executors.newFixedThreadPool(numGroups);
+    try {
+      List<Future<?>> commits = Lists.newArrayList();
+      for (RewriteFileGroup group : groups) {
+        commits.add(
+            executorService.submit(
+                () -> {
+                  start.await();
+                  commitManager.commitFileGroups(Set.of(group));
+                  return null;
+                }));
+      }
+
+      for (Future<?> commit : commits) {
+        commit.get(10, TimeUnit.SECONDS);
+      }
+    } finally {
+      executorService.shutdownNow();
+    }
+
+    assertThat(table.snapshots()).hasSize(numGroups + 1);
+    assertThat(metadataFiles()).containsExactlyInAnyOrderElementsOf(referencedMetadataFiles());
+  }
+
+  @TestTemplate
+  void failedCommitRemovesWrittenManifests() throws IOException {
+    table.newAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+    RewriteDataFilesCommitManager commitManager = new RewriteDataFilesCommitManager(table);
+    RewriteFileGroup group = group(planFiles(), FILE_C);
+    table.newDelete().deleteFile(FILE_B).commit();
+    List<String> metadataFilesBeforeCommit = metadataFiles();
+
+    // apply() rewrites the manifest holding FILE_A before the missing FILE_B fails validation
+    assertThatThrownBy(() -> commitManager.commitFileGroups(Set.of(group)))
+        .isInstanceOf(ValidationException.class)
+        .hasMessageStartingWith("Missing required files to delete");
+
+    assertThat(metadataFiles()).containsExactlyInAnyOrderElementsOf(metadataFilesBeforeCommit);
+  }
+
+  private List<FileScanTask> planFiles() throws IOException {
+    try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
+      return Lists.newArrayList(tasks);
+    }
+  }
+
+  private RewriteFileGroup group(List<FileScanTask> tasks, DataFile outputFile) {
+    RewriteFileGroup group =
+        new RewriteFileGroup(
+            ImmutableRewriteDataFiles.FileGroupInfo.builder()
+                .globalIndex(0)
+                .partitionIndex(0)
+                .partition(outputFile.partition())
+                .build(),
+            tasks,
+            table.spec().specId(),
+            Long.MAX_VALUE,
+            Long.MAX_VALUE,
+            1);
+    group.setOutputFiles(Set.of(outputFile));
+    return group;
+  }
+
+  private static DataFile copy(DataFile file, String path) {
+    return DataFiles.builder(SPEC).copy(file).withPath(path).build();
+  }
+
+  private List<String> metadataFiles() {
+    return Arrays.stream(metadataDir.listFiles()).map(File::getName).toList();
+  }
+
+  private Set<String> referencedMetadataFiles() {
+    Set<String> referencedFiles = Sets.newHashSet();
+    for (Snapshot snapshot : table.snapshots()) {
+      referencedFiles.add(new File(snapshot.manifestListLocation()).getName());
+      for (ManifestFile manifest : snapshot.allManifests(table.io())) {
+        referencedFiles.add(new File(manifest.path()).getName());
+      }
+    }
+
+    return referencedFiles;
+  }
+}