You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by am...@apache.org on 2025/03/24 15:14:41 UTC
(iceberg) branch main updated: Spark 3.4: Propagate snapshot properties / Add max allowed failed commits
This is an automated email from the ASF dual-hosted git repository.
amoghj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 50c8697f93 Spark 3.4: Propagate snapshot properties / Add max allowed failed commits
50c8697f93 is described below
commit 50c8697f937e0fea3c87fe8422bed881792a8129
Author: Eduard Tudenhoefner <et...@gmail.com>
AuthorDate: Mon Mar 24 11:33:40 2025 +0100
Spark 3.4: Propagate snapshot properties / Add max allowed failed commits
this backports #9449 and #9611 to Spark 3.4
---
.../actions/BaseSnapshotUpdateSparkAction.java | 5 ++
.../spark/actions/RewriteDataFilesSparkAction.java | 40 +++++++++++----
.../RewritePositionDeleteFilesSparkAction.java | 2 +-
.../spark/actions/TestRewriteDataFilesAction.java | 60 ++++++++++++++++++++++
.../TestRewritePositionDeleteFilesAction.java | 36 ++++++++++++-
.../TestRewritePositionDeleteFilesAction.java | 40 +++++++--------
6 files changed, 152 insertions(+), 31 deletions(-)
diff --git a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java
index 77debe1e58..b69b80a8d3 100644
--- a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java
+++ b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.actions;
import java.util.Map;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.spark.sql.SparkSession;
@@ -39,4 +40,8 @@ abstract class BaseSnapshotUpdateSparkAction<ThisT> extends BaseSparkAction<This
summary.forEach(update::set);
update.commit();
}
+
+ protected Map<String, String> commitSummary() {
+ return ImmutableMap.copyOf(summary);
+ }
}
diff --git a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
index ce0808da50..e04a0c88b4 100644
--- a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
+++ b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
@@ -82,6 +82,7 @@ public class RewriteDataFilesSparkAction
MAX_FILE_GROUP_SIZE_BYTES,
PARTIAL_PROGRESS_ENABLED,
PARTIAL_PROGRESS_MAX_COMMITS,
+ PARTIAL_PROGRESS_MAX_FAILED_COMMITS,
TARGET_FILE_SIZE_BYTES,
USE_STARTING_SEQUENCE_NUMBER,
REWRITE_JOB_ORDER,
@@ -96,6 +97,7 @@ public class RewriteDataFilesSparkAction
private Expression filter = Expressions.alwaysTrue();
private int maxConcurrentFileGroupRewrites;
private int maxCommits;
+ private int maxFailedCommits;
private boolean partialProgressEnabled;
private boolean removeDanglingDeletes;
private boolean useStartingSequenceNumber;
@@ -184,6 +186,7 @@ public class RewriteDataFilesSparkAction
partialProgressEnabled
? doExecuteWithPartialProgress(ctx, groupStream, commitManager(startingSnapshotId))
: doExecute(ctx, groupStream, commitManager(startingSnapshotId));
+
if (removeDanglingDeletes) {
RemoveDanglingDeletesSparkAction action =
new RemoveDanglingDeletesSparkAction(spark(), table);
@@ -272,7 +275,8 @@ public class RewriteDataFilesSparkAction
@VisibleForTesting
RewriteDataFilesCommitManager commitManager(long startingSnapshotId) {
- return new RewriteDataFilesCommitManager(table, startingSnapshotId, useStartingSequenceNumber);
+ return new RewriteDataFilesCommitManager(
+ table, startingSnapshotId, useStartingSequenceNumber, commitSummary());
}
private Builder doExecute(
@@ -372,20 +376,31 @@ public class RewriteDataFilesSparkAction
// stop commit service
commitService.close();
- List<RewriteFileGroup> commitResults = commitService.results();
- if (commitResults.size() == 0) {
- LOG.error(
- "{} is true but no rewrite commits succeeded. Check the logs to determine why the individual "
- + "commits failed. If this is persistent it may help to increase {} which will break the rewrite operation "
+
+ int failedCommits = maxCommits - commitService.succeededCommits();
+ if (failedCommits > 0 && failedCommits <= maxFailedCommits) {
+ LOG.warn(
+ "{} is true but {} rewrite commits failed. Check the logs to determine why the individual "
+ + "commits failed. If this is persistent it may help to increase {} which will split the rewrite operation "
+ "into smaller commits.",
PARTIAL_PROGRESS_ENABLED,
+ failedCommits,
PARTIAL_PROGRESS_MAX_COMMITS);
+ } else if (failedCommits > maxFailedCommits) {
+ String errorMessage =
+ String.format(
+ "%s is true but %d rewrite commits failed. This is more than the maximum allowed failures of %d. "
+ + "Check the logs to determine why the individual commits failed. If this is persistent it may help to "
+ + "increase %s which will split the rewrite operation into smaller commits.",
+ PARTIAL_PROGRESS_ENABLED,
+ failedCommits,
+ maxFailedCommits,
+ PARTIAL_PROGRESS_MAX_COMMITS);
+ throw new RuntimeException(errorMessage);
}
- List<FileGroupRewriteResult> rewriteResults =
- commitResults.stream().map(RewriteFileGroup::asResult).collect(Collectors.toList());
return ImmutableRewriteDataFiles.Result.builder()
- .rewriteResults(rewriteResults)
+ .rewriteResults(toRewriteResults(commitService.results()))
.rewriteFailures(rewriteFailures);
}
@@ -415,6 +430,10 @@ public class RewriteDataFilesSparkAction
return new RewriteFileGroup(info, tasks);
}
+ private Iterable<FileGroupRewriteResult> toRewriteResults(List<RewriteFileGroup> commitResults) {
+ return commitResults.stream().map(RewriteFileGroup::asResult).collect(Collectors.toList());
+ }
+
void validateAndInitOptions() {
Set<String> validOptions = Sets.newHashSet(rewriter.validOptions());
validOptions.addAll(VALID_OPTIONS);
@@ -440,6 +459,9 @@ public class RewriteDataFilesSparkAction
PropertyUtil.propertyAsInt(
options(), PARTIAL_PROGRESS_MAX_COMMITS, PARTIAL_PROGRESS_MAX_COMMITS_DEFAULT);
+ maxFailedCommits =
+ PropertyUtil.propertyAsInt(options(), PARTIAL_PROGRESS_MAX_FAILED_COMMITS, maxCommits);
+
partialProgressEnabled =
PropertyUtil.propertyAsBoolean(
options(), PARTIAL_PROGRESS_ENABLED, PARTIAL_PROGRESS_ENABLED_DEFAULT);
diff --git a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java
index 7d283ab746..6a4822b683 100644
--- a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java
+++ b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java
@@ -249,7 +249,7 @@ public class RewritePositionDeleteFilesSparkAction
}
private RewritePositionDeletesCommitManager commitManager() {
- return new RewritePositionDeletesCommitManager(table);
+ return new RewritePositionDeletesCommitManager(table, commitSummary());
}
private Result doExecute(
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
index c078015c21..a99f8c234f 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java
@@ -67,6 +67,7 @@ import org.apache.iceberg.RewriteJobOrder;
import org.apache.iceberg.RowDelta;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.SortOrder;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
@@ -1277,6 +1278,48 @@ public class TestRewriteDataFilesAction extends TestBase {
shouldHaveACleanCache(table);
}
+ @TestTemplate
+ public void testParallelPartialProgressWithMaxFailedCommits() {
+ Table table = createTable(20);
+ int fileSize = averageFileSize(table);
+
+ List<Object[]> originalData = currentData();
+
+ RewriteDataFilesSparkAction realRewrite =
+ basicRewrite(table)
+ .option(
+ RewriteDataFiles.MAX_FILE_GROUP_SIZE_BYTES, Integer.toString(fileSize * 2 + 1000))
+ .option(RewriteDataFiles.MAX_CONCURRENT_FILE_GROUP_REWRITES, "3")
+ .option(RewriteDataFiles.PARTIAL_PROGRESS_ENABLED, "true")
+ .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_COMMITS, "3")
+ .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "0");
+
+ RewriteDataFilesSparkAction spyRewrite = Mockito.spy(realRewrite);
+
+ // Fail groups 1, 3, and 7 during rewrite
+ GroupInfoMatcher failGroup = new GroupInfoMatcher(1, 3, 7);
+ doThrow(new RuntimeException("Rewrite Failed"))
+ .when(spyRewrite)
+ .rewriteFiles(any(), argThat(failGroup));
+
+ assertThatThrownBy(() -> spyRewrite.execute())
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining(
+ "1 rewrite commits failed. This is more than the maximum allowed failures of 0");
+
+ table.refresh();
+
+ List<Object[]> postRewriteData = currentData();
+ assertEquals("We shouldn't have changed the data", originalData, postRewriteData);
+
+ // With 10 original groups and max commits of 3, we have 4 groups per commit.
+ // Removing 3 groups, we are left with 4 groups and 3 groups in two commits.
+ // Adding max allowed failed commits doesn't change the number of successful commits.
+ shouldHaveSnapshots(table, 3);
+ shouldHaveNoOrphans(table);
+ shouldHaveACleanCache(table);
+ }
+
@TestTemplate
public void testInvalidOptions() {
Table table = createTable(20);
@@ -1816,6 +1859,23 @@ public class TestRewriteDataFilesAction extends TestBase {
assertThat(actual).as("Number of files order should not be ascending").isNotEqualTo(expected);
}
+ @TestTemplate
+ public void testSnapshotProperty() {
+ Table table = createTable(4);
+ Result ignored = basicRewrite(table).snapshotProperty("key", "value").execute();
+ assertThat(table.currentSnapshot().summary())
+ .containsAllEntriesOf(ImmutableMap.of("key", "value"));
+ // make sure internal produced properties are not lost
+ String[] commitMetricsKeys =
+ new String[] {
+ SnapshotSummary.ADDED_FILES_PROP,
+ SnapshotSummary.DELETED_FILES_PROP,
+ SnapshotSummary.TOTAL_DATA_FILES_PROP,
+ SnapshotSummary.CHANGED_PARTITION_COUNT_PROP
+ };
+ assertThat(table.currentSnapshot().summary()).containsKeys(commitMetricsKeys);
+ }
+
@TestTemplate
public void testBinPackRewriterWithSpecificUnparitionedOutputSpec() {
Table table = createTable(10);
diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
index fe21f510da..7d5e501c41 100644
--- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
+++ b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
@@ -52,6 +52,7 @@ import org.apache.iceberg.PositionDeletesScanTask;
import org.apache.iceberg.RowDelta;
import org.apache.iceberg.ScanTask;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
@@ -381,7 +382,7 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
assertThat(actualDeletes).as("New position deletes").isEmpty();
}
@@ -723,6 +724,39 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
assertEquals("Rows must match", expectedRecords, actualRecords);
}
+ @TestTemplate
+ public void testSnapshotProperty() throws Exception {
+ Table table = createTableUnpartitioned(2, SCALE);
+ List<DataFile> dataFiles = TestHelpers.dataFiles(table);
+ writePosDeletesForFiles(table, 2, DELETES_SCALE, dataFiles);
+ assertThat(dataFiles).hasSize(2);
+
+ List<DeleteFile> deleteFiles = deleteFiles(table);
+ assertThat(deleteFiles).hasSize(2);
+
+ Result ignored =
+ SparkActions.get(spark)
+ .rewritePositionDeletes(table)
+ .snapshotProperty("key", "value")
+ .option(SizeBasedFileRewriter.REWRITE_ALL, "true")
+ .execute();
+ assertThat(table.currentSnapshot().summary())
+ .containsAllEntriesOf(ImmutableMap.of("key", "value"));
+
+ // make sure internal produced properties are not lost
+ String[] commitMetricsKeys =
+ new String[] {
+ SnapshotSummary.ADDED_DELETE_FILES_PROP,
+ SnapshotSummary.ADDED_POS_DELETES_PROP,
+ SnapshotSummary.CHANGED_PARTITION_COUNT_PROP,
+ SnapshotSummary.REMOVED_DELETE_FILES_PROP,
+ SnapshotSummary.REMOVED_POS_DELETES_PROP,
+ SnapshotSummary.TOTAL_DATA_FILES_PROP,
+ SnapshotSummary.TOTAL_DELETE_FILES_PROP,
+ };
+ assertThat(table.currentSnapshot().summary()).containsKeys(commitMetricsKeys);
+ }
+
@TestTemplate
public void testRewriteManyColumns() throws Exception {
List<Types.NestedField> fields =
diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
index 73ff4f6504..87088e0201 100644
--- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
+++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewritePositionDeleteFilesAction.java
@@ -205,8 +205,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletes, actualDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes, actualDeletes);
}
@TestTemplate
@@ -241,8 +241,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletes, actualDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes, actualDeletes);
}
@TestTemplate
@@ -287,8 +287,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletes, actualDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes, actualDeletes);
withSQLConf(
ImmutableMap.of(SQLConf.CASE_SENSITIVE().key(), "true"),
@@ -337,8 +337,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletes, actualDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes, actualDeletes);
}
@TestTemplate
@@ -382,7 +382,7 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
assertThat(actualDeletes).as("New position deletes").isEmpty();
}
@@ -524,8 +524,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletes, actualDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes, actualDeletes);
}
@TestTemplate
@@ -572,8 +572,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
// Only non-compacted deletes remain
List<Object[]> expectedDeletesFiltered =
filterDeletes(expectedDeletes, ImmutableList.of(2), ImmutableList.of(3));
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletesFiltered, allDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletesFiltered, allDeletes);
}
@TestTemplate
@@ -624,8 +624,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletes, actualDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes, actualDeletes);
}
@TestTemplate
@@ -671,8 +671,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletes, actualDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes, actualDeletes);
}
@TestTemplate
@@ -721,7 +721,7 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
checkSequenceNumbers(table, rewrittenDeleteFiles, newDeleteFiles);
List<Object[]> actualRecords = records(table);
- assertEquals("Rows", expectedRecords, actualRecords);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
}
@TestTemplate
@@ -816,8 +816,8 @@ public class TestRewritePositionDeleteFilesAction extends CatalogTestBase {
List<Object[]> actualRecords = records(table);
List<Object[]> actualDeletes = deleteRecords(table);
- assertEquals("Rows", expectedRecords, actualRecords);
- assertEquals("Position deletes", expectedDeletes, actualDeletes);
+ assertEquals("Rows must match", expectedRecords, actualRecords);
+ assertEquals("Position deletes must match", expectedDeletes, actualDeletes);
}
@TestTemplate