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/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java b/core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java index 727bd6ab84ef..76e697024452 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(); + // 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(); + } } /** 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..8fb680e2c3e6 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(); + // 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(); + } } /** 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); 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..878e1749ba2e --- /dev/null +++ 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 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); + 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 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 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() { + 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; + } +} 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 5bc3293e910f..38fab01ce71d 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 @@ -1431,16 +1431,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 bc8d922f621f..def08b63a391 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 @@ -1433,16 +1433,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.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 4ea6448f38b1..d586043ab698 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 @@ -1433,16 +1433,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 4ea6448f38b1..d586043ab698 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 @@ -1433,16 +1433,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); }