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 8d790f178a [spark] Resolve V1 UPDATE expressions against the relation
on tables with CHAR columns (#10061)
8d790f178a is described below
commit 8d790f178ab3dc141bdf09b2ad645b06f9f7b412
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Tue Sep 22 10:43:38 2026 +0800
[spark] Resolve V1 UPDATE expressions against the relation on tables with
CHAR columns (#10061)
---
.../catalyst/analysis/PaimonUpdateTable.scala | 53 ++++++++++++++++++----
.../paimon/spark/sql/RowTrackingTestBase.scala | 28 ++++++++++++
.../paimon/spark/sql/UpdateTableTestBase.scala | 52 +++++++++++++++++++++
3 files changed, 125 insertions(+), 8 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 ab1761bbff..d0ec9f80a4 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
@@ -22,6 +22,7 @@ import
org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
import
org.apache.paimon.spark.commands.{UpdatePaimonDataEvolutionTableCommand,
UpdatePaimonTableCommand}
import org.apache.paimon.table.FileStoreTable
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, Expression, ExprId, OuterReference, SubqueryExpression}
import org.apache.spark.sql.catalyst.expressions.Literal.TrueLiteral
import org.apache.spark.sql.catalyst.plans.logical.{AnalysisHelper,
LogicalPlan, UpdateTable}
import org.apache.spark.sql.catalyst.rules.Rule
@@ -58,9 +59,17 @@ object PaimonUpdateTable extends Rule[LogicalPlan] with
RowLevelHelper with Expr
assignments,
fromStar = false,
mergeSchemaEnabled = false)
- val alignedExpressions =
alignedAssignments.map(_.value).zip(relation.output)
-
val alignedUpdateTable = u.copy(assignments = alignedAssignments)
+
+ // The V1 commands plan on `relation` directly, without that
Project, so the aligned
+ // values and the condition must reference `relation.output`: an
attribute of the
+ // padding Project (an untouched CHAR column, or a CHAR column
read by the condition
+ // or an assignment value) would otherwise be unresolvable
there. Reads still go
+ // through read-side padding, which the analyzer re-applies on
top of `relation`.
+ val toRelationAttributes =
relationAttributeRewriter(u.table.output, relation.output)
+ val alignedExpressions =
+ alignedAssignments.map(a =>
toRelationAttributes(a.value)).zip(relation.output)
+ val v1Condition =
toRelationAttributes(condition.getOrElse(TrueLiteral))
val dataEvolutionEnabled =
paimonTable.coreOptions().dataEvolutionEnabled()
if (dataEvolutionEnabled) {
@@ -88,14 +97,10 @@ object PaimonUpdateTable extends Rule[LogicalPlan] with
RowLevelHelper with Expr
UpdatePaimonDataEvolutionTableCommand(
relation,
table,
- condition.getOrElse(TrueLiteral),
+ v1Condition,
alignedExpressions)
} else {
- UpdatePaimonTableCommand(
- relation,
- paimonTable,
- condition.getOrElse(TrueLiteral),
- alignedExpressions)
+ UpdatePaimonTableCommand(relation, paimonTable, v1Condition,
alignedExpressions)
}
}
@@ -105,4 +110,36 @@ object PaimonUpdateTable extends Rule[LogicalPlan] with
RowLevelHelper with Expr
}
}
}
+
+ /**
+ * Rewrites references to the attributes of `from` into the positionally
matching attributes of
+ * `to`. Both come from the same table, so they line up 1:1; attributes that
already share an
+ * exprId (every non-CHAR column) are left alone.
+ */
+ private def relationAttributeRewriter(
+ from: Seq[Attribute],
+ to: Seq[Attribute]): Expression => Expression = {
+ require(
+ from.size == to.size,
+ s"UPDATE table output ${from.mkString(", ")} does not line up with
relation output " +
+ s"${to.mkString(", ")}.")
+ val mapping: Map[ExprId, Attribute] = from
+ .zip(to)
+ .collect { case (f, t) if f.exprId != t.exprId => f.exprId -> t }
+ .toMap
+ if (mapping.isEmpty) {
+ identity
+ } else {
+ expression =>
+ expression.transform {
+ case attr: AttributeReference if mapping.contains(attr.exprId) =>
mapping(attr.exprId)
+ case subquery: SubqueryExpression =>
+ // A correlated subquery refers to the outer row through
OuterReference.
+ subquery.withNewPlan(subquery.plan.transformAllExpressions {
+ case OuterReference(attr) if mapping.contains(attr.exprId) =>
+ OuterReference(mapping(attr.exprId))
+ })
+ }
+ }
+ }
}
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 dc31d88092..f3ba6808be 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
@@ -1983,6 +1983,34 @@ abstract class RowTrackingTestBase extends
PaimonSparkTestBase with AdaptiveSpar
}
}
+ test("Data Evolution: V1 update on a table with a CHAR column") {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
+ withTable("t") {
+ sql(
+ "CREATE TABLE t (id INT, c CHAR(3), b INT) TBLPROPERTIES
('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
+ sql("INSERT INTO t VALUES (1, 'a', 10), (2, 'b', 20)")
+
+ // The untouched CHAR column must not be treated as an updated column:
only `b` is
+ // written to the new column-group file.
+ sql("UPDATE t SET b = 0 WHERE id = 1")
+ checkAnswer(
+ sql("SELECT id, c, b FROM t ORDER BY id"),
+ Seq(Row(1, "a ", 0), Row(2, "b ", 20)))
+ checkAnswer(
+ sql("SELECT write_cols FROM `t$files` ORDER BY max_sequence_number
DESC LIMIT 1"),
+ Seq(Row(Seq("b"))))
+
+ // The condition and an assignment value read the CHAR column, padded
like a SELECT.
+ val (mergeRowsPlans, _) =
+ executeMergeIntoAndCollectPlans("UPDATE t SET b = length(c) WHERE c
= 'b'")
+ assertSelfMergeShortcut(mergeRowsPlans)
+ checkAnswer(
+ sql("SELECT id, c, b FROM t ORDER BY id"),
+ Seq(Row(1, "a ", 0), Row(2, "b ", 3)))
+ }
+ }
+ }
+
test("Data Evolution: V1 update with subquery condition keeps the source
filter") {
withSparkSQLConf("spark.paimon.write.use-v2-write" -> "false") {
withTable("t", "s") {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/UpdateTableTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/UpdateTableTestBase.scala
index 665a08017a..8c58b80d31 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/UpdateTableTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/UpdateTableTestBase.scala
@@ -363,6 +363,58 @@ abstract class UpdateTableTestBase extends
PaimonSparkTestBase {
checkAnswer(sql("SELECT * FROM T"), Seq(Row(1, "s", "b")))
}
+ test("Paimon update: CHAR column that is read or left untouched") {
+ Seq(
+ "",
+ "TBLPROPERTIES ('deletion-vectors.enabled' = 'true')",
+ "TBLPROPERTIES ('primary-key' = 'id', 'bucket' = '1')"
+ ).foreach {
+ props =>
+ withTable("t", "s") {
+ sql(s"CREATE TABLE t (id INT, c CHAR(3), b INT) $props")
+ sql("INSERT INTO t VALUES (1, 'a', 10), (2, 'b', 20)")
+
+ // The CHAR column is neither assigned nor read.
+ sql("UPDATE t SET b = 0 WHERE id = 1")
+ checkAnswer(
+ sql("SELECT id, c, b FROM t ORDER BY id"),
+ Seq(Row(1, "a ", 0), Row(2, "b ", 20)))
+
+ // The condition reads it: the literal is padded the same way as the
column.
+ sql("UPDATE t SET b = 1 WHERE c = 'a'")
+ checkAnswer(
+ sql("SELECT id, c, b FROM t ORDER BY id"),
+ Seq(Row(1, "a ", 1), Row(2, "b ", 20)))
+
+ // An assignment value reads it and sees the padded value, like a
SELECT does.
+ sql("UPDATE t SET b = length(c) WHERE id = 2")
+ checkAnswer(
+ sql("SELECT id, c, b FROM t ORDER BY id"),
+ Seq(Row(1, "a ", 1), Row(2, "b ", 3)))
+
+ // A correlated subquery reads it through an outer reference. Only
on the primary-key
+ // table, whose UPDATE keeps the condition in a Filter: the non-pk
UPDATE moves it into a
+ // Project, and correlated subqueries there depend on the Spark
version.
+ if (props.contains("primary-key")) {
+ sql("CREATE TABLE s (k CHAR(3))")
+ sql("INSERT INTO s VALUES ('b')")
+ sql("UPDATE t SET b = 7 WHERE EXISTS (SELECT 1 FROM s WHERE s.k =
t.c)")
+ } else {
+ sql("UPDATE t SET b = 7 WHERE id = 2")
+ }
+ checkAnswer(
+ sql("SELECT id, c, b FROM t ORDER BY id"),
+ Seq(Row(1, "a ", 1), Row(2, "b ", 7)))
+
+ // No condition at all.
+ sql("UPDATE t SET b = 5")
+ checkAnswer(
+ sql("SELECT id, c, b FROM t ORDER BY id"),
+ Seq(Row(1, "a ", 5), Row(2, "b ", 5)))
+ }
+ }
+ }
+
test("Paimon update: overlong CHAR value throws (same as INSERT)") {
withTable("t_char") {
sql("CREATE TABLE t_char (id INT, c CHAR(2))")
