This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 0db3f89bca [spark] Fix MERGE INTO delete clause corrupting
partial-update tables (#9356)
0db3f89bca is described below
commit 0db3f89bca743183027cc812c3c3f88b430e9254
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Sun Aug 23 23:19:30 2026 +0800
[spark] Fix MERGE INTO delete clause corrupting partial-update tables
(#9356)
---
.../spark/catalyst/analysis/PaimonMergeInto.scala | 32 ++++++++
.../paimon/spark/sql/MergeIntoTableTestBase.scala | 89 ++++++++++++++++++++++
2 files changed, 121 insertions(+)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeInto.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeInto.scala
index be8424bf9b..cd159dce8d 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeInto.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonMergeInto.scala
@@ -21,6 +21,8 @@ package org.apache.paimon.spark.catalyst.analysis
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
import org.apache.paimon.spark.commands.{MergeIntoPaimonDataEvolutionTable,
MergeIntoPaimonTable}
+import org.apache.paimon.table.PrimaryKeyTableUtils.validatePKUpsertDeletable
+import org.apache.paimon.table.Table
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet,
Expression, SubqueryExpression}
@@ -51,6 +53,7 @@ case class PaimonMergeInto(spark: SparkSession)
var v2Table = relation.table.asInstanceOf[SparkTable]
checkPaimonTable(v2Table.getTable)
+ checkDeleteActionValidity(v2Table.getTable, merge)
checkCondition(merge.mergeCondition)
(merge.matchedActions ++ merge.notMatchedActions)
.flatMap(_.condition)
@@ -123,6 +126,35 @@ case class PaimonMergeInto(spark: SparkSession)
}
}
+ /**
+ * A `WHEN MATCHED ... THEN DELETE` or `WHEN NOT MATCHED BY SOURCE ... THEN
DELETE` clause emits
+ * real [[org.apache.paimon.types.RowKind.DELETE]] records into the LSM tree
(see
+ * `MergeIntoPaimonTable.constructChangedRows`), so the target's merge
engine has to be able to
+ * consume them.
+ *
+ * Without this check a merge engine that rejects delete records --
`partial-update` without
+ * `partial-update.remove-record-on-delete` -- accepts the write and commits
a snapshot that later
+ * makes the table unreadable: the failure only surfaces much later, in
+ * `PartialUpdateMergeFunction#add`, at *read* time.
+ *
+ * This mirrors the validation Flink runs for SQL `DELETE`
+ * (`SupportsRowLevelOperationFlinkTableSink#applyRowLevelDelete`) and the
one
+ * `DeleteFromPaimonTableCommand` already runs, so all three paths reject
the same tables with the
+ * same message.
+ */
+ private def checkDeleteActionValidity(table: Table, merge: MergeIntoTable):
Unit = {
+ // Tables without primary keys rewrite the touched files (or maintain
deletion vectors) instead
+ // of emitting delete records, so no merge engine is involved.
+ if (table.primaryKeys().isEmpty) {
+ return
+ }
+ val deletesRows = (merge.matchedActions ++
resolveNotMatchedBySourceActions(merge))
+ .exists(_.isInstanceOf[DeleteAction])
+ if (deletesRows) {
+ validatePKUpsertDeletable(table)
+ }
+ }
+
private def checkUpdateActionValidity(
targetOutput: AttributeSet,
mergeCondition: Expression,
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoTableTestBase.scala
index 05804740a2..78039e6083 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoTableTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MergeIntoTableTestBase.scala
@@ -1565,6 +1565,95 @@ trait MergeIntoPrimaryKeyTableTest extends
PaimonSparkTestBase with PaimonPrimar
Row(1, 10, "c111") :: Row(2, 20, "c2") :: Row(103, 30, "c333") :: Nil)
}
}
+
+ /**
+ * A DELETE clause emits real `RowKind.DELETE` records, so the target's
merge engine must be able
+ * to consume them. `partial-update` without
`partial-update.remove-record-on-delete` cannot:
+ * before the check in `PaimonMergeInto`, the merge committed happily and
the table only blew up
+ * later at read time in `PartialUpdateMergeFunction#add`.
+ */
+ test("Paimon MergeInto: reject DELETE clause when the merge engine can not
accept deletes") {
+ val deleteClauses =
+ Seq("WHEN MATCHED THEN DELETE", "WHEN MATCHED AND target.b > 0 THEN
DELETE") ++
+ // `WHEN NOT MATCHED BY SOURCE` was only added to Spark's parser in
3.4; on 3.2/3.3 the
+ // statement fails to parse before it ever reaches the analysis rule
under test.
+ (if (gteqSpark3_4) Seq("WHEN NOT MATCHED BY SOURCE THEN DELETE") else
Nil)
+
+ deleteClauses.foreach {
+ deleteClause =>
+ withTable("source", "target") {
+ Seq((1, 100, "c11")).toDF("a", "b",
"c").createOrReplaceTempView("source")
+ createTable(
+ "target",
+ "a INT, b INT, c STRING",
+ Seq("a"),
+ extraProps = Map("merge-engine" -> "partial-update"))
+ spark.sql("INSERT INTO target values (1, 10, 'c1'), (2, 20, 'c2')")
+
+ val error = intercept[UnsupportedOperationException] {
+ spark.sql(s"""
+ |MERGE INTO target
+ |USING source
+ |ON target.a = source.a
+ |$deleteClause
+ |""".stripMargin)
+ }.getMessage
+ assert(error.contains("partial-update.remove-record-on-delete"),
error)
+
+ // The statement was rejected before anything was committed.
+ checkAnswer(
+ spark.sql("SELECT * FROM target ORDER BY a"),
+ Row(1, 10, "c1") :: Row(2, 20, "c2") :: Nil)
+ }
+ }
+ }
+
+ test("Paimon MergeInto: allow DELETE clause once the merge engine can accept
deletes") {
+ withTable("source", "target") {
+ Seq((1, 100, "c11")).toDF("a", "b",
"c").createOrReplaceTempView("source")
+ createTable(
+ "target",
+ "a INT, b INT, c STRING",
+ Seq("a"),
+ extraProps = Map(
+ "merge-engine" -> "partial-update",
+ "partial-update.remove-record-on-delete" -> "true"))
+ spark.sql("INSERT INTO target values (1, 10, 'c1'), (2, 20, 'c2')")
+
+ spark.sql(s"""
+ |MERGE INTO target
+ |USING source
+ |ON target.a = source.a
+ |WHEN MATCHED THEN DELETE
+ |""".stripMargin)
+
+ checkAnswer(spark.sql("SELECT * FROM target ORDER BY a"), Row(2, 20,
"c2") :: Nil)
+ }
+ }
+
+ test("Paimon MergeInto: a delete-free merge on partial-update is
unaffected") {
+ withTable("source", "target") {
+ Seq((1, 100, "c11"), (3, 300, "c33")).toDF("a", "b",
"c").createOrReplaceTempView("source")
+ createTable(
+ "target",
+ "a INT, b INT, c STRING",
+ Seq("a"),
+ extraProps = Map("merge-engine" -> "partial-update"))
+ spark.sql("INSERT INTO target values (1, 10, 'c1'), (2, 20, 'c2')")
+
+ spark.sql(s"""
+ |MERGE INTO target
+ |USING source
+ |ON target.a = source.a
+ |WHEN MATCHED THEN UPDATE SET b = source.b
+ |WHEN NOT MATCHED THEN INSERT *
+ |""".stripMargin)
+
+ checkAnswer(
+ spark.sql("SELECT * FROM target ORDER BY a"),
+ Row(1, 100, "c1") :: Row(2, 20, "c2") :: Row(3, 300, "c33") :: Nil)
+ }
+ }
}
trait MergeIntoAppendTableTest extends PaimonSparkTestBase with
PaimonAppendTable {
