Skip to content

Spark: Custom snapshot property from session configuration #14545

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
+60 −0
Open

Spark: Custom snapshot property from session configuration #14545

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
60dbf68
spark session config change
owen6314 Nov 9, 2025
10af568
add one unit test
owen6314 Nov 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkSQLProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,4 +106,7 @@ private SparkSQLProperties() {}
// Controls whether to report available column statistics to Spark for query optimization.
public static final String REPORT_COLUMN_STATS = "spark.sql.iceberg.report-column-stats";
public static final boolean REPORT_COLUMN_STATS_DEFAULT = true;

// Prefix for custom snapshot properties
public static final String SNAPSHOT_PROPERTY_PREFIX = "spark.sql.iceberg.snapshot-property.";
}
9 changes: 9 additions & 0 deletions spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,12 +50,14 @@
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.spark.sql.RuntimeConfig;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.connector.write.RowLevelOperation.Command;
import org.apache.spark.sql.internal.SQLConf;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.JavaConverters;

/**
* A class for common Iceberg configs for Spark writes.
Expand Down Expand Up @@ -252,6 +254,13 @@ public long targetDeleteFileSize() {
public Map<String, String> extraSnapshotMetadata() {
Map<String, String> 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));

// Add write options, overriding session configuration if necessary
writeOptions.forEach(
(key, value) -> {
if (key.startsWith(SnapshotSummary.EXTRA_METADATA_PREFIX)) {
Expand Down
31 changes: 31 additions & 0 deletions spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/TestSparkWriteConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,6 +396,37 @@ public void testSparkConfOverride() {
}
}

@TestTemplate
public void testExtraSnapshotMetadataReflectsSessionConfig() {
Copy link
Contributor

@singhpk234 singhpk234 Nov 10, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we add a E2E test which verifies that when this is set the snapshot created actually persists the given property as snapshot property ?

Copy link
Author

@owen6314 owen6314 Nov 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the comment (and sorry for the late reply). Added a E2E test testExtraSnapshotMetadataPersistedOnWrite for this purpose.

withSQLConf(
ImmutableMap.of("spark.sql.iceberg.snapshot-property.test-key", "session-value"),
() -> {
Table table = validationCatalog.loadTable(tableIdent);
SparkWriteConf writeConf = new SparkWriteConf(spark, table, ImmutableMap.of());

Map<String, String> metadata = writeConf.extraSnapshotMetadata();

assertThat(metadata).containsEntry("test-key", "session-value");
});
}

@TestTemplate
public void testExtraSnapshotMetadataWriteOptionsOverrideSessionConfig() {
withSQLConf(
ImmutableMap.of("spark.sql.iceberg.snapshot-property.test-key", "session-value"),
() -> {
Table table = validationCatalog.loadTable(tableIdent);
Map<String, String> writeOptions =
ImmutableMap.of("snapshot-property.test-key", "write-option-value");
SparkWriteConf writeConf = new SparkWriteConf(spark, table, writeOptions);

Map<String, String> metadata = writeConf.extraSnapshotMetadata();

// Assert that writeOptions take precedence over session config
assertThat(metadata).containsEntry("test-key", "write-option-value");
});
}

@TestTemplate
public void testDataPropsDefaultsAsDeleteProps() {
List<List<Map<String, String>>> propertiesSuites =
Expand Down