This is an automated email from the ASF dual-hosted git repository.
huaxingao 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 f72900ee02 Spark 3.5: Backport apply session snapshot properties to
maintenance actions and procedures (#18801)
f72900ee02 is described below
commit f72900ee02cbb9bde4b1523d696a7258690a943c
Author: Anurag Mantripragada <[email protected]>
AuthorDate: Wed Nov 4 15:47:12 2026 -0800
Spark 3.5: Backport apply session snapshot properties to maintenance
actions and procedures (#18801)
Spark 3.5: Backport session snapshot properties for actions and procedures
(#18406)
---
.../extensions/TestRewriteManifestsProcedure.java | 17 ++++++++
.../java/org/apache/iceberg/spark/SparkUtil.java | 14 +++++++
.../org/apache/iceberg/spark/SparkWriteConf.java | 7 +---
.../actions/BaseSnapshotUpdateSparkAction.java | 8 ++++
.../actions/RemoveDanglingDeletesSparkAction.java | 6 ++-
.../spark/actions/RewriteDataFilesSparkAction.java | 4 +-
.../apache/iceberg/spark/actions/SparkActions.java | 3 +-
.../actions/TestRemoveDanglingDeleteAction.java | 3 +-
.../spark/actions/TestRewriteDataFilesAction.java | 47 ++++++++++++++++++++++
.../spark/actions/TestRewriteManifestsAction.java | 26 ++++++++++++
10 files changed, 124 insertions(+), 11 deletions(-)
diff --git
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
index eafad31b2b..d14546ac8d 100644
---
a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
+++
b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestRewriteManifestsProcedure.java
@@ -34,6 +34,7 @@ import org.apache.iceberg.io.CloseableIterable;
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 class TestRewriteManifestsProcedure extends
ExtensionsTestBase {
.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/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java
index 4a7d205466..4ba74f57c8 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkUtil.java
@@ -49,6 +49,7 @@ import org.apache.iceberg.types.Type;
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;
@@ -292,6 +293,19 @@ public class SparkUtil {
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<String, String> sessionSnapshotProperties(SparkSession
spark) {
+ return PropertyUtil.propertiesWithPrefix(
+ JavaConverters.mapAsJavaMap(spark.conf().getAll()),
+ SparkSQLProperties.SNAPSHOT_PROPERTY_PREFIX);
+ }
+
public static List<String> executorLocations() {
BlockManager driverBlockManager = SparkEnv.get().blockManager();
List<BlockManagerId> executorBlockManagerIds =
fetchPeers(driverBlockManager);
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
index 8bc91d6bc3..fa33e45857 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
@@ -51,14 +51,12 @@ import org.apache.iceberg.exceptions.ValidationException;
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.
@@ -286,10 +284,7 @@ public class SparkWriteConf {
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));
+ extraSnapshotMetadata.putAll(SparkUtil.sessionSnapshotProperties(spark));
// Add write options, overriding session configuration if necessary
writeOptions.forEach(
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java
index b69b80a8d3..182b2488c2 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSnapshotUpdateSparkAction.java
@@ -21,6 +21,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.iceberg.spark.SparkUtil;
import org.apache.spark.sql.SparkSession;
abstract class BaseSnapshotUpdateSparkAction<ThisT> extends
BaseSparkAction<ThisT> {
@@ -28,7 +29,14 @@ abstract class BaseSnapshotUpdateSparkAction<ThisT> extends
BaseSparkAction<This
private final Map<String, String> summary = Maps.newHashMap();
protected BaseSnapshotUpdateSparkAction(SparkSession spark) {
+ this(spark, ImmutableMap.of());
+ }
+
+ protected BaseSnapshotUpdateSparkAction(
+ SparkSession spark, Map<String, String> snapshotProperties) {
super(spark);
+ summary.putAll(SparkUtil.sessionSnapshotProperties(spark));
+ summary.putAll(snapshotProperties);
}
public ThisT snapshotProperty(String property, String value) {
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java
index d585750a76..300d12dfd2 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java
@@ -22,6 +22,7 @@ import java.io.Closeable;
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;
@@ -53,8 +54,9 @@ class RemoveDanglingDeletesSparkAction
private final Table table;
private final RemoveDanglingDeleteFilesAction action;
- protected RemoveDanglingDeletesSparkAction(SparkSession spark, Table table) {
- super(spark);
+ RemoveDanglingDeletesSparkAction(
+ SparkSession spark, Table table, Map<String, String> snapshotProperties)
{
+ super(spark, snapshotProperties);
this.table = table;
this.action = new RemoveDanglingDeleteFilesAction(table,
this::findDanglingDeletes);
}
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
index b444755f60..692fe133c0 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
@@ -480,7 +480,9 @@ public class RewriteDataFilesSparkAction
private RewriteDataFiles.Result executeRemoveDanglingDeletes(
ImmutableRewriteDataFiles.Result rewriteResult) {
RemoveDanglingDeletesSparkAction.Result result =
- new RemoveDanglingDeletesSparkAction(spark(),
table).toBranch(branch).execute();
+ new RemoveDanglingDeletesSparkAction(spark(), table, commitSummary())
+ .toBranch(branch)
+ .execute();
return rewriteResult.withRemovedDeleteFilesCount(
rewriteResult.removedDeleteFilesCount() +
Iterables.size(result.removedDeleteFiles()));
}
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java
index b7361c336a..161d3b10fa 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java
@@ -23,6 +23,7 @@ import org.apache.iceberg.actions.ActionsProvider;
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 class SparkActions implements ActionsProvider {
@Override
public RemoveDanglingDeleteFiles removeDanglingDeleteFiles(Table table) {
- return new RemoveDanglingDeletesSparkAction(spark, table);
+ return new RemoveDanglingDeletesSparkAction(spark, table,
ImmutableMap.of());
}
@Override
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
index 3f78bc2027..d07fb10e15 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRemoveDanglingDeleteAction.java
@@ -22,6 +22,7 @@ import java.net.InetAddress;
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 class TestRemoveDanglingDeleteAction extends
TestRemoveDanglingDeleteFile
@Override
protected RemoveDanglingDeleteFiles removeDanglingDeleteFiles(Table
actionTable) {
- return new RemoveDanglingDeletesSparkAction(spark, actionTable);
+ return new RemoveDanglingDeletesSparkAction(spark, actionTable,
ImmutableMap.of());
}
}
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 b893484c1d..831f933f8a 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
@@ -112,6 +112,7 @@ import org.apache.iceberg.spark.FileRewriteCoordinator;
import org.apache.iceberg.spark.ScanTaskSetManager;
import org.apache.iceberg.spark.SparkReadConf;
import org.apache.iceberg.spark.SparkReadOptions;
+import org.apache.iceberg.spark.SparkSQLProperties;
import org.apache.iceberg.spark.SparkSchemaUtil;
import org.apache.iceberg.spark.SparkTableUtil;
import org.apache.iceberg.spark.SparkWriteOptions;
@@ -1959,6 +1960,52 @@ public class TestRewriteDataFilesAction extends TestBase
{
assertThat(table.currentSnapshot().summary()).containsKeys(commitMetricsKeys);
}
+ @TestTemplate
+ void snapshotPropertyFromSessionConf() {
+ Table table = createTable(4);
+ withSQLConf(
+ ImmutableMap.of(SparkSQLProperties.SNAPSHOT_PROPERTY_PREFIX + "key",
"session-value"),
+ () -> {
+ 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
+ void snapshotPropertyAppliedToRemoveDanglingDeletesCommit() {
+ Table table =
+ TABLES.create(
+ SCHEMA,
+ SPEC,
+ ImmutableMap.of(TableProperties.FORMAT_VERSION,
String.valueOf(formatVersion)),
+ tableLocation);
+ writeRecords(Lists.newArrayList(new ThreeColumnRecord(1, null, "AAAA")));
+ writeEqDeleteRecord(table, "c1", 2, "c3", "CCCC");
+ table.refresh();
+
+ Result ignored =
+ basicRewrite(table)
+ .option(RewriteDataFiles.REMOVE_DANGLING_DELETES, "true")
+ .snapshotProperty("key", "value")
+ .execute();
+
+ table.refresh();
+ assertThat(table.currentSnapshot().summary())
+ .containsEntry(SnapshotSummary.REMOVED_EQ_DELETE_FILES_PROP, "1")
+ .containsEntry("key", "value");
+ }
+
@TestTemplate
public void testBinPackRewriterWithSpecificUnparitionedOutputSpec() {
Table table = createTable(10);
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
index 8d9619b6d7..a2de3c6737 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteManifestsAction.java
@@ -77,6 +77,7 @@ import
org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
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 class TestRewriteManifestsAction extends TestBase {
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();
