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 b4e556950a [spark] Support V1 UPDATE for data evolution tables (#8389)
b4e556950a is described below
commit b4e556950a5746000c0204657cd578aa16ad4704
Author: Kerwin Zhang <[email protected]>
AuthorDate: Tue Jun 30 17:24:06 2026 +0800
[spark] Support V1 UPDATE for data evolution tables (#8389)
---
.../catalyst/analysis/PaimonUpdateTable.scala | 44 ++++++++---
.../UpdatePaimonDataEvolutionTableCommand.scala | 92 ++++++++++++++++++++++
.../paimon/spark/sql/RowTrackingTestBase.scala | 61 +++++++++++---
3 files changed, 176 insertions(+), 21 deletions(-)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonUpdateTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonUpdateTable.scala
index e03e658141..ab1761bbff 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonUpdateTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonUpdateTable.scala
@@ -19,7 +19,7 @@
package org.apache.paimon.spark.catalyst.analysis
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
-import org.apache.paimon.spark.commands.UpdatePaimonTableCommand
+import
org.apache.paimon.spark.commands.{UpdatePaimonDataEvolutionTableCommand,
UpdatePaimonTableCommand}
import org.apache.paimon.table.FileStoreTable
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
@@ -49,11 +49,6 @@ object PaimonUpdateTable extends Rule[LogicalPlan] with
RowLevelHelper with Expr
throw new RuntimeException("Can't update the primary key
column.")
}
- if (paimonTable.coreOptions().dataEvolutionEnabled()) {
- throw new RuntimeException(
- "Update operation is not supported when data evolution is
enabled yet.")
- }
-
// Align against `u.table.output`: for CHAR/VARCHAR columns the
analyzer adds a
// `readSidePadding` Project whose output has different exprIds
than `relation`, and
// the parsed assignment keys reference the Project's
attributes. Order matches
@@ -66,15 +61,42 @@ object PaimonUpdateTable extends Rule[LogicalPlan] with
RowLevelHelper with Expr
val alignedExpressions =
alignedAssignments.map(_.value).zip(relation.output)
val alignedUpdateTable = u.copy(assignments = alignedAssignments)
+ val dataEvolutionEnabled =
paimonTable.coreOptions().dataEvolutionEnabled()
+
+ if (dataEvolutionEnabled) {
+ // The rewritten files keep the original row ids, which are
derived from the
+ // file's firstRowId per partition; moving a row to another
partition would need
+ // re-assigned row ids (delete + insert semantics).
+ val partitionKeys = paimonTable.partitionKeys().asScala.toSeq
+ if (!validUpdateAssignment(u.table.outputSet, partitionKeys,
assignments)) {
+ throw new RuntimeException(
+ "Update to partition columns is not supported for data
evolution tables.")
+ }
+ }
if (!shouldFallbackToV1Update(table, alignedUpdateTable)) {
+ if (dataEvolutionEnabled) {
+ // Data-evolution tables do not currently expose Spark V2
row-level operations.
+ // Keep this guard in case capability rules change; this
branch intentionally
+ // implements only V1 data-evolution UPDATE.
+ throw new RuntimeException(
+ "Update operation is not supported when data evolution is
enabled yet.")
+ }
alignedUpdateTable
} else {
- UpdatePaimonTableCommand(
- relation,
- paimonTable,
- condition.getOrElse(TrueLiteral),
- alignedExpressions)
+ if (dataEvolutionEnabled) {
+ UpdatePaimonDataEvolutionTableCommand(
+ relation,
+ table,
+ condition.getOrElse(TrueLiteral),
+ alignedExpressions)
+ } else {
+ UpdatePaimonTableCommand(
+ relation,
+ paimonTable,
+ condition.getOrElse(TrueLiteral),
+ alignedExpressions)
+ }
}
case _ =>
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
new file mode 100644
index 0000000000..9e2c35e3e4
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonDataEvolutionTableCommand.scala
@@ -0,0 +1,92 @@
+/*
+ * 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.paimon.spark.commands
+
+import org.apache.paimon.spark.SparkTable
+import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.ROW_ID_COLUMN
+
+import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference, EqualTo, Expression}
+import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
+import org.apache.spark.sql.catalyst.plans.logical.{Assignment, Filter,
Project, SupportsSubquery}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.paimon.shims.SparkShimLoader
+
+/** V1 UPDATE command for data-evolution tables, implemented through
partial-column MERGE. */
+case class UpdatePaimonDataEvolutionTableCommand(
+ relation: DataSourceV2Relation,
+ v2Table: SparkTable,
+ condition: Expression,
+ alignedExpressions: Seq[(Expression, Attribute)])
+ extends PaimonLeafRunnableCommand
+ with SupportsSubquery {
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ val targetRowId = rowIdAttribute(relation)
+ val sourceTable = updatedRowIdSource(targetRowId)
+ val sourceRowId = sourceTable.output.head.asInstanceOf[AttributeReference]
+
+ val matchedCondition = EqualTo(targetRowId, sourceRowId)
+ val updateAction = SparkShimLoader.shim.createUpdateAction(
+ None,
+ alignedExpressions.map { case (expression, attribute) =>
Assignment(attribute, expression) })
+
+ MergeIntoPaimonDataEvolutionTable(
+ v2Table,
+ relation,
+ sourceTable,
+ matchedCondition,
+ Seq(updateAction),
+ Nil,
+ Nil).run(sparkSession)
+ }
+
+ private def updatedRowIdSource(targetRowId: AttributeReference): Project = {
+ val conditionReferences = condition.references.toSeq.collect {
+ case attr: AttributeReference => attr
+ }
+ val readOutput = deduplicateByExprId(conditionReferences :+ targetRowId)
+ val sourceScan =
+ SparkShimLoader.shim.copyDataSourceV2Relation(relation, v2Table,
readOutput)
+ // Keep the Filter visible for conditional UPDATEs. The data-evolution
MERGE command uses a
+ // self-merge shortcut for Project(PaimonRelation); if a WHERE update were
shaped that way, the
+ // shortcut would bypass the source join path and update every row.
+ val filteredSource = if (condition == TrueLiteral) sourceScan else
Filter(condition, sourceScan)
+
+ Project(Seq(Alias(targetRowId, ROW_ID_COLUMN)()), filteredSource)
+ }
+
+ private def rowIdAttribute(relation: DataSourceV2Relation):
AttributeReference = {
+ (relation.output ++ relation.metadataOutput)
+ .collectFirst {
+ case attr: AttributeReference if attr.name == ROW_ID_COLUMN => attr
+ }
+ .getOrElse(throw new RuntimeException(
+ s"Cannot find $ROW_ID_COLUMN metadata column for data-evolution
UPDATE."))
+ }
+
+ private def deduplicateByExprId(attributes: Seq[AttributeReference]):
Seq[AttributeReference] = {
+ attributes
+ .foldLeft(Seq.empty[AttributeReference]) {
+ case (deduplicated, attr) if deduplicated.exists(_.exprId ==
attr.exprId) => deduplicated
+ case (deduplicated, attr) => deduplicated :+ attr
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
index f53089470d..c7f6ea86e3 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala
@@ -1011,16 +1011,57 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
- test("Data Evolution: update table throws exception") {
- withTable("t") {
- sql(
- "CREATE TABLE t (id INT, b INT, c INT) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
- sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(2, 4)")
- assert(
- intercept[RuntimeException] {
- sql("UPDATE t SET b = 22")
- }.getMessage
- .contains("Update operation is not supported when data evolution is
enabled yet."))
+ test("Data Evolution: V1 update table with data-evolution") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, b INT, c INT) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(2, 4)")
+
+ sql("UPDATE t SET b = 22 WHERE id = 2")
+ checkAnswer(
+ sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id"),
+ Seq(Row(2, 22, 2, 0, 2), Row(3, 3, 3, 1, 2))
+ )
+ }
+ }
+ }
+
+ test("Data Evolution: V1 update table with data-evolution without
condition") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, b INT, c INT) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
+ sql("INSERT INTO t SELECT /*+ REPARTITION(1) */ id, id AS b, id AS c
FROM range(2, 4)")
+
+ sql("UPDATE t SET b = 22")
+ checkAnswer(
+ sql("SELECT *, _ROW_ID, _SEQUENCE_NUMBER FROM t ORDER BY id"),
+ Seq(Row(2, 22, 2, 0, 2), Row(3, 22, 3, 1, 2))
+ )
+ }
+ }
+ }
+
+ test("Data Evolution: V1 update partition column throws exception") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
+ withTable("t") {
+ sql("""
+ |CREATE TABLE t (id INT, b INT, dt STRING)
+ |PARTITIONED BY (dt)
+ |TBLPROPERTIES ('row-tracking.enabled' = 'true',
'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+ sql("INSERT INTO t VALUES (1, 1, 'p1'), (2, 2, 'p2')")
+
+ assert(
+ intercept[RuntimeException] {
+ sql("UPDATE t SET dt = 'p3' WHERE id = 1")
+ }.getMessage
+ .contains("Update to partition columns is not supported for data
evolution tables."))
+
+ sql("UPDATE t SET b = 10 WHERE id = 1")
+ checkAnswer(sql("SELECT * FROM t ORDER BY id"), Seq(Row(1, 10, "p1"),
Row(2, 2, "p2")))
+ }
}
}
