From 0b1d5db9f0e7811693984ebc00f3c075888a5ec5 Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Tue, 6 Oct 2026 15:19:17 -0700 Subject: [PATCH 1/4] Spark 4.2: Apply session snapshot properties to actions and procedures Co-authored-by: Pucheng Yang --- docs/docs/spark-configuration.md | 1 + .../TestRewriteManifestsProcedure.java | 17 ++++++++++++ .../BaseSnapshotUpdateSparkAction.java | 7 +++++ .../actions/TestRewriteDataFilesAction.java | 23 ++++++++++++++++ .../actions/TestRewriteManifestsAction.java | 26 +++++++++++++++++++ 5 files changed, 74 insertions(+) diff --git a/docs/docs/spark-configuration.md b/docs/docs/spark-configuration.md index 857f7c40f38c..3994c90c86a8 100644 --- a/docs/docs/spark-configuration.md +++ b/docs/docs/spark-configuration.md @@ -206,6 +206,7 @@ val spark = SparkSession.builder() | spark.sql.iceberg.executor-cache.max-entry-size | 67108864 (64MB) | Max size per cache entry (bytes) | | spark.sql.iceberg.executor-cache.max-total-size | 134217728 (128MB) | Max total executor cache size (bytes) | | spark.sql.iceberg.executor-cache.locality.enabled | false | Enables locality-aware executor cache usage | +| spark.sql.iceberg.snapshot-property._custom-key_ | null | Adds an entry with custom-key and corresponding value to the summary of snapshots committed by writes and by the `rewrite_data_files`, `rewrite_position_delete_files`, and `rewrite_manifests` procedures. Write options and properties set on an action take precedence | | spark.sql.iceberg.merge-schema | false | Enables modifying the table schema to match the write schema. Only adds missing columns | | spark.sql.iceberg.view.schema-binding-mode | BINDING | Coercion applied to view columns: `BINDING` (widening only), `COMPENSATION` (any ANSI cast) | | spark.sql.iceberg.report-column-stats | true | Report Puffin Table Statistics if available to Spark's Cost Based Optimizer. CBO must be enabled for this to be effective | diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java index ffc0383362ce..723a952f24d4 100644 --- a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java +++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java @@ -34,6 +34,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.spark.SparkSQLProperties; import org.apache.spark.sql.AnalysisException; import org.apache.spark.sql.RowFactory; import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; @@ -106,6 +107,22 @@ public void testRewriteManifestsNoOp() { .hasSize(1); } + @TestTemplate + void rewriteManifestsWithSessionSnapshotProperty() { + sql( + "CREATE TABLE %s (id bigint NOT NULL, data string) USING iceberg PARTITIONED BY (data)", + tableName); + sql("INSERT INTO TABLE %s VALUES (1, 'a'), (2, 'b'), (3, 'c'), (4, 'd')", tableName); + sql("ALTER TABLE %s SET TBLPROPERTIES ('commit.manifest.target-size-bytes' '1')", tableName); + + withSQLConf( + ImmutableMap.of(SparkSQLProperties.SNAPSHOT_PROPERTY_PREFIX + "key", "session-value"), + () -> sql("CALL %s.system.rewrite_manifests('%s')", catalogName, tableIdent)); + + Table table = validationCatalog.loadTable(tableIdent); + assertThat(table.currentSnapshot().summary()).containsEntry("key", "session-value"); + } + @TestTemplate public void testRewriteLargeManifestsOnDatePartitionedTableWithJava8APIEnabled() { withSQLConf( diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java index b69b80a8d3a6..bc22c5f9e021 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java @@ -21,7 +21,10 @@ 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.iceberg.spark.SparkSQLProperties; +import org.apache.iceberg.util.PropertyUtil; import org.apache.spark.sql.SparkSession; +import scala.collection.JavaConverters; abstract class BaseSnapshotUpdateSparkAction extends BaseSparkAction { @@ -29,6 +32,10 @@ abstract class BaseSnapshotUpdateSparkAction extends BaseSparkAction { + Result ignored = basicRewrite(table).execute(); + assertThat(table.currentSnapshot().summary()).containsEntry("key", "session-value"); + }); + } + + @TestTemplate + void explicitSnapshotPropertyOverridesSessionConf() { + Table table = createTable(4); + withSQLConf( + ImmutableMap.of(SparkSQLProperties.SNAPSHOT_PROPERTY_PREFIX + "key", "session-value"), + () -> { + Result ignored = basicRewrite(table).snapshotProperty("key", "explicit-value").execute(); + assertThat(table.currentSnapshot().summary()).containsEntry("key", "explicit-value"); + }); + } + @TestTemplate public void testBinPackRewriterWithSpecificUnparitionedOutputSpec() { Table table = createTable(10); diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java index 8d9619b6d714..a2de3c6737f0 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java @@ -77,6 +77,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.spark.SparkSQLProperties; import org.apache.iceberg.spark.SparkTableUtil; import org.apache.iceberg.spark.SparkWriteOptions; import org.apache.iceberg.spark.TestBase; @@ -389,6 +390,31 @@ public void testRewriteSmallManifestsNonPartitionedTable() { assertThat(actualRecords).as("Rows must match").isEqualTo(expectedRecords); } + @TestTemplate + void snapshotPropertyFromSessionConf() { + Table table = + TABLES.create( + SCHEMA, + PartitionSpec.unpartitioned(), + ImmutableMap.of(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)), + tableLocation); + writeRecords(Lists.newArrayList(new ThreeColumnRecord(1, null, "AAAA"))); + writeRecords(Lists.newArrayList(new ThreeColumnRecord(2, "CCCC", "CCCC"))); + table.refresh(); + + withSQLConf( + ImmutableMap.of(SparkSQLProperties.SNAPSHOT_PROPERTY_PREFIX + "key", "session-value"), + () -> { + SparkActions.get() + .rewriteManifests(table) + .rewriteIf(manifest -> true) + .option(RewriteManifestsSparkAction.USE_CACHING, useCaching) + .execute(); + table.refresh(); + assertThat(table.currentSnapshot().summary()).containsEntry("key", "session-value"); + }); + } + @TestTemplate public void testRewriteManifestsWithCommitStateUnknownException() { PartitionSpec spec = PartitionSpec.unpartitioned(); From e6977a535529669fc642f519590edb810bda2594 Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Wed, 7 Oct 2026 17:02:22 -0700 Subject: [PATCH 2/4] Address review comments --- .../org/apache/iceberg/spark/SparkUtil.java | 14 +++++++++++ .../apache/iceberg/spark/SparkWriteConf.java | 6 +---- .../BaseSnapshotUpdateSparkAction.java | 9 ++----- .../actions/RewriteDataFilesSparkAction.java | 6 +++-- .../actions/TestRewriteDataFilesAction.java | 24 +++++++++++++++++++ 5 files changed, 45 insertions(+), 14 deletions(-) diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java index 39cd3d77b654..4c0d08cb8ba9 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java @@ -49,6 +49,7 @@ import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ByteBuffers; import org.apache.iceberg.util.Pair; +import org.apache.iceberg.util.PropertyUtil; import org.apache.spark.SparkEnv; import org.apache.spark.scheduler.ExecutorCacheTaskLocation; import org.apache.spark.sql.SparkSession; @@ -293,6 +294,19 @@ public static boolean caseSensitive(SparkSession spark) { return Boolean.parseBoolean(spark.conf().get("spark.sql.caseSensitive")); } + /** + * Returns the snapshot summary properties set in the session through {@link + * SparkSQLProperties#SNAPSHOT_PROPERTY_PREFIX}, with the prefix removed. + * + * @param spark a Spark session + * @return a map of snapshot summary property names to values + */ + public static Map sessionSnapshotProperties(SparkSession spark) { + return PropertyUtil.propertiesWithPrefix( + JavaConverters.mapAsJavaMap(spark.conf().getAll()), + SparkSQLProperties.SNAPSHOT_PROPERTY_PREFIX); + } + public static List executorLocations() { BlockManager driverBlockManager = SparkEnv.get().blockManager(); List executorBlockManagerIds = fetchPeers(driverBlockManager); diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java index 1bbffb9366a1..39aad041e6f8 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java @@ -60,7 +60,6 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import scala.collection.JavaConverters; /** * A class for common Iceberg configs for Spark writes. @@ -282,10 +281,7 @@ public Map extraSnapshotMetadata() { Map extraSnapshotMetadata = Maps.newHashMap(); // Add session configuration properties with SNAPSHOT_PROPERTY_PREFIX if necessary - extraSnapshotMetadata.putAll( - PropertyUtil.propertiesWithPrefix( - JavaConverters.mapAsJavaMap(sessionConf.getAll()), - SparkSQLProperties.SNAPSHOT_PROPERTY_PREFIX)); + extraSnapshotMetadata.putAll(SparkUtil.sessionSnapshotProperties(spark)); // Add write options, overriding session configuration if necessary extraSnapshotMetadata.putAll( diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java index bc22c5f9e021..c51f738d510e 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java @@ -21,10 +21,8 @@ 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.iceberg.spark.SparkSQLProperties; -import org.apache.iceberg.util.PropertyUtil; +import org.apache.iceberg.spark.SparkUtil; import org.apache.spark.sql.SparkSession; -import scala.collection.JavaConverters; abstract class BaseSnapshotUpdateSparkAction extends BaseSparkAction { @@ -32,10 +30,7 @@ abstract class BaseSnapshotUpdateSparkAction extends BaseSparkAction Date: Thu, 8 Oct 2026 10:55:28 -0700 Subject: [PATCH 3/4] Pass dangling deletes snapshot properties through the constructor --- .../spark/actions/BaseSnapshotUpdateSparkAction.java | 6 ++++++ .../spark/actions/RemoveDanglingDeletesSparkAction.java | 9 ++++++++- .../spark/actions/RewriteDataFilesSparkAction.java | 8 ++++---- 3 files changed, 18 insertions(+), 5 deletions(-) diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java index c51f738d510e..182b2488c277 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java @@ -29,8 +29,14 @@ abstract class BaseSnapshotUpdateSparkAction extends BaseSparkAction summary = Maps.newHashMap(); protected BaseSnapshotUpdateSparkAction(SparkSession spark) { + this(spark, ImmutableMap.of()); + } + + protected BaseSnapshotUpdateSparkAction( + SparkSession spark, Map snapshotProperties) { super(spark); summary.putAll(SparkUtil.sessionSnapshotProperties(spark)); + summary.putAll(snapshotProperties); } public ThisT snapshotProperty(String property, String value) { diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java index d585750a76d7..177b24eb72ee 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.stream.StreamSupport; import org.apache.iceberg.DeleteFile; import org.apache.iceberg.FileScanTask; @@ -36,6 +37,7 @@ import org.apache.iceberg.io.CloseableIterator; import org.apache.iceberg.io.ClosingIterator; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.spark.source.SerializableTableWithSize; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.broadcast.Broadcast; @@ -54,7 +56,12 @@ class RemoveDanglingDeletesSparkAction private final RemoveDanglingDeleteFilesAction action; protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) { - super(spark); + this(spark, table, ImmutableMap.of()); + } + + RemoveDanglingDeletesSparkAction( + SparkSession spark, Table table, Map snapshotProperties) { + super(spark, snapshotProperties); this.table = table; this.action = new RemoveDanglingDeleteFilesAction(table, this::findDanglingDeletes); } diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java index e29a60c1dfe7..cef0870765fb 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java @@ -486,10 +486,10 @@ private String jobDesc( private RewriteDataFiles.Result executeRemoveDanglingDeletes( ImmutableRewriteDataFiles.Result rewriteResult) { - RemoveDanglingDeletesSparkAction removeDanglingDeletesAction = - new RemoveDanglingDeletesSparkAction(spark(), table).toBranch(branch); - commitSummary().forEach(removeDanglingDeletesAction::snapshotProperty); - RemoveDanglingDeletesSparkAction.Result result = removeDanglingDeletesAction.execute(); + RemoveDanglingDeletesSparkAction.Result result = + new RemoveDanglingDeletesSparkAction(spark(), table, commitSummary()) + .toBranch(branch) + .execute(); return rewriteResult.withRemovedDeleteFilesCount( rewriteResult.removedDeleteFilesCount() + Iterables.size(result.removedDeleteFiles())); } From 458862073d820bc4d024612f5c16e25634dc36d5 Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Fri, 9 Oct 2026 16:49:06 -0700 Subject: [PATCH 4/4] document session snapshot properties --- docs/docs/spark-configuration.md | 14 ++++++++++++++ .../actions/RemoveDanglingDeletesSparkAction.java | 5 ----- .../apache/iceberg/spark/actions/SparkActions.java | 3 ++- .../actions/TestRemoveDanglingDeleteAction.java | 3 ++- 4 files changed, 18 insertions(+), 7 deletions(-) diff --git a/docs/docs/spark-configuration.md b/docs/docs/spark-configuration.md index 3994c90c86a8..d67ca9a5df76 100644 --- a/docs/docs/spark-configuration.md +++ b/docs/docs/spark-configuration.md @@ -287,3 +287,17 @@ CommitMetadata.withCommitProperties(properties, }, RuntimeException.class); ``` + +Custom metadata can also be added to snapshot summaries for an entire Spark session by setting properties with the `spark.sql.iceberg.snapshot-property.` prefix. The prefix is removed from each property. These properties apply to writes and to the `rewrite_data_files`, `rewrite_position_delete_files`, and `rewrite_manifests` procedures. Session properties act as defaults. A property set explicitly, through a write option, `CommitMetadata`, or `snapshotProperty()` on a Spark action, overrides a session property with the same key. Here is an example: + +```sql +SET spark.sql.iceberg.snapshot-property.created-by=maintenance-job; +CALL catalog.system.rewrite_data_files(table => 'db.sample'); +``` + +```java +SparkActions.get(spark) + .rewriteDataFiles(table) + .snapshotProperty("created-by", "backfill-job") + .execute(); +``` diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java index 177b24eb72ee..300d12dfd2b3 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java @@ -37,7 +37,6 @@ import org.apache.iceberg.io.CloseableIterator; import org.apache.iceberg.io.ClosingIterator; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; -import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.spark.source.SerializableTableWithSize; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.broadcast.Broadcast; @@ -55,10 +54,6 @@ class RemoveDanglingDeletesSparkAction private final Table table; private final RemoveDanglingDeleteFilesAction action; - protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) { - this(spark, table, ImmutableMap.of()); - } - RemoveDanglingDeletesSparkAction( SparkSession spark, Table table, Map snapshotProperties) { super(spark, snapshotProperties); diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java index b7361c336a69..161d3b10fa21 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java @@ -23,6 +23,7 @@ import org.apache.iceberg.actions.ComputePartitionStats; import org.apache.iceberg.actions.ComputeTableStats; import org.apache.iceberg.actions.RemoveDanglingDeleteFiles; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.spark.Spark3Util; import org.apache.iceberg.spark.Spark3Util.CatalogAndIdentifier; import org.apache.spark.sql.SparkSession; @@ -112,7 +113,7 @@ public ComputePartitionStats computePartitionStats(Table table) { @Override public RemoveDanglingDeleteFiles removeDanglingDeleteFiles(Table table) { - return new RemoveDanglingDeletesSparkAction(spark, table); + return new RemoveDanglingDeletesSparkAction(spark, table, ImmutableMap.of()); } @Override diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java index 3f78bc20275a..d07fb10e15a6 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java @@ -22,6 +22,7 @@ import org.apache.iceberg.Table; import org.apache.iceberg.actions.RemoveDanglingDeleteFiles; import org.apache.iceberg.actions.TestRemoveDanglingDeleteFilesAction; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.spark.TestBase; import org.apache.spark.sql.SparkSession; import org.junit.jupiter.api.AfterAll; @@ -55,6 +56,6 @@ public static void stopSpark() { @Override protected RemoveDanglingDeleteFiles removeDanglingDeleteFiles(Table actionTable) { - return new RemoveDanglingDeletesSparkAction(spark, actionTable); + return new RemoveDanglingDeletesSparkAction(spark, actionTable, ImmutableMap.of()); } }