You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by sz...@apache.org on 2026/09/14 23:30:52 UTC
(iceberg) branch main updated: Spark 4.2: Add insert, update, and delete write summaries to snapshot summary (#17955)
This is an automated email from the ASF dual-hosted git repository.
szehon-ho 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 c4784768b2 Spark 4.2: Add insert, update, and delete write summaries to snapshot summary (#17955)
c4784768b2 is described below
commit c4784768b2f38c5a8c70d8e0e5a07f4302f9282f
Author: Rahul Shivu Mahadev <51...@users.noreply.github.com>
AuthorDate: Mon Sep 14 16:30:45 2026 -0700
Spark 4.2: Add insert, update, and delete write summaries to snapshot summary (#17955)
---
docs/docs/spark-writes.md | 35 ++++++++++
.../extensions/TestCopyOnWriteDeleteMetrics.java | 41 ++++++++++++
.../extensions/TestCopyOnWriteUpdateMetrics.java | 41 ++++++++++++
.../spark/extensions/TestDeleteMetrics.java | 78 ++++++++++++++++++++++
.../spark/extensions/TestInsertMetrics.java | 71 ++++++++++++++++++++
.../extensions/TestMergeOnReadDeleteMetrics.java | 41 ++++++++++++
.../extensions/TestMergeOnReadUpdateMetrics.java | 41 ++++++++++++
.../spark/extensions/TestUpdateMetrics.java | 78 ++++++++++++++++++++++
.../iceberg/spark/source/BaseSparkWrite.java | 20 ++++++
.../spark/source/SparkPositionDeltaWrite.java | 9 +++
.../apache/iceberg/spark/source/SparkWrite.java | 16 ++++-
11 files changed, 470 insertions(+), 1 deletion(-)
diff --git a/docs/docs/spark-writes.md b/docs/docs/spark-writes.md
index 959993b20f..44d343738d 100644
--- a/docs/docs/spark-writes.md
+++ b/docs/docs/spark-writes.md
@@ -53,6 +53,17 @@ INSERT INTO prod.db.table VALUES (1, 'a'), (2, 'b')
INSERT INTO prod.db.table SELECT ...
```
+#### Snapshot summary
+
+After an `INSERT INTO` commit, the [snapshot summary](../../spec.md#optional-snapshot-summary-fields) may include the following field. The value is the string form of a non-negative count. The field is omitted when the value is unknown (e.g., not reported by Spark).
+
+!!! info
+ Only available in Spark 4.2 and higher.
+
+| Field | Description |
+|--------------------------------------|-------------------------|
+| **`spark.insert.num-inserted-rows`** | Number of rows inserted |
+
### `MERGE INTO`
Spark supports `MERGE INTO` queries that can express row-level updates.
@@ -196,6 +207,18 @@ WHERE EXISTS (SELECT oid FROM prod.db.returned_orders WHERE t1.oid = oid)
If the delete filter matches entire partitions of the table, Iceberg will perform a metadata-only delete. If the filter matches individual rows of a table, then Iceberg will rewrite only the affected data files.
+#### Snapshot summary
+
+After a `DELETE FROM` commit, the [snapshot summary](../../spec.md#optional-snapshot-summary-fields) may include the following fields. Each value is the string form of a non-negative count. A field is omitted when the value is unknown (e.g., not reported by Spark).
+
+!!! info
+ Only available in Spark 4.2 and higher.
+
+| Field | Description |
+|-------------------------------------|------------------------------------------------------------------------------------------|
+| **`spark.delete.num-deleted-rows`** | Number of rows deleted |
+| **`spark.delete.num-copied-rows`** | Number of unmodified rows copied (rewritten in copy-on-write mode; `0` in merge-on-read) |
+
### `UPDATE`
Update queries accept a filter to match rows to update.
@@ -216,6 +239,18 @@ WHERE EXISTS (SELECT oid FROM prod.db.returned_orders WHERE t1.oid = oid)
For more complex row-level updates based on incoming data, see the section on `MERGE INTO`.
+#### Snapshot summary
+
+After an `UPDATE` commit, the [snapshot summary](../../spec.md#optional-snapshot-summary-fields) may include the following fields. Each value is the string form of a non-negative count. A field is omitted when the value is unknown (e.g., not reported by Spark).
+
+!!! info
+ Only available in Spark 4.2 and higher.
+
+| Field | Description |
+|-------------------------------------|------------------------------------------------------------------------------------------|
+| **`spark.update.num-updated-rows`** | Number of rows updated |
+| **`spark.update.num-copied-rows`** | Number of unmodified rows copied (rewritten in copy-on-write mode; `0` in merge-on-read) |
+
## Writing to Branches
The branch must exist before performing write. Operations do **not** create the branch if it does not exist.
diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCopyOnWriteDeleteMetrics.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCopyOnWriteDeleteMetrics.java
new file mode 100644
index 0000000000..8abd8c232b
--- /dev/null
+++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCopyOnWriteDeleteMetrics.java
@@ -0,0 +1,41 @@
+/*
+ * 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.spark.extensions;
+
+import java.util.Map;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.RowLevelOperationMode;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestCopyOnWriteDeleteMetrics extends TestDeleteMetrics {
+
+ @Override
+ protected Map<String, String> extraTableProperties() {
+ return ImmutableMap.of(
+ TableProperties.DELETE_MODE, RowLevelOperationMode.COPY_ON_WRITE.modeName());
+ }
+
+ @Override
+ protected long expectedRowsCopied(long unchangedRowsInModifiedFiles) {
+ return unchangedRowsInModifiedFiles;
+ }
+}
diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCopyOnWriteUpdateMetrics.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCopyOnWriteUpdateMetrics.java
new file mode 100644
index 0000000000..4caf89a781
--- /dev/null
+++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCopyOnWriteUpdateMetrics.java
@@ -0,0 +1,41 @@
+/*
+ * 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.spark.extensions;
+
+import java.util.Map;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.RowLevelOperationMode;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestCopyOnWriteUpdateMetrics extends TestUpdateMetrics {
+
+ @Override
+ protected Map<String, String> extraTableProperties() {
+ return ImmutableMap.of(
+ TableProperties.UPDATE_MODE, RowLevelOperationMode.COPY_ON_WRITE.modeName());
+ }
+
+ @Override
+ protected long expectedRowsCopied(long unchangedRowsInModifiedFiles) {
+ return unchangedRowsInModifiedFiles;
+ }
+}
diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDeleteMetrics.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDeleteMetrics.java
new file mode 100644
index 0000000000..bca940aefa
--- /dev/null
+++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDeleteMetrics.java
@@ -0,0 +1,78 @@
+/*
+ * 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.spark.extensions;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests for verifying that delete metrics from Spark's DeleteSummary are persisted in Iceberg's
+ * snapshot summary.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public abstract class TestDeleteMetrics extends SparkRowLevelOperationsTestBase {
+
+ @BeforeAll
+ public static void setupSparkConf() {
+ spark.conf().set("spark.sql.shuffle.partitions", "4");
+ }
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ protected abstract long expectedRowsCopied(long unchangedRowsInModifiedFiles);
+
+ @TestTemplate
+ public void testDeleteMetrics() {
+ createAndInitTable(
+ "id INT, dep STRING",
+ "{ \"id\": 1, \"dep\": \"hr\" }\n"
+ + "{ \"id\": 2, \"dep\": \"software\" }\n"
+ + "{ \"id\": 3, \"dep\": \"finance\" }");
+
+ sql("DELETE FROM %s WHERE id IN (1, 2)", commitTarget());
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot currentSnapshot = SnapshotUtil.latestSnapshot(table, branch);
+ Map<String, String> summary = currentSnapshot.summary();
+
+ assertDeleteMetric(summary, "spark.delete.num-deleted-rows", 2);
+ assertDeleteMetric(summary, "spark.delete.num-copied-rows", expectedRowsCopied(1));
+ }
+
+ private void assertDeleteMetric(Map<String, String> summary, String key, long expectedValue) {
+ assertThat(summary)
+ .as("Snapshot summary should contain delete metric: " + key)
+ .containsKey(key);
+ assertThat(summary.get(key))
+ .as("Delete metric " + key + " should have expected value")
+ .isEqualTo(String.valueOf(expectedValue));
+ }
+}
diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestInsertMetrics.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestInsertMetrics.java
new file mode 100644
index 0000000000..dd23b1e1f9
--- /dev/null
+++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestInsertMetrics.java
@@ -0,0 +1,71 @@
+/*
+ * 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.spark.extensions;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests for verifying that insert metrics from Spark's InsertSummary are persisted in Iceberg's
+ * snapshot summary.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestInsertMetrics extends SparkRowLevelOperationsTestBase {
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @Override
+ protected Map<String, String> extraTableProperties() {
+ return ImmutableMap.of();
+ }
+
+ @TestTemplate
+ public void testInsertMetrics() {
+ createAndInitTable("id INT, dep STRING", "{ \"id\": 1, \"dep\": \"hr\" }");
+
+ sql("INSERT INTO %s VALUES (2, 'software'), (3, 'finance'), (4, 'marketing')", commitTarget());
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot currentSnapshot = SnapshotUtil.latestSnapshot(table, branch);
+ Map<String, String> summary = currentSnapshot.summary();
+
+ assertInsertMetric(summary, "spark.insert.num-inserted-rows", 3);
+ }
+
+ private void assertInsertMetric(Map<String, String> summary, String key, long expectedValue) {
+ assertThat(summary)
+ .as("Snapshot summary should contain insert metric: " + key)
+ .containsKey(key);
+ assertThat(summary.get(key))
+ .as("Insert metric " + key + " should have expected value")
+ .isEqualTo(String.valueOf(expectedValue));
+ }
+}
diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeOnReadDeleteMetrics.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeOnReadDeleteMetrics.java
new file mode 100644
index 0000000000..32177aa59c
--- /dev/null
+++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeOnReadDeleteMetrics.java
@@ -0,0 +1,41 @@
+/*
+ * 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.spark.extensions;
+
+import java.util.Map;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.RowLevelOperationMode;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestMergeOnReadDeleteMetrics extends TestDeleteMetrics {
+
+ @Override
+ protected Map<String, String> extraTableProperties() {
+ return ImmutableMap.of(
+ TableProperties.DELETE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName());
+ }
+
+ @Override
+ protected long expectedRowsCopied(long unchangedRowsInModifiedFiles) {
+ return 0;
+ }
+}
diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeOnReadUpdateMetrics.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeOnReadUpdateMetrics.java
new file mode 100644
index 0000000000..4651a4e331
--- /dev/null
+++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeOnReadUpdateMetrics.java
@@ -0,0 +1,41 @@
+/*
+ * 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.spark.extensions;
+
+import java.util.Map;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.RowLevelOperationMode;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestMergeOnReadUpdateMetrics extends TestUpdateMetrics {
+
+ @Override
+ protected Map<String, String> extraTableProperties() {
+ return ImmutableMap.of(
+ TableProperties.UPDATE_MODE, RowLevelOperationMode.MERGE_ON_READ.modeName());
+ }
+
+ @Override
+ protected long expectedRowsCopied(long unchangedRowsInModifiedFiles) {
+ return 0;
+ }
+}
diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestUpdateMetrics.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestUpdateMetrics.java
new file mode 100644
index 0000000000..b5ecb98b91
--- /dev/null
+++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestUpdateMetrics.java
@@ -0,0 +1,78 @@
+/*
+ * 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.spark.extensions;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.util.SnapshotUtil;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests for verifying that update metrics from Spark's UpdateSummary are persisted in Iceberg's
+ * snapshot summary.
+ */
+@ExtendWith(ParameterizedTestExtension.class)
+public abstract class TestUpdateMetrics extends SparkRowLevelOperationsTestBase {
+
+ @BeforeAll
+ public static void setupSparkConf() {
+ spark.conf().set("spark.sql.shuffle.partitions", "4");
+ }
+
+ @AfterEach
+ public void removeTables() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ protected abstract long expectedRowsCopied(long unchangedRowsInModifiedFiles);
+
+ @TestTemplate
+ public void testUpdateMetrics() {
+ createAndInitTable(
+ "id INT, dep STRING",
+ "{ \"id\": 1, \"dep\": \"hr\" }\n"
+ + "{ \"id\": 2, \"dep\": \"software\" }\n"
+ + "{ \"id\": 3, \"dep\": \"finance\" }");
+
+ sql("UPDATE %s SET dep = 'updated' WHERE id = 1", commitTarget());
+
+ Table table = validationCatalog.loadTable(tableIdent);
+ Snapshot currentSnapshot = SnapshotUtil.latestSnapshot(table, branch);
+ Map<String, String> summary = currentSnapshot.summary();
+
+ assertUpdateMetric(summary, "spark.update.num-updated-rows", 1);
+ assertUpdateMetric(summary, "spark.update.num-copied-rows", expectedRowsCopied(2));
+ }
+
+ private void assertUpdateMetric(Map<String, String> summary, String key, long expectedValue) {
+ assertThat(summary)
+ .as("Snapshot summary should contain update metric: " + key)
+ .containsKey(key);
+ assertThat(summary.get(key))
+ .as("Update metric " + key + " should have expected value")
+ .isEqualTo(String.valueOf(expectedValue));
+ }
+}
diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkWrite.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkWrite.java
index 3ba11f3bbd..57c3d3ba22 100644
--- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkWrite.java
+++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseSparkWrite.java
@@ -19,7 +19,10 @@
package org.apache.iceberg.spark.source;
import org.apache.iceberg.SnapshotUpdate;
+import org.apache.spark.sql.connector.write.DeleteSummary;
+import org.apache.spark.sql.connector.write.InsertSummary;
import org.apache.spark.sql.connector.write.MergeSummary;
+import org.apache.spark.sql.connector.write.UpdateSummary;
/** Base class for Spark write implementations with shared utility methods. */
abstract class BaseSparkWrite {
@@ -53,6 +56,23 @@ abstract class BaseSparkWrite {
mergeSummary.numTargetRowsNotMatchedBySourceDeleted());
}
+ protected void setUpdateSummaryProperties(
+ SnapshotUpdate<?> operation, UpdateSummary updateSummary) {
+ setIfPositive(operation, "spark.update.num-updated-rows", updateSummary.numUpdatedRows());
+ setIfPositive(operation, "spark.update.num-copied-rows", updateSummary.numCopiedRows());
+ }
+
+ protected void setDeleteSummaryProperties(
+ SnapshotUpdate<?> operation, DeleteSummary deleteSummary) {
+ setIfPositive(operation, "spark.delete.num-deleted-rows", deleteSummary.numDeletedRows());
+ setIfPositive(operation, "spark.delete.num-copied-rows", deleteSummary.numCopiedRows());
+ }
+
+ protected void setInsertSummaryProperties(
+ SnapshotUpdate<?> operation, InsertSummary insertSummary) {
+ setIfPositive(operation, "spark.insert.num-inserted-rows", insertSummary.numInsertedRows());
+ }
+
private void setIfPositive(SnapshotUpdate<?> operation, String key, long value) {
if (value >= 0) {
operation.set(key, String.valueOf(value));
diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java
index 3ede6996a8..5754783f23 100644
--- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java
+++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java
@@ -87,15 +87,18 @@ import org.apache.spark.sql.connector.distributions.Distribution;
import org.apache.spark.sql.connector.expressions.SortOrder;
import org.apache.spark.sql.connector.metric.CustomMetric;
import org.apache.spark.sql.connector.metric.CustomTaskMetric;
+import org.apache.spark.sql.connector.write.DeleteSummary;
import org.apache.spark.sql.connector.write.DeltaBatchWrite;
import org.apache.spark.sql.connector.write.DeltaWrite;
import org.apache.spark.sql.connector.write.DeltaWriter;
import org.apache.spark.sql.connector.write.DeltaWriterFactory;
+import org.apache.spark.sql.connector.write.InsertSummary;
import org.apache.spark.sql.connector.write.LogicalWriteInfo;
import org.apache.spark.sql.connector.write.MergeSummary;
import org.apache.spark.sql.connector.write.PhysicalWriteInfo;
import org.apache.spark.sql.connector.write.RequiresDistributionAndOrdering;
import org.apache.spark.sql.connector.write.RowLevelOperation.Command;
+import org.apache.spark.sql.connector.write.UpdateSummary;
import org.apache.spark.sql.connector.write.WriteSummary;
import org.apache.spark.sql.connector.write.WriterCommitMessage;
import org.apache.spark.sql.types.LongType$;
@@ -351,6 +354,12 @@ class SparkPositionDeltaWrite extends BaseSparkWrite
if (summary instanceof MergeSummary) {
setMergeSummaryProperties(operation, (MergeSummary) summary);
+ } else if (summary instanceof UpdateSummary) {
+ setUpdateSummaryProperties(operation, (UpdateSummary) summary);
+ } else if (summary instanceof DeleteSummary) {
+ setDeleteSummaryProperties(operation, (DeleteSummary) summary);
+ } else if (summary instanceof InsertSummary) {
+ setInsertSummaryProperties(operation, (InsertSummary) summary);
}
if (wapEnabled && wapId != null) {
diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
index f589fdd8f4..5f06699d2a 100644
--- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
+++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
@@ -79,10 +79,13 @@ import org.apache.spark.sql.connector.metric.CustomTaskMetric;
import org.apache.spark.sql.connector.write.BatchWrite;
import org.apache.spark.sql.connector.write.DataWriter;
import org.apache.spark.sql.connector.write.DataWriterFactory;
+import org.apache.spark.sql.connector.write.DeleteSummary;
+import org.apache.spark.sql.connector.write.InsertSummary;
import org.apache.spark.sql.connector.write.LogicalWriteInfo;
import org.apache.spark.sql.connector.write.MergeSummary;
import org.apache.spark.sql.connector.write.PhysicalWriteInfo;
import org.apache.spark.sql.connector.write.RequiresDistributionAndOrdering;
+import org.apache.spark.sql.connector.write.UpdateSummary;
import org.apache.spark.sql.connector.write.Write;
import org.apache.spark.sql.connector.write.WriteSummary;
import org.apache.spark.sql.connector.write.WriterCommitMessage;
@@ -251,6 +254,12 @@ abstract class SparkWrite extends BaseSparkWrite implements Write, RequiresDistr
if (summary instanceof MergeSummary) {
setMergeSummaryProperties(operation, (MergeSummary) summary);
+ } else if (summary instanceof UpdateSummary) {
+ setUpdateSummaryProperties(operation, (UpdateSummary) summary);
+ } else if (summary instanceof DeleteSummary) {
+ setDeleteSummaryProperties(operation, (DeleteSummary) summary);
+ } else if (summary instanceof InsertSummary) {
+ setInsertSummaryProperties(operation, (InsertSummary) summary);
}
if (wapEnabled && wapId != null) {
@@ -331,6 +340,11 @@ abstract class SparkWrite extends BaseSparkWrite implements Write, RequiresDistr
private class BatchAppend extends BaseBatchWrite {
@Override
public void commit(WriterCommitMessage[] messages) {
+ commit(messages, null);
+ }
+
+ @Override
+ public void commit(WriterCommitMessage[] messages, WriteSummary summary) {
AppendFiles append = table.newAppend();
int numFiles = 0;
@@ -340,7 +354,7 @@ abstract class SparkWrite extends BaseSparkWrite implements Write, RequiresDistr
}
commitOperation(
- append, String.format(Locale.ROOT, "append with %d new data files", numFiles));
+ append, String.format(Locale.ROOT, "append with %d new data files", numFiles), summary);
}
}