From 62bfb8f2916b50cc8d28c0d2aae0dc1c0ea535ee Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar <186172142+smaheshwar-pltr@users.noreply.github.com> Date: Wed, 26 Aug 2026 02:32:29 +0100 Subject: [PATCH 1/7] Core: Serialize commits in BaseCommitService Give one offering thread non-blocking drain ownership so ready groups never commit concurrently and competing offers do not wait. Reject non-positive batch sizes so the drain condition always makes progress. Generated-by: OpenAI Codex --- .../iceberg/actions/BaseCommitService.java | 74 +++- .../RewriteDataFilesCommitManager.java | 8 +- .../RewritePositionDeletesCommitManager.java | 8 +- .../iceberg/actions/TestCommitService.java | 401 ++++++++++++++++-- .../actions/TestRewriteDataFilesAction.java | 11 +- 5 files changed, 417 insertions(+), 85 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/actions/BaseCommitService.java b/core/src/main/java/org/apache/iceberg/actions/BaseCommitService.java index 4c79781efb07..8c846edcb929 100644 --- a/core/src/main/java/org/apache/iceberg/actions/BaseCommitService.java +++ b/core/src/main/java/org/apache/iceberg/actions/BaseCommitService.java @@ -27,6 +27,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; @@ -37,9 +38,10 @@ import org.slf4j.LoggerFactory; /** - * An async service which allows for committing multiple file groups as their rewrites complete. The - * service also allows for partial-progress since commits can fail. Once the service has been closed - * no new file groups should not be offered. + * A service that commits file groups as rewrites complete. Commits are serialized and may run in a + * thread offering a file group or in the service executor during close. Other offering threads + * continue without waiting for an active commit. Failed commits permit partial progress. No groups + * may be offered after the service is closed. * *

Specific implementations provide implementations for {@link #commitOrClean(Set)} and {@link * #abortFileGroup(Object)} @@ -59,13 +61,14 @@ abstract class BaseCommitService implements Closeable { private final int rewritesPerCommit; private final AtomicBoolean running = new AtomicBoolean(false); private final long timeoutInMS; + private final ReentrantLock commitLock = new ReentrantLock(); private int succeededCommits = 0; /** * Constructs a {@link BaseCommitService} * * @param table table to perform commit on - * @param rewritesPerCommit number of file groups to include in a commit + * @param rewritesPerCommit positive number of file groups to include in a commit */ BaseCommitService(Table table, int rewritesPerCommit) { this(table, rewritesPerCommit, TIMEOUT_IN_MS_DEFAULT); @@ -75,11 +78,15 @@ abstract class BaseCommitService implements Closeable { * Constructs a {@link BaseCommitService} * * @param table table to perform commit on - * @param rewritesPerCommit number of file groups to include in a commit + * @param rewritesPerCommit positive number of file groups to include in a commit * @param timeoutInMS The timeout to wait for commits to complete after all rewrite jobs have been * completed */ BaseCommitService(Table table, int rewritesPerCommit, long timeoutInMS) { + Preconditions.checkArgument( + rewritesPerCommit > 0, + "Invalid number of file groups per commit: %s (must be positive)", + rewritesPerCommit); this.table = table; LOG.info( "Creating commit service for table {} with {} groups per commit", table, rewritesPerCommit); @@ -209,29 +216,50 @@ public void close() { } private void commitReadyCommitGroups() { - Set batch = null; - if (canCreateCommitGroup()) { - synchronized (completedRewrites) { - if (canCreateCommitGroup()) { - batch = Sets.newHashSetWithExpectedSize(rewritesPerCommit); - for (int i = 0; i < rewritesPerCommit && !completedRewrites.isEmpty(); i++) { - batch.add(completedRewrites.poll()); + while (canCreateCommitGroup()) { + if (!commitLock.tryLock()) { + return; + } + + try { + while (canCreateCommitGroup()) { + Set batch = null; + String inProgressCommitToken = null; + try { + synchronized (completedRewrites) { + if (canCreateCommitGroup()) { + batch = Sets.newHashSetWithExpectedSize(rewritesPerCommit); + // The committer may exit as soon as both queues are empty. + inProgressCommitToken = UUID.randomUUID().toString(); + inProgressCommits.add(inProgressCommitToken); + for (int i = 0; i < rewritesPerCommit && !completedRewrites.isEmpty(); i++) { + batch.add(completedRewrites.poll()); + } + } + } + + if (batch != null) { + commitBatch(batch); + } + } finally { + if (inProgressCommitToken != null) { + inProgressCommits.remove(inProgressCommitToken); + } } } + } finally { + commitLock.unlock(); } } + } - if (batch != null) { - String inProgressCommitToken = UUID.randomUUID().toString(); - inProgressCommits.add(inProgressCommitToken); - try { - commitOrClean(batch); - committedRewrites.addAll(batch); - succeededCommits++; - } catch (Exception e) { - LOG.error("Failure during rewrite commit process, partial progress enabled. Ignoring", e); - } - inProgressCommits.remove(inProgressCommitToken); + private void commitBatch(Set batch) { + try { + commitOrClean(batch); + committedRewrites.addAll(batch); + succeededCommits++; + } catch (Exception e) { + LOG.error("Failure during rewrite commit process, partial progress enabled. Ignoring", e); } } diff --git a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java index 727bd6ab84ef..1d2d637d70e1 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java +++ b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java @@ -150,11 +150,11 @@ public void commitOrClean(Set rewriteGroups) { } /** - * An async service which allows for committing multiple file groups as their rewrites complete. - * The service also allows for partial-progress since commits can fail. Once the service has been - * closed no new file groups should not be offered. + * A service that commits file groups as rewrites complete. Commits are serialized and may run in + * the thread offering a file group. Failed commits permit partial progress. No groups may be + * offered after the service is closed. * - * @param rewritesPerCommit number of file groups to include in a commit + * @param rewritesPerCommit positive number of file groups to include in a commit * @return the service for handling commits */ public CommitService service(int rewritesPerCommit) { diff --git a/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java index 31b7f36af150..17277f2b8186 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java +++ b/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java @@ -115,11 +115,11 @@ public void commitOrClean(Set rewriteGroups) { } /** - * An async service which allows for committing multiple file groups as their rewrites complete. - * The service also allows for partial-progress since commits can fail. Once the service has been - * closed no new file groups should not be offered. + * A service that commits file groups as rewrites complete. Commits are serialized and may run in + * the thread offering a file group. Failed commits permit partial progress. No groups may be + * offered after the service is closed. * - * @param rewritesPerCommit number of file groups to include in a commit + * @param rewritesPerCommit positive number of file groups to include in a commit * @return the service for handling commits */ public CommitService service(int rewritesPerCommit) { diff --git a/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java b/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java index a40e7f617e7b..5a6918c72132 100644 --- a/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java +++ b/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java @@ -26,9 +26,13 @@ import java.util.Arrays; import java.util.List; import java.util.Set; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; import org.apache.iceberg.ParameterizedTestExtension; import org.apache.iceberg.Parameters; @@ -42,12 +46,23 @@ @ExtendWith(ParameterizedTestExtension.class) public class TestCommitService extends TestBase { + private static final int BATCH_EXTRACTION_TIMEOUT_MS = 1000; @Parameters(name = "formatVersion = {0}") protected static List parameters() { return Arrays.asList(1); } + @TestTemplate + void rejectsNonPositiveCommitGroupSize() { + assertThatThrownBy(() -> new CustomCommitService(table, 0, 10000)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid number of file groups per commit: 0 (must be positive)"); + assertThatThrownBy(() -> new CustomCommitService(table, -1, 10000)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid number of file groups per commit: -1 (must be positive)"); + } + @TestTemplate public void testCommittedResultsCorrectly() { CustomCommitService commitService = new CustomCommitService(table, 5, 10000); @@ -64,76 +79,214 @@ public void testCommittedResultsCorrectly() { } @TestTemplate - public void testAbortFileGroupsAfterTimeout() { - CustomCommitService commitService = new CustomCommitService(table, 5, 200); + public void testAbortFileGroupsAfterTimeout() throws Exception { + BlockingCommitService commitService = new BlockingCommitService(table, 5, 100); commitService.start(); - // Add file groups [0-3] for commit. - // There are less than the rewritesPerCommit, and thus will not trigger a commit action. Those - // file groups will be added to the completedRewrites queue. - // Now the queue has 4 file groups that need to commit. - for (int i = 0; i < 4; i++) { - commitService.offer(i); - } - - // Add file groups [4-7] for commit - // These are gated to not be able to commit, so all those 4 file groups will be added to the - // queue as well. - // Now the queue has 8 file groups that need to commit. CustomCommitService spyCommitService = spy(commitService); doReturn(false).when(spyCommitService).canCreateCommitGroup(); - for (int i = 4; i < 8; i++) { + for (int i = 0; i < 7; i++) { spyCommitService.offer(i); } - // close commitService. - // This allows committerService thread to start to commit the remaining file groups [0-7] in the - // completedRewrites queue. And also the main thread waits for the committerService thread to - // finish within a timeout. + ExecutorService offerExecutor = Executors.newSingleThreadExecutor(); + ExecutorService closeExecutor = Executors.newSingleThreadExecutor(); + try { + Future commitResult = offerExecutor.submit(() -> commitService.offer(7)); + assertThat(commitService.commitStarted.await(5, TimeUnit.SECONDS)).isTrue(); - // The committerService thread commits file groups [0-4]. These will wait a fixed duration to - // simulate timeout on the main thread, which then tries to abort file groups [5-7]. - // This tests the race conditions, as the committerService is also trying to commit groups - // [5-7]. - assertThatThrownBy(commitService::close) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Timeout occurred when waiting for commits") - .hasMessageNotContaining("{}") - .hasMessageMatching( - ".*\\d+ file groups committed\\. \\d+ file groups remain uncommitted\\..*"); + Future closeResult = + closeExecutor.submit( + () -> { + try { + commitService.close(); + return null; + } catch (Throwable e) { + return e; + } + }); + + assertThat(closeResult.get(5, TimeUnit.SECONDS)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Timeout occurred when waiting for commits") + .hasMessageNotContaining("{}") + .hasMessageMatching( + ".*\\d+ file groups committed\\. \\d+ file groups remain uncommitted\\..*"); + assertThat(commitService.aborted).containsExactlyInAnyOrder(5, 6, 7); + commitService.releaseCommit.countDown(); + commitResult.get(5, TimeUnit.SECONDS); + } finally { + commitService.releaseCommit.countDown(); + offerExecutor.shutdownNow(); + closeExecutor.shutdownNow(); + closeQuietly(commitService); + } + + assertThat(offerExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + assertThat(closeExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); - // Wait for the commitService to finish. Committed all file groups or aborted remaining file - // groups. Awaitility.await() .atMost(5, TimeUnit.SECONDS) .pollInSameThread() .untilAsserted(() -> assertThat(commitService.completedRewritesAllCommitted()).isTrue()); - if (commitService.aborted.isEmpty()) { - // All file groups are committed - assertThat(commitService.results()).containsExactly(0, 1, 2, 3, 4, 5, 6, 7); - } else { - // File groups [5-7] are aborted - assertThat(commitService.results()).doesNotContainAnyElementsOf(commitService.aborted); - assertThat(commitService.results()).containsExactly(0, 1, 2, 3, 4); - assertThat(commitService.aborted).containsExactly(5, 6, 7); + assertThat(commitService.results()).containsExactlyInAnyOrder(0, 1, 2, 3, 4); + } + + @TestTemplate + void closeReportsTimeoutWhileBatchExtractionIsInProgress() throws Exception { + BlockingHashCodeCommitService commitService = + new BlockingHashCodeCommitService(table, 1, BATCH_EXTRACTION_TIMEOUT_MS); + BlockingHashCodeGroup group = new BlockingHashCodeGroup(); + commitService.start(); + + ExecutorService offerExecutor = Executors.newSingleThreadExecutor(); + ExecutorService closeExecutor = Executors.newSingleThreadExecutor(); + try { + Future commitResult = offerExecutor.submit(() -> commitService.offer(group)); + assertThat(group.hashCodeStarted.await(5, TimeUnit.SECONDS)).isTrue(); + + Future closeResult = + closeExecutor.submit( + () -> { + try { + commitService.close(); + return null; + } catch (Throwable e) { + return e; + } + }); + + assertThat(closeResult.get(5, TimeUnit.SECONDS)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Timeout occurred when waiting for commits"); + group.releaseHashCode.countDown(); + commitResult.get(5, TimeUnit.SECONDS); + } finally { + group.releaseHashCode.countDown(); + offerExecutor.shutdownNow(); + closeExecutor.shutdownNow(); + closeQuietly(commitService); + } + + assertThat(offerExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + assertThat(closeExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + assertThat(commitService.results()).containsExactly(group); + } + + @TestTemplate + void failedBatchExtractionDoesNotStrandClose() { + ThrowingHashCodeCommitService commitService = + new ThrowingHashCodeCommitService(table, 1, BATCH_EXTRACTION_TIMEOUT_MS); + commitService.start(); + + boolean closed = false; + try { + assertThatThrownBy(() -> commitService.offer(new ThrowingHashCodeGroup())) + .isInstanceOf(IllegalStateException.class) + .hasMessage("Cannot hash file group"); + + commitService.close(); + closed = true; + } finally { + if (!closed) { + closeQuietly(commitService); + } } } + @TestTemplate + void serializesCommitsWithoutBlockingOtherOffers() throws Exception { + TrackingCommitService commitService = new TrackingCommitService(table); + commitService.start(); + + ExecutorService executorService = Executors.newFixedThreadPool(2); + boolean closed = false; + try { + Future firstCommit = executorService.submit(() -> commitService.offer(0)); + assertThat(commitService.firstCommitStarted.await(5, TimeUnit.SECONDS)).isTrue(); + + Future secondCommit = executorService.submit(() -> commitService.offer(1)); + secondCommit.get(5, TimeUnit.SECONDS); + assertThat(commitService.secondCommitStarted.getCount()).isEqualTo(1); + assertThat(commitService.maxConcurrentCommits.get()).isEqualTo(1); + + commitService.releaseCommits.countDown(); + firstCommit.get(5, TimeUnit.SECONDS); + assertThat(commitService.secondCommitStarted.await(5, TimeUnit.SECONDS)).isTrue(); + commitService.close(); + closed = true; + } finally { + commitService.releaseCommits.countDown(); + executorService.shutdownNow(); + if (!closed) { + closeQuietly(commitService); + } + } + + assertThat(executorService.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + assertThat(commitService.results()).containsExactlyInAnyOrder(0, 1); + } + + @TestTemplate + void drainOwnerProcessesBatchOfferedDuringHandoff() throws Exception { + HandoffCommitService commitService = new HandoffCommitService(table); + commitService.start(); + + ExecutorService executorService = Executors.newFixedThreadPool(2); + boolean closed = false; + try { + Future firstCommit = executorService.submit(() -> commitService.offer(0)); + assertThat(commitService.firstCommitStarted.await(5, TimeUnit.SECONDS)).isTrue(); + commitService.releaseFirstCommit.countDown(); + assertThat(commitService.ownerObservedEmptyQueue.await(5, TimeUnit.SECONDS)).isTrue(); + + Future handoffOffer = executorService.submit(() -> commitService.offer(1)); + handoffOffer.get(5, TimeUnit.SECONDS); + assertThat(commitService.secondCommitStarted.getCount()).isEqualTo(1); + + commitService.releaseOwner.countDown(); + assertThat(commitService.secondCommitStarted.await(5, TimeUnit.SECONDS)).isTrue(); + firstCommit.get(5, TimeUnit.SECONDS); + commitService.close(); + closed = true; + } finally { + commitService.releaseFirstCommit.countDown(); + commitService.releaseOwner.countDown(); + executorService.shutdownNow(); + if (!closed) { + closeQuietly(commitService); + } + } + + assertThat(executorService.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + assertThat(commitService.results()).containsExactlyInAnyOrder(0, 1); + } + private static class CustomCommitService extends BaseCommitService { + CustomCommitService(Table table, int rewritesPerCommit, int timeoutInMS) { + super(table, rewritesPerCommit, timeoutInMS); + } + + @Override + protected void commitOrClean(Set batch) {} + + @Override + protected void abortFileGroup(Integer group) {} + } + + private static class BlockingCommitService extends CustomCommitService { private final Set aborted = Sets.newConcurrentHashSet(); + private final CountDownLatch commitStarted = new CountDownLatch(1); + private final CountDownLatch releaseCommit = new CountDownLatch(1); - CustomCommitService(Table table, int rewritesPerCommit, int timeoutInSeconds) { - super(table, rewritesPerCommit, timeoutInSeconds); + BlockingCommitService(Table table, int rewritesPerCommit, int timeoutInMS) { + super(table, rewritesPerCommit, timeoutInMS); } @Override protected void commitOrClean(Set batch) { - try { - // Slightly longer than timeout - Thread.sleep(210); - } catch (InterruptedException e) { - throw new RuntimeException(e); - } + commitStarted.countDown(); + await(releaseCommit); } @Override @@ -141,4 +294,158 @@ protected void abortFileGroup(Integer group) { aborted.add(group); } } + + private static class BlockingHashCodeCommitService + extends BaseCommitService { + + BlockingHashCodeCommitService(Table table, int rewritesPerCommit, int timeoutInMS) { + super(table, rewritesPerCommit, timeoutInMS); + } + + @Override + protected void commitOrClean(Set batch) {} + + @Override + protected void abortFileGroup(BlockingHashCodeGroup group) {} + } + + private static class BlockingHashCodeGroup { + private final CountDownLatch hashCodeStarted = new CountDownLatch(1); + private final CountDownLatch releaseHashCode = new CountDownLatch(1); + + @Override + public int hashCode() { + hashCodeStarted.countDown(); + await(releaseHashCode); + return 0; + } + + @Override + public boolean equals(Object other) { + return this == other; + } + } + + private static class ThrowingHashCodeCommitService + extends BaseCommitService { + + ThrowingHashCodeCommitService(Table table, int rewritesPerCommit, int timeoutInMS) { + super(table, rewritesPerCommit, timeoutInMS); + } + + @Override + protected void commitOrClean(Set batch) {} + + @Override + protected void abortFileGroup(ThrowingHashCodeGroup group) {} + } + + private static class ThrowingHashCodeGroup { + @Override + public int hashCode() { + throw new IllegalStateException("Cannot hash file group"); + } + + @Override + public boolean equals(Object other) { + return this == other; + } + } + + private static class TrackingCommitService extends BaseCommitService { + private final CountDownLatch firstCommitStarted = new CountDownLatch(1); + private final CountDownLatch secondCommitStarted = new CountDownLatch(1); + private final CountDownLatch releaseCommits = new CountDownLatch(1); + private final AtomicInteger activeCommits = new AtomicInteger(0); + private final AtomicInteger maxConcurrentCommits = new AtomicInteger(0); + + TrackingCommitService(Table table) { + this(table, 1); + } + + TrackingCommitService(Table table, int rewritesPerCommit) { + super(table, rewritesPerCommit, 10000); + } + + @Override + protected void commitOrClean(Set batch) { + int active = activeCommits.incrementAndGet(); + maxConcurrentCommits.accumulateAndGet(active, Math::max); + if (batch.contains(0)) { + firstCommitStarted.countDown(); + } else { + secondCommitStarted.countDown(); + } + + try { + await(releaseCommits); + } finally { + activeCommits.decrementAndGet(); + } + } + + @Override + protected void abortFileGroup(Integer group) {} + } + + private static class HandoffCommitService extends BaseCommitService { + private final CountDownLatch firstCommitStarted = new CountDownLatch(1); + private final CountDownLatch releaseFirstCommit = new CountDownLatch(1); + private final CountDownLatch ownerObservedEmptyQueue = new CountDownLatch(1); + private final CountDownLatch releaseOwner = new CountDownLatch(1); + private final CountDownLatch secondCommitStarted = new CountDownLatch(1); + private final AtomicBoolean pauseOwner = new AtomicBoolean(false); + private final AtomicBoolean ownerPaused = new AtomicBoolean(false); + private volatile Thread ownerThread = null; + + HandoffCommitService(Table table) { + super(table, 1, 10000); + } + + @Override + protected void commitOrClean(Set batch) { + if (batch.contains(0)) { + ownerThread = Thread.currentThread(); + firstCommitStarted.countDown(); + await(releaseFirstCommit); + pauseOwner.set(true); + } else { + secondCommitStarted.countDown(); + } + } + + @Override + protected void abortFileGroup(Integer group) {} + + @Override + boolean canCreateCommitGroup() { + boolean canCreate = super.canCreateCommitGroup(); + if (!canCreate + && pauseOwner.get() + && Thread.currentThread().equals(ownerThread) + && ownerPaused.compareAndSet(false, true)) { + ownerObservedEmptyQueue.countDown(); + await(releaseOwner); + } + + return canCreate; + } + } + + private static void await(CountDownLatch latch) { + try { + latch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + + private static void closeQuietly(BaseCommitService commitService) { + try { + commitService.close(); + } catch (RuntimeException ignored) { + // The original test failure should not be replaced by a cleanup failure. + } + } } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java index f13de20568c2..5c844a25046a 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java @@ -1429,16 +1429,13 @@ public void testParallelPartialProgressWithMaxCommitsLargerThanTotalGroupCount() // Since we can have at most one commit per file group and there are only 10 file // groups, actual number of commits is 10 .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_COMMITS, "20") - // Setting max-failed-commits to 1 to tolerate random commit failure - .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "1"); - rewrite.execute(); + .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "0"); + Result result = rewrite.execute(); List postRewriteData = currentData(); assertEquals("We shouldn't have changed the data", originalData, postRewriteData); - assertThat(table.snapshots()) - .as("Table did not have the expected number of snapshots") - // To tolerate 1 random commit failure - .hasSizeGreaterThanOrEqualTo(10); + assertThat(result.rewriteResults()).as("Should have 10 file groups").hasSize(10); + shouldHaveSnapshots(table, 11); shouldHaveNoOrphans(table); shouldHaveACleanCache(table); } From 454a336bf0e6198810be7a6b93e832035010147a Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar <186172142+smaheshwar-pltr@users.noreply.github.com> Date: Thu, 27 Aug 2026 00:24:55 +0100 Subject: [PATCH 2/7] Spark: Extend partial-progress commit test fix to all versions Apply the zero-failure assertions from Spark 4.1 to the identical 3.5, 4.0, and 4.2 tests. Also remove an unused TrackingCommitService constructor from the core test. --- .../org/apache/iceberg/actions/TestCommitService.java | 6 +----- .../spark/actions/TestRewriteDataFilesAction.java | 11 ++++------- .../spark/actions/TestRewriteDataFilesAction.java | 11 ++++------- .../spark/actions/TestRewriteDataFilesAction.java | 11 ++++------- 4 files changed, 13 insertions(+), 26 deletions(-) diff --git a/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java b/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java index 5a6918c72132..991a520d69dd 100644 --- a/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java +++ b/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java @@ -360,11 +360,7 @@ private static class TrackingCommitService extends BaseCommitService { private final AtomicInteger maxConcurrentCommits = new AtomicInteger(0); TrackingCommitService(Table table) { - this(table, 1); - } - - TrackingCommitService(Table table, int rewritesPerCommit) { - super(table, rewritesPerCommit, 10000); + super(table, 1, 10000); } @Override diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java index 3c0b58d30592..4c05ffc6fea4 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java @@ -1428,16 +1428,13 @@ public void testParallelPartialProgressWithMaxCommitsLargerThanTotalGroupCount() // Since we can have at most one commit per file group and there are only 10 file // groups, actual number of commits is 10 .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_COMMITS, "20") - // Setting max-failed-commits to 1 to tolerate random commit failure - .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "1"); - rewrite.execute(); + .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "0"); + Result result = rewrite.execute(); List postRewriteData = currentData(); assertEquals("We shouldn't have changed the data", originalData, postRewriteData); - assertThat(table.snapshots()) - .as("Table did not have the expected number of snapshots") - // To tolerate 1 random commit failure - .hasSizeGreaterThanOrEqualTo(10); + assertThat(result.rewriteResults()).as("Should have 10 file groups").hasSize(10); + shouldHaveSnapshots(table, 11); shouldHaveNoOrphans(table); shouldHaveACleanCache(table); } diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java index 3096f1b55196..51d8b78b313d 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java @@ -1429,16 +1429,13 @@ public void testParallelPartialProgressWithMaxCommitsLargerThanTotalGroupCount() // Since we can have at most one commit per file group and there are only 10 file // groups, actual number of commits is 10 .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_COMMITS, "20") - // Setting max-failed-commits to 1 to tolerate random commit failure - .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "1"); - rewrite.execute(); + .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "0"); + Result result = rewrite.execute(); List postRewriteData = currentData(); assertEquals("We shouldn't have changed the data", originalData, postRewriteData); - assertThat(table.snapshots()) - .as("Table did not have the expected number of snapshots") - // To tolerate 1 random commit failure - .hasSizeGreaterThanOrEqualTo(10); + assertThat(result.rewriteResults()).as("Should have 10 file groups").hasSize(10); + shouldHaveSnapshots(table, 11); shouldHaveNoOrphans(table); shouldHaveACleanCache(table); } diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java index 5c1281ea129d..63a6fe364b1a 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java @@ -1429,16 +1429,13 @@ public void testParallelPartialProgressWithMaxCommitsLargerThanTotalGroupCount() // Since we can have at most one commit per file group and there are only 10 file // groups, actual number of commits is 10 .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_COMMITS, "20") - // Setting max-failed-commits to 1 to tolerate random commit failure - .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "1"); - rewrite.execute(); + .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "0"); + Result result = rewrite.execute(); List postRewriteData = currentData(); assertEquals("We shouldn't have changed the data", originalData, postRewriteData); - assertThat(table.snapshots()) - .as("Table did not have the expected number of snapshots") - // To tolerate 1 random commit failure - .hasSizeGreaterThanOrEqualTo(10); + assertThat(result.rewriteResults()).as("Should have 10 file groups").hasSize(10); + shouldHaveSnapshots(table, 11); shouldHaveNoOrphans(table); shouldHaveACleanCache(table); } From 7d235337617f9e8b6e54ef625548f551708b4b05 Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar <186172142+smaheshwar-pltr@users.noreply.github.com> Date: Wed, 9 Sep 2026 01:11:37 +0100 Subject: [PATCH 3/7] Core: Keep commit test logging independent of hashing Give the blocking file-group fixture a constant string representation so debug logging does not release the extraction latch before enqueueing. Generated-by: OpenAI Codex --- .../java/org/apache/iceberg/actions/TestCommitService.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java b/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java index 991a520d69dd..6bc27c57cc41 100644 --- a/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java +++ b/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java @@ -324,6 +324,11 @@ public int hashCode() { public boolean equals(Object other) { return this == other; } + + @Override + public String toString() { + return "BlockingHashCodeGroup"; + } } private static class ThrowingHashCodeCommitService From 4e995b191a55e452fb4c13c71ccd4e7f1d5d0f17 Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar Date: Mon, 19 Oct 2026 22:10:49 +0100 Subject: [PATCH 4/7] Core: Write rewrite manifests concurrently but commit one at a time Concurrent partial-progress commits to the same table conflict, retry with backoff, and can drop rewritten groups when retries run out. Commit under a lock in the rewrite commit managers, applying each rewrite first so manifest writing stays concurrent and only the metadata update is serialized. This replaces the BaseCommitService lock, which serialized manifest writing as well. Generated-by: Claude Code Co-Authored-By: Claude Opus 5.5 --- .../iceberg/actions/BaseCommitService.java | 74 +--- .../RewriteDataFilesCommitManager.java | 22 +- .../RewritePositionDeletesCommitManager.java | 22 +- .../iceberg/actions/TestCommitService.java | 402 ++---------------- .../TestRewriteDataFilesCommitManager.java | 159 +++++++ 5 files changed, 263 insertions(+), 416 deletions(-) create mode 100644 core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java diff --git a/core/src/main/java/org/apache/iceberg/actions/BaseCommitService.java b/core/src/main/java/org/apache/iceberg/actions/BaseCommitService.java index 8c846edcb929..4c79781efb07 100644 --- a/core/src/main/java/org/apache/iceberg/actions/BaseCommitService.java +++ b/core/src/main/java/org/apache/iceberg/actions/BaseCommitService.java @@ -27,7 +27,6 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.locks.ReentrantLock; import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; @@ -38,10 +37,9 @@ import org.slf4j.LoggerFactory; /** - * A service that commits file groups as rewrites complete. Commits are serialized and may run in a - * thread offering a file group or in the service executor during close. Other offering threads - * continue without waiting for an active commit. Failed commits permit partial progress. No groups - * may be offered after the service is closed. + * An async service which allows for committing multiple file groups as their rewrites complete. The + * service also allows for partial-progress since commits can fail. Once the service has been closed + * no new file groups should not be offered. * *

Specific implementations provide implementations for {@link #commitOrClean(Set)} and {@link * #abortFileGroup(Object)} @@ -61,14 +59,13 @@ abstract class BaseCommitService implements Closeable { private final int rewritesPerCommit; private final AtomicBoolean running = new AtomicBoolean(false); private final long timeoutInMS; - private final ReentrantLock commitLock = new ReentrantLock(); private int succeededCommits = 0; /** * Constructs a {@link BaseCommitService} * * @param table table to perform commit on - * @param rewritesPerCommit positive number of file groups to include in a commit + * @param rewritesPerCommit number of file groups to include in a commit */ BaseCommitService(Table table, int rewritesPerCommit) { this(table, rewritesPerCommit, TIMEOUT_IN_MS_DEFAULT); @@ -78,15 +75,11 @@ abstract class BaseCommitService implements Closeable { * Constructs a {@link BaseCommitService} * * @param table table to perform commit on - * @param rewritesPerCommit positive number of file groups to include in a commit + * @param rewritesPerCommit number of file groups to include in a commit * @param timeoutInMS The timeout to wait for commits to complete after all rewrite jobs have been * completed */ BaseCommitService(Table table, int rewritesPerCommit, long timeoutInMS) { - Preconditions.checkArgument( - rewritesPerCommit > 0, - "Invalid number of file groups per commit: %s (must be positive)", - rewritesPerCommit); this.table = table; LOG.info( "Creating commit service for table {} with {} groups per commit", table, rewritesPerCommit); @@ -216,50 +209,29 @@ public void close() { } private void commitReadyCommitGroups() { - while (canCreateCommitGroup()) { - if (!commitLock.tryLock()) { - return; - } - - try { - while (canCreateCommitGroup()) { - Set batch = null; - String inProgressCommitToken = null; - try { - synchronized (completedRewrites) { - if (canCreateCommitGroup()) { - batch = Sets.newHashSetWithExpectedSize(rewritesPerCommit); - // The committer may exit as soon as both queues are empty. - inProgressCommitToken = UUID.randomUUID().toString(); - inProgressCommits.add(inProgressCommitToken); - for (int i = 0; i < rewritesPerCommit && !completedRewrites.isEmpty(); i++) { - batch.add(completedRewrites.poll()); - } - } - } - - if (batch != null) { - commitBatch(batch); - } - } finally { - if (inProgressCommitToken != null) { - inProgressCommits.remove(inProgressCommitToken); - } + Set batch = null; + if (canCreateCommitGroup()) { + synchronized (completedRewrites) { + if (canCreateCommitGroup()) { + batch = Sets.newHashSetWithExpectedSize(rewritesPerCommit); + for (int i = 0; i < rewritesPerCommit && !completedRewrites.isEmpty(); i++) { + batch.add(completedRewrites.poll()); } } - } finally { - commitLock.unlock(); } } - } - private void commitBatch(Set batch) { - try { - commitOrClean(batch); - committedRewrites.addAll(batch); - succeededCommits++; - } catch (Exception e) { - LOG.error("Failure during rewrite commit process, partial progress enabled. Ignoring", e); + if (batch != null) { + String inProgressCommitToken = UUID.randomUUID().toString(); + inProgressCommits.add(inProgressCommitToken); + try { + commitOrClean(batch); + committedRewrites.addAll(batch); + succeededCommits++; + } catch (Exception e) { + LOG.error("Failure during rewrite commit process, partial progress enabled. Ignoring", e); + } + inProgressCommits.remove(inProgressCommitToken); } } diff --git a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java index 1d2d637d70e1..8090d6008133 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java +++ b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java @@ -41,6 +41,7 @@ public class RewriteDataFilesCommitManager { private final boolean useStartingSequenceNumber; private final Map 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 fileGroups) { rewrite.toBranch(branch); } - rewrite.commit(); + // Concurrent commits to the same table conflict and retry, so commit one at a time. Manifests + // are written first so that work stays concurrent: commit() reuses them, or applies again and + // cleans up if apply() failed. + try { + rewrite.apply(); + } catch (RuntimeException e) { + LOG.warn("Failed to apply rewrite before committing, retrying in commit", e); + } + + synchronized (commitLock) { + rewrite.commit(); + } } /** @@ -150,11 +162,11 @@ public void commitOrClean(Set rewriteGroups) { } /** - * A service that commits file groups as rewrites complete. Commits are serialized and may run in - * the thread offering a file group. Failed commits permit partial progress. No groups may be - * offered after the service is closed. + * An async service which allows for committing multiple file groups as their rewrites complete. + * The service also allows for partial-progress since commits can fail. Once the service has been + * closed no new file groups should not be offered. * - * @param rewritesPerCommit positive number of file groups to include in a commit + * @param rewritesPerCommit number of file groups to include in a commit * @return the service for handling commits */ public CommitService service(int rewritesPerCommit) { diff --git a/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java index 17277f2b8186..6a42814bda58 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java +++ b/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 snapshotProperties; + private final Object commitLock = new Object(); public RewritePositionDeletesCommitManager(Table table) { this(table, ImmutableMap.of()); @@ -76,7 +77,18 @@ public void commit(Set fileGroups) { snapshotProperties.forEach(rewriteFiles::set); - rewriteFiles.commit(); + // Concurrent commits to the same table conflict and retry, so commit one at a time. Manifests + // are written first so that work stays concurrent: commit() reuses them, or applies again and + // cleans up if apply() failed. + try { + rewriteFiles.apply(); + } catch (RuntimeException e) { + LOG.warn("Failed to apply rewrite before committing, retrying in commit", e); + } + + synchronized (commitLock) { + rewriteFiles.commit(); + } } /** @@ -115,11 +127,11 @@ public void commitOrClean(Set rewriteGroups) { } /** - * A service that commits file groups as rewrites complete. Commits are serialized and may run in - * the thread offering a file group. Failed commits permit partial progress. No groups may be - * offered after the service is closed. + * An async service which allows for committing multiple file groups as their rewrites complete. + * The service also allows for partial-progress since commits can fail. Once the service has been + * closed no new file groups should not be offered. * - * @param rewritesPerCommit positive number of file groups to include in a commit + * @param rewritesPerCommit number of file groups to include in a commit * @return the service for handling commits */ public CommitService service(int rewritesPerCommit) { diff --git a/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java b/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java index 6bc27c57cc41..a40e7f617e7b 100644 --- a/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java +++ b/core/src/test/java/org/apache/iceberg/actions/TestCommitService.java @@ -26,13 +26,9 @@ import java.util.Arrays; import java.util.List; import java.util.Set; -import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; import org.apache.iceberg.ParameterizedTestExtension; import org.apache.iceberg.Parameters; @@ -46,23 +42,12 @@ @ExtendWith(ParameterizedTestExtension.class) public class TestCommitService extends TestBase { - private static final int BATCH_EXTRACTION_TIMEOUT_MS = 1000; @Parameters(name = "formatVersion = {0}") protected static List parameters() { return Arrays.asList(1); } - @TestTemplate - void rejectsNonPositiveCommitGroupSize() { - assertThatThrownBy(() -> new CustomCommitService(table, 0, 10000)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Invalid number of file groups per commit: 0 (must be positive)"); - assertThatThrownBy(() -> new CustomCommitService(table, -1, 10000)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("Invalid number of file groups per commit: -1 (must be positive)"); - } - @TestTemplate public void testCommittedResultsCorrectly() { CustomCommitService commitService = new CustomCommitService(table, 5, 10000); @@ -79,374 +64,81 @@ public void testCommittedResultsCorrectly() { } @TestTemplate - public void testAbortFileGroupsAfterTimeout() throws Exception { - BlockingCommitService commitService = new BlockingCommitService(table, 5, 100); + public void testAbortFileGroupsAfterTimeout() { + CustomCommitService commitService = new CustomCommitService(table, 5, 200); commitService.start(); + // Add file groups [0-3] for commit. + // There are less than the rewritesPerCommit, and thus will not trigger a commit action. Those + // file groups will be added to the completedRewrites queue. + // Now the queue has 4 file groups that need to commit. + for (int i = 0; i < 4; i++) { + commitService.offer(i); + } + + // Add file groups [4-7] for commit + // These are gated to not be able to commit, so all those 4 file groups will be added to the + // queue as well. + // Now the queue has 8 file groups that need to commit. CustomCommitService spyCommitService = spy(commitService); doReturn(false).when(spyCommitService).canCreateCommitGroup(); - for (int i = 0; i < 7; i++) { + for (int i = 4; i < 8; i++) { spyCommitService.offer(i); } - ExecutorService offerExecutor = Executors.newSingleThreadExecutor(); - ExecutorService closeExecutor = Executors.newSingleThreadExecutor(); - try { - Future commitResult = offerExecutor.submit(() -> commitService.offer(7)); - assertThat(commitService.commitStarted.await(5, TimeUnit.SECONDS)).isTrue(); - - Future closeResult = - closeExecutor.submit( - () -> { - try { - commitService.close(); - return null; - } catch (Throwable e) { - return e; - } - }); + // close commitService. + // This allows committerService thread to start to commit the remaining file groups [0-7] in the + // completedRewrites queue. And also the main thread waits for the committerService thread to + // finish within a timeout. - assertThat(closeResult.get(5, TimeUnit.SECONDS)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Timeout occurred when waiting for commits") - .hasMessageNotContaining("{}") - .hasMessageMatching( - ".*\\d+ file groups committed\\. \\d+ file groups remain uncommitted\\..*"); - assertThat(commitService.aborted).containsExactlyInAnyOrder(5, 6, 7); - commitService.releaseCommit.countDown(); - commitResult.get(5, TimeUnit.SECONDS); - } finally { - commitService.releaseCommit.countDown(); - offerExecutor.shutdownNow(); - closeExecutor.shutdownNow(); - closeQuietly(commitService); - } - - assertThat(offerExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); - assertThat(closeExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + // The committerService thread commits file groups [0-4]. These will wait a fixed duration to + // simulate timeout on the main thread, which then tries to abort file groups [5-7]. + // This tests the race conditions, as the committerService is also trying to commit groups + // [5-7]. + assertThatThrownBy(commitService::close) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Timeout occurred when waiting for commits") + .hasMessageNotContaining("{}") + .hasMessageMatching( + ".*\\d+ file groups committed\\. \\d+ file groups remain uncommitted\\..*"); + // Wait for the commitService to finish. Committed all file groups or aborted remaining file + // groups. Awaitility.await() .atMost(5, TimeUnit.SECONDS) .pollInSameThread() .untilAsserted(() -> assertThat(commitService.completedRewritesAllCommitted()).isTrue()); - assertThat(commitService.results()).containsExactlyInAnyOrder(0, 1, 2, 3, 4); - } - - @TestTemplate - void closeReportsTimeoutWhileBatchExtractionIsInProgress() throws Exception { - BlockingHashCodeCommitService commitService = - new BlockingHashCodeCommitService(table, 1, BATCH_EXTRACTION_TIMEOUT_MS); - BlockingHashCodeGroup group = new BlockingHashCodeGroup(); - commitService.start(); - - ExecutorService offerExecutor = Executors.newSingleThreadExecutor(); - ExecutorService closeExecutor = Executors.newSingleThreadExecutor(); - try { - Future commitResult = offerExecutor.submit(() -> commitService.offer(group)); - assertThat(group.hashCodeStarted.await(5, TimeUnit.SECONDS)).isTrue(); - - Future closeResult = - closeExecutor.submit( - () -> { - try { - commitService.close(); - return null; - } catch (Throwable e) { - return e; - } - }); - - assertThat(closeResult.get(5, TimeUnit.SECONDS)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("Timeout occurred when waiting for commits"); - group.releaseHashCode.countDown(); - commitResult.get(5, TimeUnit.SECONDS); - } finally { - group.releaseHashCode.countDown(); - offerExecutor.shutdownNow(); - closeExecutor.shutdownNow(); - closeQuietly(commitService); - } - - assertThat(offerExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); - assertThat(closeExecutor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); - assertThat(commitService.results()).containsExactly(group); - } - - @TestTemplate - void failedBatchExtractionDoesNotStrandClose() { - ThrowingHashCodeCommitService commitService = - new ThrowingHashCodeCommitService(table, 1, BATCH_EXTRACTION_TIMEOUT_MS); - commitService.start(); - - boolean closed = false; - try { - assertThatThrownBy(() -> commitService.offer(new ThrowingHashCodeGroup())) - .isInstanceOf(IllegalStateException.class) - .hasMessage("Cannot hash file group"); - - commitService.close(); - closed = true; - } finally { - if (!closed) { - closeQuietly(commitService); - } + if (commitService.aborted.isEmpty()) { + // All file groups are committed + assertThat(commitService.results()).containsExactly(0, 1, 2, 3, 4, 5, 6, 7); + } else { + // File groups [5-7] are aborted + assertThat(commitService.results()).doesNotContainAnyElementsOf(commitService.aborted); + assertThat(commitService.results()).containsExactly(0, 1, 2, 3, 4); + assertThat(commitService.aborted).containsExactly(5, 6, 7); } } - @TestTemplate - void serializesCommitsWithoutBlockingOtherOffers() throws Exception { - TrackingCommitService commitService = new TrackingCommitService(table); - commitService.start(); - - ExecutorService executorService = Executors.newFixedThreadPool(2); - boolean closed = false; - try { - Future firstCommit = executorService.submit(() -> commitService.offer(0)); - assertThat(commitService.firstCommitStarted.await(5, TimeUnit.SECONDS)).isTrue(); - - Future secondCommit = executorService.submit(() -> commitService.offer(1)); - secondCommit.get(5, TimeUnit.SECONDS); - assertThat(commitService.secondCommitStarted.getCount()).isEqualTo(1); - assertThat(commitService.maxConcurrentCommits.get()).isEqualTo(1); - - commitService.releaseCommits.countDown(); - firstCommit.get(5, TimeUnit.SECONDS); - assertThat(commitService.secondCommitStarted.await(5, TimeUnit.SECONDS)).isTrue(); - commitService.close(); - closed = true; - } finally { - commitService.releaseCommits.countDown(); - executorService.shutdownNow(); - if (!closed) { - closeQuietly(commitService); - } - } - - assertThat(executorService.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); - assertThat(commitService.results()).containsExactlyInAnyOrder(0, 1); - } - - @TestTemplate - void drainOwnerProcessesBatchOfferedDuringHandoff() throws Exception { - HandoffCommitService commitService = new HandoffCommitService(table); - commitService.start(); - - ExecutorService executorService = Executors.newFixedThreadPool(2); - boolean closed = false; - try { - Future firstCommit = executorService.submit(() -> commitService.offer(0)); - assertThat(commitService.firstCommitStarted.await(5, TimeUnit.SECONDS)).isTrue(); - commitService.releaseFirstCommit.countDown(); - assertThat(commitService.ownerObservedEmptyQueue.await(5, TimeUnit.SECONDS)).isTrue(); - - Future handoffOffer = executorService.submit(() -> commitService.offer(1)); - handoffOffer.get(5, TimeUnit.SECONDS); - assertThat(commitService.secondCommitStarted.getCount()).isEqualTo(1); - - commitService.releaseOwner.countDown(); - assertThat(commitService.secondCommitStarted.await(5, TimeUnit.SECONDS)).isTrue(); - firstCommit.get(5, TimeUnit.SECONDS); - commitService.close(); - closed = true; - } finally { - commitService.releaseFirstCommit.countDown(); - commitService.releaseOwner.countDown(); - executorService.shutdownNow(); - if (!closed) { - closeQuietly(commitService); - } - } - - assertThat(executorService.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); - assertThat(commitService.results()).containsExactlyInAnyOrder(0, 1); - } - private static class CustomCommitService extends BaseCommitService { - CustomCommitService(Table table, int rewritesPerCommit, int timeoutInMS) { - super(table, rewritesPerCommit, timeoutInMS); - } - - @Override - protected void commitOrClean(Set batch) {} - - @Override - protected void abortFileGroup(Integer group) {} - } - - private static class BlockingCommitService extends CustomCommitService { private final Set aborted = Sets.newConcurrentHashSet(); - private final CountDownLatch commitStarted = new CountDownLatch(1); - private final CountDownLatch releaseCommit = new CountDownLatch(1); - - BlockingCommitService(Table table, int rewritesPerCommit, int timeoutInMS) { - super(table, rewritesPerCommit, timeoutInMS); - } - - @Override - protected void commitOrClean(Set batch) { - commitStarted.countDown(); - await(releaseCommit); - } - - @Override - protected void abortFileGroup(Integer group) { - aborted.add(group); - } - } - - private static class BlockingHashCodeCommitService - extends BaseCommitService { - - BlockingHashCodeCommitService(Table table, int rewritesPerCommit, int timeoutInMS) { - super(table, rewritesPerCommit, timeoutInMS); - } - - @Override - protected void commitOrClean(Set batch) {} - - @Override - protected void abortFileGroup(BlockingHashCodeGroup group) {} - } - - private static class BlockingHashCodeGroup { - private final CountDownLatch hashCodeStarted = new CountDownLatch(1); - private final CountDownLatch releaseHashCode = new CountDownLatch(1); - - @Override - public int hashCode() { - hashCodeStarted.countDown(); - await(releaseHashCode); - return 0; - } - - @Override - public boolean equals(Object other) { - return this == other; - } - - @Override - public String toString() { - return "BlockingHashCodeGroup"; - } - } - - private static class ThrowingHashCodeCommitService - extends BaseCommitService { - - ThrowingHashCodeCommitService(Table table, int rewritesPerCommit, int timeoutInMS) { - super(table, rewritesPerCommit, timeoutInMS); - } - - @Override - protected void commitOrClean(Set batch) {} - - @Override - protected void abortFileGroup(ThrowingHashCodeGroup group) {} - } - private static class ThrowingHashCodeGroup { - @Override - public int hashCode() { - throw new IllegalStateException("Cannot hash file group"); - } - - @Override - public boolean equals(Object other) { - return this == other; - } - } - - private static class TrackingCommitService extends BaseCommitService { - private final CountDownLatch firstCommitStarted = new CountDownLatch(1); - private final CountDownLatch secondCommitStarted = new CountDownLatch(1); - private final CountDownLatch releaseCommits = new CountDownLatch(1); - private final AtomicInteger activeCommits = new AtomicInteger(0); - private final AtomicInteger maxConcurrentCommits = new AtomicInteger(0); - - TrackingCommitService(Table table) { - super(table, 1, 10000); + CustomCommitService(Table table, int rewritesPerCommit, int timeoutInSeconds) { + super(table, rewritesPerCommit, timeoutInSeconds); } @Override protected void commitOrClean(Set batch) { - int active = activeCommits.incrementAndGet(); - maxConcurrentCommits.accumulateAndGet(active, Math::max); - if (batch.contains(0)) { - firstCommitStarted.countDown(); - } else { - secondCommitStarted.countDown(); - } - try { - await(releaseCommits); - } finally { - activeCommits.decrementAndGet(); + // Slightly longer than timeout + Thread.sleep(210); + } catch (InterruptedException e) { + throw new RuntimeException(e); } } @Override - protected void abortFileGroup(Integer group) {} - } - - private static class HandoffCommitService extends BaseCommitService { - private final CountDownLatch firstCommitStarted = new CountDownLatch(1); - private final CountDownLatch releaseFirstCommit = new CountDownLatch(1); - private final CountDownLatch ownerObservedEmptyQueue = new CountDownLatch(1); - private final CountDownLatch releaseOwner = new CountDownLatch(1); - private final CountDownLatch secondCommitStarted = new CountDownLatch(1); - private final AtomicBoolean pauseOwner = new AtomicBoolean(false); - private final AtomicBoolean ownerPaused = new AtomicBoolean(false); - private volatile Thread ownerThread = null; - - HandoffCommitService(Table table) { - super(table, 1, 10000); - } - - @Override - protected void commitOrClean(Set batch) { - if (batch.contains(0)) { - ownerThread = Thread.currentThread(); - firstCommitStarted.countDown(); - await(releaseFirstCommit); - pauseOwner.set(true); - } else { - secondCommitStarted.countDown(); - } - } - - @Override - protected void abortFileGroup(Integer group) {} - - @Override - boolean canCreateCommitGroup() { - boolean canCreate = super.canCreateCommitGroup(); - if (!canCreate - && pauseOwner.get() - && Thread.currentThread().equals(ownerThread) - && ownerPaused.compareAndSet(false, true)) { - ownerObservedEmptyQueue.countDown(); - await(releaseOwner); - } - - return canCreate; - } - } - - private static void await(CountDownLatch latch) { - try { - latch.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new RuntimeException(e); - } - } - - private static void closeQuietly(BaseCommitService commitService) { - try { - commitService.close(); - } catch (RuntimeException ignored) { - // The original test failure should not be replaced by a cleanup failure. + protected void abortFileGroup(Integer group) { + aborted.add(group); } } } diff --git a/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java b/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java new file mode 100644 index 000000000000..36d7fc0e40eb --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java @@ -0,0 +1,159 @@ +/* + * 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.ParameterizedTestExtension; +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.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() + .set(TableProperties.COMMIT_NUM_RETRIES, "0") + // test tables otherwise assign sequential snapshot IDs, which concurrent producers reuse + .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 groups = Lists.newArrayList(); + try (CloseableIterable 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> 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); + } + + @TestTemplate + void commitRemovesUnusedManifestLists() throws IOException { + table.newAppend().appendFile(FILE_A).commit(); + RewriteDataFilesCommitManager commitManager = new RewriteDataFilesCommitManager(table); + + commitManager.commitFileGroups(Set.of(group(planFiles(), FILE_C))); + + assertThat(table.snapshots()).hasSize(2); + assertThat(metadataFiles("snap-")).hasSize(2); + } + + @TestTemplate + void failedCommitRemovesWrittenManifests() throws IOException { + table.newAppend().appendFile(FILE_A).commit(); + table.newAppend().appendFile(FILE_B).commit(); + RewriteDataFilesCommitManager commitManager = new RewriteDataFilesCommitManager(table); + RewriteFileGroup group = group(planFiles(), FILE_C); + table.newDelete().deleteFile(FILE_B).commit(); + List metadataFilesBeforeCommit = metadataFiles(""); + + assertThatThrownBy(() -> commitManager.commitFileGroups(Set.of(group))) + .isInstanceOf(ValidationException.class) + .hasMessageStartingWith("Missing required files to delete"); + + assertThat(metadataFiles("")).containsExactlyInAnyOrderElementsOf(metadataFilesBeforeCommit); + } + + private List planFiles() throws IOException { + try (CloseableIterable tasks = table.newScan().planFiles()) { + return Lists.newArrayList(tasks); + } + } + + private RewriteFileGroup group(List 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 metadataFiles(String prefix) { + return Arrays.stream(metadataDir.listFiles()) + .map(File::getName) + .filter(name -> name.startsWith(prefix) && name.endsWith(".avro")) + .toList(); + } +} From 7e4df94b57b2adead688ead4ac3d43e501fc2bdb Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar Date: Mon, 19 Oct 2026 22:29:55 +0100 Subject: [PATCH 5/7] Core: Tighten rewrite commit lock logging and tests Log a failed early apply at debug level since commit() reports any failure that persists. Check Parquet manifests on v4 tables and that concurrent commits leave no unreferenced metadata files. Generated-by: Claude Code Co-Authored-By: Claude Opus 5.5 --- .../actions/RewriteDataFilesCommitManager.java | 8 ++++---- .../RewritePositionDeletesCommitManager.java | 8 ++++---- .../TestRewriteDataFilesCommitManager.java | 16 ++++++++++++++-- 3 files changed, 22 insertions(+), 10 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java index 8090d6008133..eea6787bf58a 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java +++ b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java @@ -110,13 +110,13 @@ public void commitFileGroups(Set fileGroups) { rewrite.toBranch(branch); } - // Concurrent commits to the same table conflict and retry, so commit one at a time. Manifests - // are written first so that work stays concurrent: commit() reuses them, or applies again and - // cleans up if apply() failed. + // Concurrent commits through this manager would conflict with each other, so run them one at a + // time. Manifests are written first so that work stays concurrent: commit() reuses them, or + // applies again and cleans up if apply() failed. try { rewrite.apply(); } catch (RuntimeException e) { - LOG.warn("Failed to apply rewrite before committing, retrying in commit", e); + LOG.debug("Failed to apply rewrite before committing, applying again in commit", e); } synchronized (commitLock) { diff --git a/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java index 6a42814bda58..94774ee9cad4 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java +++ b/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java @@ -77,13 +77,13 @@ public void commit(Set fileGroups) { snapshotProperties.forEach(rewriteFiles::set); - // Concurrent commits to the same table conflict and retry, so commit one at a time. Manifests - // are written first so that work stays concurrent: commit() reuses them, or applies again and - // cleans up if apply() failed. + // Concurrent commits through this manager would conflict with each other, so run them one at a + // time. Manifests are written first so that work stays concurrent: commit() reuses them, or + // applies again and cleans up if apply() failed. try { rewriteFiles.apply(); } catch (RuntimeException e) { - LOG.warn("Failed to apply rewrite before committing, retrying in commit", e); + LOG.debug("Failed to apply rewrite before committing, applying again in commit", e); } synchronized (commitLock) { diff --git a/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java b/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java index 36d7fc0e40eb..0b1657955ad0 100644 --- a/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java +++ b/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java @@ -35,12 +35,15 @@ 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; @@ -53,7 +56,7 @@ void concurrentCommitsDoNotConflict() throws Exception { table .updateProperties() .set(TableProperties.COMMIT_NUM_RETRIES, "0") - // test tables otherwise assign sequential snapshot IDs, which concurrent producers reuse + // test tables otherwise assign sequential snapshot IDs, and a reused ID drops a commit .set("random-snapshot-ids", "true") .commit(); AppendFiles append = table.newFastAppend(); @@ -94,6 +97,15 @@ void concurrentCommitsDoNotConflict() throws Exception { } assertThat(table.snapshots()).hasSize(numGroups + 1); + Set 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()); + } + } + + assertThat(metadataFiles("")).containsExactlyInAnyOrderElementsOf(referencedFiles); } @TestTemplate @@ -153,7 +165,7 @@ private static DataFile copy(DataFile file, String path) { private List metadataFiles(String prefix) { return Arrays.stream(metadataDir.listFiles()) .map(File::getName) - .filter(name -> name.startsWith(prefix) && name.endsWith(".avro")) + .filter(name -> name.startsWith(prefix)) .toList(); } } From 668eadacf33f072cf08e759fa0c9c886efc16377 Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar Date: Tue, 20 Oct 2026 13:57:43 +0100 Subject: [PATCH 6/7] Core: Rebuild added delete summary when retrying a failed apply If writing new delete manifests failed, applying again counted the added delete files twice in the snapshot summary. The early apply in the rewrite commit managers makes this retry reachable. Generated-by: Claude Code Co-Authored-By: Claude Opus 5.5 --- .../iceberg/MergingSnapshotProducer.java | 8 ++-- .../java/org/apache/iceberg/TestRowDelta.java | 43 +++++++++++++++++++ 2 files changed, 47 insertions(+), 4 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java index 70617cbc427a..e6bcc233513c 100644 --- a/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java @@ -1180,13 +1180,13 @@ private List newDeleteFilesAsManifests() { // 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 mergedDVs = mergeDVs(); Map> newDeleteFilesBySpec = Streams.stream(Iterables.concat(mergedDVs, DeleteFileSet.of(v2Deletes))) diff --git a/core/src/test/java/org/apache/iceberg/TestRowDelta.java b/core/src/test/java/org/apache/iceberg/TestRowDelta.java index bacd4b9546cf..934be19f04a2 100644 --- a/core/src/test/java/org/apache/iceberg/TestRowDelta.java +++ b/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); From 72b00c00e53ce6cbbf7ad910463d57f44cfb73ec Mon Sep 17 00:00:00 2001 From: Sreesh Maheshwar Date: Tue, 20 Oct 2026 13:57:43 +0100 Subject: [PATCH 7/7] Core: Test position-delete commit serialization Add commit manager tests for position deletes, describe what commit() reuses from the early apply, and drop a data-file test that the concurrent test already covers. Generated-by: Claude Code Co-Authored-By: Claude Opus 5.5 --- .../RewriteDataFilesCommitManager.java | 6 +- .../RewritePositionDeletesCommitManager.java | 6 +- .../TestRewriteDataFilesCommitManager.java | 49 ++--- ...stRewritePositionDeletesCommitManager.java | 183 ++++++++++++++++++ 4 files changed, 209 insertions(+), 35 deletions(-) create mode 100644 core/src/test/java/org/apache/iceberg/actions/TestRewritePositionDeletesCommitManager.java diff --git a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java index eea6787bf58a..76e697024452 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java +++ b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java @@ -110,9 +110,9 @@ public void commitFileGroups(Set fileGroups) { rewrite.toBranch(branch); } - // Concurrent commits through this manager would conflict with each other, so run them one at a - // time. Manifests are written first so that work stays concurrent: commit() reuses them, or - // applies again and cleans up if apply() failed. + // 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) { diff --git a/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java index 94774ee9cad4..8fb680e2c3e6 100644 --- a/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java +++ b/core/src/main/java/org/apache/iceberg/actions/RewritePositionDeletesCommitManager.java @@ -77,9 +77,9 @@ public void commit(Set fileGroups) { snapshotProperties.forEach(rewriteFiles::set); - // Concurrent commits through this manager would conflict with each other, so run them one at a - // time. Manifests are written first so that work stays concurrent: commit() reuses them, or - // applies again and cleans up if apply() failed. + // 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) { diff --git a/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java b/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java index 0b1657955ad0..878e1749ba2e 100644 --- a/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java +++ b/core/src/test/java/org/apache/iceberg/actions/TestRewriteDataFilesCommitManager.java @@ -55,6 +55,7 @@ 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") @@ -97,42 +98,23 @@ void concurrentCommitsDoNotConflict() throws Exception { } assertThat(table.snapshots()).hasSize(numGroups + 1); - Set 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()); - } - } - - assertThat(metadataFiles("")).containsExactlyInAnyOrderElementsOf(referencedFiles); - } - - @TestTemplate - void commitRemovesUnusedManifestLists() throws IOException { - table.newAppend().appendFile(FILE_A).commit(); - RewriteDataFilesCommitManager commitManager = new RewriteDataFilesCommitManager(table); - - commitManager.commitFileGroups(Set.of(group(planFiles(), FILE_C))); - - assertThat(table.snapshots()).hasSize(2); - assertThat(metadataFiles("snap-")).hasSize(2); + assertThat(metadataFiles()).containsExactlyInAnyOrderElementsOf(referencedMetadataFiles()); } @TestTemplate void failedCommitRemovesWrittenManifests() throws IOException { - table.newAppend().appendFile(FILE_A).commit(); - table.newAppend().appendFile(FILE_B).commit(); + 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 metadataFilesBeforeCommit = metadataFiles(""); + List 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); + assertThat(metadataFiles()).containsExactlyInAnyOrderElementsOf(metadataFilesBeforeCommit); } private List planFiles() throws IOException { @@ -162,10 +144,19 @@ private static DataFile copy(DataFile file, String path) { return DataFiles.builder(SPEC).copy(file).withPath(path).build(); } - private List metadataFiles(String prefix) { - return Arrays.stream(metadataDir.listFiles()) - .map(File::getName) - .filter(name -> name.startsWith(prefix)) - .toList(); + private List metadataFiles() { + return Arrays.stream(metadataDir.listFiles()).map(File::getName).toList(); + } + + private Set referencedMetadataFiles() { + Set 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; } } diff --git a/core/src/test/java/org/apache/iceberg/actions/TestRewritePositionDeletesCommitManager.java b/core/src/test/java/org/apache/iceberg/actions/TestRewritePositionDeletesCommitManager.java new file mode 100644 index 000000000000..2122d1691bb5 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/actions/TestRewritePositionDeletesCommitManager.java @@ -0,0 +1,183 @@ +/* + * 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.DeleteFile; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.MetadataTableType; +import org.apache.iceberg.MetadataTableUtils; +import org.apache.iceberg.ParameterizedTestExtension; +import org.apache.iceberg.Parameters; +import org.apache.iceberg.PositionDeletesScanTask; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +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 TestRewritePositionDeletesCommitManager extends TestBase { + + // v3 and later tables reject position delete files that are not DVs + @Parameters(name = "formatVersion = {0}") + protected static List formatVersions() { + return List.of(2); + } + + @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(); + RowDelta rowDelta = table.newRowDelta().addRows(FILE_A); + for (int i = 0; i < numGroups; i++) { + rowDelta.addDeletes(copy(FILE_A_DELETES, "/path/to/data-a-deletes-" + i + ".parquet")); + } + + rowDelta.commit(); + + List groups = Lists.newArrayList(); + List outputLocations = Lists.newArrayList(); + for (PositionDeletesScanTask task : planFiles()) { + DeleteFile output = copy(task.file(), task.file().location() + ".rewritten"); + groups.add(group(List.of(task), output)); + outputLocations.add(output.location()); + } + + RewritePositionDeletesCommitManager commitManager = + new RewritePositionDeletesCommitManager(table); + CyclicBarrier start = new CyclicBarrier(numGroups); + ExecutorService executorService = Executors.newFixedThreadPool(numGroups); + try { + List> commits = Lists.newArrayList(); + for (RewritePositionDeletesGroup group : groups) { + commits.add( + executorService.submit( + () -> { + start.await(); + commitManager.commit(Set.of(group)); + return null; + })); + } + + for (Future commit : commits) { + commit.get(10, TimeUnit.SECONDS); + } + } finally { + executorService.shutdownNow(); + } + + assertThat(table.snapshots()).hasSize(numGroups + 1); + assertThat(planFiles()) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrderElementsOf(outputLocations); + assertThat(metadataFiles()).containsExactlyInAnyOrderElementsOf(referencedMetadataFiles()); + } + + @TestTemplate + void failedCommitRemovesWrittenManifests() throws IOException { + table.newRowDelta().addRows(FILE_A).addDeletes(FILE_A_DELETES).commit(); + table.newRowDelta().addRows(FILE_B).addDeletes(FILE_B_DELETES).commit(); + RewritePositionDeletesCommitManager commitManager = + new RewritePositionDeletesCommitManager(table); + DeleteFile output = copy(FILE_A_DELETES, "/path/to/data-a-deletes-rewritten.parquet"); + RewritePositionDeletesGroup group = group(planFiles(), output); + table.newRowDelta().removeDeletes(FILE_B_DELETES).commit(); + List metadataFilesBeforeCommit = metadataFiles(); + + // apply() rewrites the FILE_A_DELETES manifest, then fails on the missing FILE_B_DELETES + assertThatThrownBy(() -> commitManager.commit(Set.of(group))) + .isInstanceOf(ValidationException.class) + .hasMessageStartingWith("Missing required files to delete"); + + assertThat(metadataFiles()).containsExactlyInAnyOrderElementsOf(metadataFilesBeforeCommit); + } + + private List planFiles() throws IOException { + Table deletesTable = + MetadataTableUtils.createMetadataTableInstance(table, MetadataTableType.POSITION_DELETES); + try (CloseableIterable tasks = + CloseableIterable.transform( + deletesTable.newBatchScan().planFiles(), PositionDeletesScanTask.class::cast)) { + return Lists.newArrayList(tasks); + } + } + + private RewritePositionDeletesGroup group( + List tasks, DeleteFile outputFile) { + RewritePositionDeletesGroup group = + new RewritePositionDeletesGroup( + ImmutableRewritePositionDeleteFiles.FileGroupInfo.builder() + .globalIndex(0) + .partitionIndex(0) + .partition(outputFile.partition()) + .build(), + tasks, + Long.MAX_VALUE, + Long.MAX_VALUE, + 1); + group.setOutputFiles(Set.of(outputFile)); + return group; + } + + private static DeleteFile copy(DeleteFile file, String path) { + return FileMetadata.deleteFileBuilder(SPEC).copy(file).withPath(path).build(); + } + + private List metadataFiles() { + return Arrays.stream(metadataDir.listFiles()).map(File::getName).toList(); + } + + private Set referencedMetadataFiles() { + Set 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; + } +}