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 2026/05/07 23:43:03 UTC
(iceberg) branch main updated: Spark: backport PR #15512 to v3.4, v3.5, v4.0 for WAP branch delete fix (#16245)
This is an automated email from the ASF dual-hosted git repository.
amogh-jahagirdar 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 57b1211b74 Spark: backport PR #15512 to v3.4, v3.5, v4.0 for WAP branch delete fix (#16245)
57b1211b74 is described below
commit 57b1211b7477f9f4a5e79a4cf6f6505d63ded4e8
Author: Steven Zhen Wu <st...@gmail.com>
AuthorDate: Thu May 7 16:42:57 2026 -0700
Spark: backport PR #15512 to v3.4, v3.5, v4.0 for WAP branch delete fix (#16245)
* Spark: backport PR #15512 to v3.4, v3.5, v4.0 for WAP branch delete fix
When WAP is enabled via spark.wap.branch, canDeleteWhere() previously
scanned the main branch while deleteWhere() committed to the WAP branch.
This could cause canDeleteWhere() to incorrectly approve a metadata-only
delete based on data that was never on the WAP branch, surfacing as
"Cannot delete file where some, but not all, rows match filter" at
commit time.
Resolve the scan branch the same way deleteWhere resolves the write
branch (with a fall-back to main when the WAP branch has not been
created yet), and pass it through canDeleteUsingMetadata.
Co-Authored-By: Claude Opus 4.7 (1M context) <no...@anthropic.com>
* Spark: add blank lines after if blocks in scanBranchForDelete (style)
Iceberg style requires an empty line between a control flow block and
the following statement.
Co-Authored-By: Claude Opus 4.7 (1M context) <no...@anthropic.com>
---
.../iceberg/spark/extensions/TestDelete.java | 56 ++++++++++++++++++++++
.../apache/iceberg/spark/source/SparkTable.java | 31 ++++++++++--
.../iceberg/spark/extensions/TestDelete.java | 56 ++++++++++++++++++++++
.../apache/iceberg/spark/source/SparkTable.java | 31 ++++++++++--
.../iceberg/spark/extensions/TestDelete.java | 56 ++++++++++++++++++++++
.../apache/iceberg/spark/source/SparkTable.java | 31 ++++++++++--
6 files changed, 246 insertions(+), 15 deletions(-)
diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index 1dd6db48f7..b106e8fc38 100644
--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -1395,6 +1395,62 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
});
}
+ @TestTemplate
+ public void testDeleteToWapBranchCanDeleteWhereScansWapBranch() throws NoSuchTableException {
+ assumeThat(branch).as("WAP branch only works for table identifier without branch").isNull();
+
+ createAndInitPartitionedTable();
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = 'true')",
+ tableName, TableProperties.WRITE_AUDIT_PUBLISH_ENABLED);
+
+ append(tableName, new Employee(1, "hr"));
+
+ spark.conf().set(SparkSQLProperties.WAP_BRANCH, "wap");
+ try {
+ append(tableName, new Employee(0, "hr"), new Employee(1, "hr"), new Employee(2, "hr"));
+
+ sql("DELETE FROM %s WHERE id = 1", tableName);
+
+ assertThat(sql("SELECT id, dep FROM %s.branch_wap ORDER BY id", tableName))
+ .as("DELETE should remove the matching rows from the WAP branch")
+ .containsExactly(row(0, "hr"), row(2, "hr"));
+ assertThat(sql("SELECT id, dep FROM %s.branch_main", tableName))
+ .as("Main branch must not be modified by a WAP-targeted DELETE")
+ .containsExactly(row(1, "hr"));
+ } finally {
+ spark.conf().unset(SparkSQLProperties.WAP_BRANCH);
+ }
+ }
+
+ @TestTemplate
+ public void testMetadataDeleteToWapBranchCommitsToWapBranch() throws NoSuchTableException {
+ assumeThat(branch).as("WAP branch only works for table identifier without branch").isNull();
+
+ createAndInitPartitionedTable();
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = 'true')",
+ tableName, TableProperties.WRITE_AUDIT_PUBLISH_ENABLED);
+
+ append(tableName, new Employee(1, "hr"), new Employee(5, "eng"));
+
+ spark.conf().set(SparkSQLProperties.WAP_BRANCH, "wap");
+ try {
+ append(tableName, new Employee(0, "hr"), new Employee(2, "eng"));
+
+ sql("DELETE FROM %s WHERE dep = 'hr'", tableName);
+
+ assertThat(sql("SELECT id, dep FROM %s.branch_wap ORDER BY id", tableName))
+ .as("Metadata delete should remove the hr partition on the WAP branch")
+ .containsExactly(row(2, "eng"), row(5, "eng"));
+ assertThat(sql("SELECT id, dep FROM %s.branch_main ORDER BY id", tableName))
+ .as("Metadata delete must not commit to main when WAP is set")
+ .containsExactly(row(1, "hr"), row(5, "eng"));
+ } finally {
+ spark.conf().unset(SparkSQLProperties.WAP_BRANCH);
+ }
+ }
+
@TestTemplate
public void testDeleteWithFilterOnNestedColumn() {
createAndInitNestedColumnsTable();
diff --git a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
index 871ef93552..1348afff64 100644
--- a/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
+++ b/spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
@@ -57,6 +57,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.spark.CommitMetadata;
import org.apache.iceberg.spark.Spark3Util;
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.SparkUtil;
@@ -334,11 +335,31 @@ public class SparkTable
}
}
- return canDeleteUsingMetadata(deleteExpr);
+ return canDeleteUsingMetadata(deleteExpr, scanBranchForDelete());
+ }
+
+ // Resolves the branch to scan during canDeleteWhere so it matches the branch deleteWhere
+ // will commit to. Falls back to main when WAP is configured but the WAP branch does not
+ // exist yet, since this is a read scan.
+ private String scanBranchForDelete() {
+ if (branch != null) {
+ return branch;
+ }
+
+ if (!SparkTableUtil.wapEnabled(table())) {
+ return null;
+ }
+
+ String wapBranch = sparkSession().conf().get(SparkSQLProperties.WAP_BRANCH, null);
+ if (wapBranch != null && table().refs().containsKey(wapBranch)) {
+ return wapBranch;
+ }
+
+ return null;
}
// a metadata delete is possible iff matching files can be deleted entirely
- private boolean canDeleteUsingMetadata(Expression deleteExpr) {
+ private boolean canDeleteUsingMetadata(Expression deleteExpr, String scanBranch) {
boolean caseSensitive = SparkUtil.caseSensitive(sparkSession());
if (ExpressionUtil.selectsPartitions(deleteExpr, table(), caseSensitive)) {
@@ -353,14 +374,14 @@ public class SparkTable
.includeColumnStats()
.ignoreResiduals();
- if (branch != null) {
- scan = scan.useRef(branch);
+ if (scanBranch != null) {
+ scan = scan.useRef(scanBranch);
}
try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
Map<Integer, Evaluator> evaluators = Maps.newHashMap();
StrictMetricsEvaluator metricsEvaluator =
- new StrictMetricsEvaluator(SnapshotUtil.schemaFor(table(), branch), deleteExpr);
+ new StrictMetricsEvaluator(SnapshotUtil.schemaFor(table(), scanBranch), deleteExpr);
return Iterables.all(
tasks,
diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index fbf6ce3559..79d6bea12f 100644
--- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -1422,6 +1422,62 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
});
}
+ @TestTemplate
+ public void testDeleteToWapBranchCanDeleteWhereScansWapBranch() throws NoSuchTableException {
+ assumeThat(branch).as("WAP branch only works for table identifier without branch").isNull();
+
+ createAndInitPartitionedTable();
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = 'true')",
+ tableName, TableProperties.WRITE_AUDIT_PUBLISH_ENABLED);
+
+ append(tableName, new Employee(1, "hr"));
+
+ spark.conf().set(SparkSQLProperties.WAP_BRANCH, "wap");
+ try {
+ append(tableName, new Employee(0, "hr"), new Employee(1, "hr"), new Employee(2, "hr"));
+
+ sql("DELETE FROM %s WHERE id = 1", tableName);
+
+ assertThat(sql("SELECT id, dep FROM %s.branch_wap ORDER BY id", tableName))
+ .as("DELETE should remove the matching rows from the WAP branch")
+ .containsExactly(row(0, "hr"), row(2, "hr"));
+ assertThat(sql("SELECT id, dep FROM %s.branch_main", tableName))
+ .as("Main branch must not be modified by a WAP-targeted DELETE")
+ .containsExactly(row(1, "hr"));
+ } finally {
+ spark.conf().unset(SparkSQLProperties.WAP_BRANCH);
+ }
+ }
+
+ @TestTemplate
+ public void testMetadataDeleteToWapBranchCommitsToWapBranch() throws NoSuchTableException {
+ assumeThat(branch).as("WAP branch only works for table identifier without branch").isNull();
+
+ createAndInitPartitionedTable();
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = 'true')",
+ tableName, TableProperties.WRITE_AUDIT_PUBLISH_ENABLED);
+
+ append(tableName, new Employee(1, "hr"), new Employee(5, "eng"));
+
+ spark.conf().set(SparkSQLProperties.WAP_BRANCH, "wap");
+ try {
+ append(tableName, new Employee(0, "hr"), new Employee(2, "eng"));
+
+ sql("DELETE FROM %s WHERE dep = 'hr'", tableName);
+
+ assertThat(sql("SELECT id, dep FROM %s.branch_wap ORDER BY id", tableName))
+ .as("Metadata delete should remove the hr partition on the WAP branch")
+ .containsExactly(row(2, "eng"), row(5, "eng"));
+ assertThat(sql("SELECT id, dep FROM %s.branch_main ORDER BY id", tableName))
+ .as("Metadata delete must not commit to main when WAP is set")
+ .containsExactly(row(1, "hr"), row(5, "eng"));
+ } finally {
+ spark.conf().unset(SparkSQLProperties.WAP_BRANCH);
+ }
+ }
+
@TestTemplate
public void testDeleteWithFilterOnNestedColumn() {
createAndInitNestedColumnsTable();
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
index 871ef93552..1348afff64 100644
--- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
+++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
@@ -57,6 +57,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.spark.CommitMetadata;
import org.apache.iceberg.spark.Spark3Util;
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.SparkUtil;
@@ -334,11 +335,31 @@ public class SparkTable
}
}
- return canDeleteUsingMetadata(deleteExpr);
+ return canDeleteUsingMetadata(deleteExpr, scanBranchForDelete());
+ }
+
+ // Resolves the branch to scan during canDeleteWhere so it matches the branch deleteWhere
+ // will commit to. Falls back to main when WAP is configured but the WAP branch does not
+ // exist yet, since this is a read scan.
+ private String scanBranchForDelete() {
+ if (branch != null) {
+ return branch;
+ }
+
+ if (!SparkTableUtil.wapEnabled(table())) {
+ return null;
+ }
+
+ String wapBranch = sparkSession().conf().get(SparkSQLProperties.WAP_BRANCH, null);
+ if (wapBranch != null && table().refs().containsKey(wapBranch)) {
+ return wapBranch;
+ }
+
+ return null;
}
// a metadata delete is possible iff matching files can be deleted entirely
- private boolean canDeleteUsingMetadata(Expression deleteExpr) {
+ private boolean canDeleteUsingMetadata(Expression deleteExpr, String scanBranch) {
boolean caseSensitive = SparkUtil.caseSensitive(sparkSession());
if (ExpressionUtil.selectsPartitions(deleteExpr, table(), caseSensitive)) {
@@ -353,14 +374,14 @@ public class SparkTable
.includeColumnStats()
.ignoreResiduals();
- if (branch != null) {
- scan = scan.useRef(branch);
+ if (scanBranch != null) {
+ scan = scan.useRef(scanBranch);
}
try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
Map<Integer, Evaluator> evaluators = Maps.newHashMap();
StrictMetricsEvaluator metricsEvaluator =
- new StrictMetricsEvaluator(SnapshotUtil.schemaFor(table(), branch), deleteExpr);
+ new StrictMetricsEvaluator(SnapshotUtil.schemaFor(table(), scanBranch), deleteExpr);
return Iterables.all(
tasks,
diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
index fbf6ce3559..79d6bea12f 100644
--- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
+++ b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestDelete.java
@@ -1422,6 +1422,62 @@ public abstract class TestDelete extends SparkRowLevelOperationsTestBase {
});
}
+ @TestTemplate
+ public void testDeleteToWapBranchCanDeleteWhereScansWapBranch() throws NoSuchTableException {
+ assumeThat(branch).as("WAP branch only works for table identifier without branch").isNull();
+
+ createAndInitPartitionedTable();
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = 'true')",
+ tableName, TableProperties.WRITE_AUDIT_PUBLISH_ENABLED);
+
+ append(tableName, new Employee(1, "hr"));
+
+ spark.conf().set(SparkSQLProperties.WAP_BRANCH, "wap");
+ try {
+ append(tableName, new Employee(0, "hr"), new Employee(1, "hr"), new Employee(2, "hr"));
+
+ sql("DELETE FROM %s WHERE id = 1", tableName);
+
+ assertThat(sql("SELECT id, dep FROM %s.branch_wap ORDER BY id", tableName))
+ .as("DELETE should remove the matching rows from the WAP branch")
+ .containsExactly(row(0, "hr"), row(2, "hr"));
+ assertThat(sql("SELECT id, dep FROM %s.branch_main", tableName))
+ .as("Main branch must not be modified by a WAP-targeted DELETE")
+ .containsExactly(row(1, "hr"));
+ } finally {
+ spark.conf().unset(SparkSQLProperties.WAP_BRANCH);
+ }
+ }
+
+ @TestTemplate
+ public void testMetadataDeleteToWapBranchCommitsToWapBranch() throws NoSuchTableException {
+ assumeThat(branch).as("WAP branch only works for table identifier without branch").isNull();
+
+ createAndInitPartitionedTable();
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = 'true')",
+ tableName, TableProperties.WRITE_AUDIT_PUBLISH_ENABLED);
+
+ append(tableName, new Employee(1, "hr"), new Employee(5, "eng"));
+
+ spark.conf().set(SparkSQLProperties.WAP_BRANCH, "wap");
+ try {
+ append(tableName, new Employee(0, "hr"), new Employee(2, "eng"));
+
+ sql("DELETE FROM %s WHERE dep = 'hr'", tableName);
+
+ assertThat(sql("SELECT id, dep FROM %s.branch_wap ORDER BY id", tableName))
+ .as("Metadata delete should remove the hr partition on the WAP branch")
+ .containsExactly(row(2, "eng"), row(5, "eng"));
+ assertThat(sql("SELECT id, dep FROM %s.branch_main ORDER BY id", tableName))
+ .as("Metadata delete must not commit to main when WAP is set")
+ .containsExactly(row(1, "hr"), row(5, "eng"));
+ } finally {
+ spark.conf().unset(SparkSQLProperties.WAP_BRANCH);
+ }
+ }
+
@TestTemplate
public void testDeleteWithFilterOnNestedColumn() {
createAndInitNestedColumnsTable();
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
index 6f0f992f1c..9e3c9a7e69 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
@@ -58,6 +58,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.spark.CommitMetadata;
import org.apache.iceberg.spark.Spark3Util;
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.SparkUtil;
@@ -376,11 +377,31 @@ public class SparkTable
}
}
- return canDeleteUsingMetadata(deleteExpr);
+ return canDeleteUsingMetadata(deleteExpr, scanBranchForDelete());
+ }
+
+ // Resolves the branch to scan during canDeleteWhere so it matches the branch deleteWhere
+ // will commit to. Falls back to main when WAP is configured but the WAP branch does not
+ // exist yet, since this is a read scan.
+ private String scanBranchForDelete() {
+ if (branch != null) {
+ return branch;
+ }
+
+ if (!SparkTableUtil.wapEnabled(table())) {
+ return null;
+ }
+
+ String wapBranch = sparkSession().conf().get(SparkSQLProperties.WAP_BRANCH, null);
+ if (wapBranch != null && table().refs().containsKey(wapBranch)) {
+ return wapBranch;
+ }
+
+ return null;
}
// a metadata delete is possible iff matching files can be deleted entirely
- private boolean canDeleteUsingMetadata(Expression deleteExpr) {
+ private boolean canDeleteUsingMetadata(Expression deleteExpr, String scanBranch) {
boolean caseSensitive = SparkUtil.caseSensitive(sparkSession());
if (ExpressionUtil.selectsPartitions(deleteExpr, table(), caseSensitive)) {
@@ -395,14 +416,14 @@ public class SparkTable
.includeColumnStats()
.ignoreResiduals();
- if (branch != null) {
- scan = scan.useRef(branch);
+ if (scanBranch != null) {
+ scan = scan.useRef(scanBranch);
}
try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
Map<Integer, Evaluator> evaluators = Maps.newHashMap();
StrictMetricsEvaluator metricsEvaluator =
- new StrictMetricsEvaluator(SnapshotUtil.schemaFor(table(), branch), deleteExpr);
+ new StrictMetricsEvaluator(SnapshotUtil.schemaFor(table(), scanBranch), deleteExpr);
return Iterables.all(
tasks,