This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 8aff800d525d [SPARK-60708][SQL] Fix DSv2 UPDATE and MERGE that
reference metadata columns of a table with CHAR/VARCHAR columns
8aff800d525d is described below
commit 8aff800d525d4669648e4e3ccda755c46ec87457
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Thu Oct 29 10:43:20 2026 -0700
[SPARK-60708][SQL] Fix DSv2 UPDATE and MERGE that reference metadata
columns of a table with CHAR/VARCHAR columns
### What changes were proposed in this pull request?
This PR fixes `ResolveRowLevelCommandAssignments` so that DSv2 `UPDATE` and
`MERGE` can reference metadata columns (e.g. `_partition`) of a target table
that has `CHAR` or `VARCHAR` columns.
When the rule aligns the assignments of a rewritable `UPDATE` or `MERGE`,
`cleanAttrMetadata` replaces the target `DataSourceV2Relation` with a copy
whose output has no `__CHAR_VARCHAR_TYPE_STRING` metadata. The copy is a new
instance, so its lazy `metadataOutput` creates the metadata attributes again
with new expression IDs. The condition and the assignments of the command still
refer to the metadata attributes of the original relation. However,
`RewriteUpdateTable` and `RewriteMerge [...]
This PR makes `cleanAttrMetadata` also return a map from the metadata
output of the original relation to that of the copy. The rule then rewrites the
references of the command with `QueryPlan.rewriteAttrs`, which also updates
outer references in subqueries. `ResolveSchemaEvolution.replaceWriteTarget`
uses the same approach for the output when it replaces the target relation of a
write. When there is nothing to clean, the relation is kept and the map is
empty.
Alternatives that were not taken:
- Keep the original relation: the cleaned output is what makes
`AssignmentUtils.aligned` accept the aligned assignments, because it compares
the assignment value type with the raw `CHAR`/`VARCHAR` type while the metadata
is present.
- Keep the original metadata attributes in the copy: each
`DataSourceV2Relation` instance derives its metadata output from the table, and
other copies such as `newInstance()` rely on getting new expression IDs.
- Resolve the referenced metadata columns by name in the rewrite rules: the
original relation is gone at that point, and a name alone cannot tell a target
metadata column from a metadata column of a DSv2 source.
### Why are the changes needed?
To fix a bug. For example, take a DSv2 table that supports row-level
operations, exposes the metadata column `_partition`, and has the schema `pk
INT NOT NULL, id INT, dep STRING, name VARCHAR(10)` (partitioned by `dep`):
```sql
UPDATE t SET id = -1 WHERE id IN (1, 2) AND _partition = 'hr'
```
```
[MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_APPEAR_IN_OPERATION] Resolved
attribute(s) "_partition" missing from "pk", "id", "dep", "name", "_partition",
"index" in operator !Filter (id#80 IN (1,2) AND (_partition#84 = hr)).
Attribute(s) with the same name appear in the operation: "_partition".
```
The same statements succeed when `name` is a `STRING` column. The failure
happens for group-based and delta-based tables (including split updates) in
these cases:
- `UPDATE` whose condition or assignment values reference a metadata column
- `MERGE` with `UPDATE` or `INSERT` actions whose clause conditions or
assignment values reference a target metadata column
- insert-only `MERGE` whose `ON` condition references a target metadata
column (`MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT`)
If the operation requests none of the referenced metadata columns, the
first two cases fail with
`MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT` instead.
`DELETE` and a `MERGE` with only `DELETE` actions are not affected because
Spark does not align them and keeps the original relation.
By code reading, the same code path exists since Apache Spark 3.5.0, where
`ResolveRowLevelCommandAssignments` started to align the assignments of
`UPDATE` (SPARK-42151) and `MERGE` (SPARK-43204) for tables that support
row-level operations.
### Does this PR introduce _any_ user-facing change?
Yes. On a DSv2 table with `CHAR`/`VARCHAR` columns, the `UPDATE` and
`MERGE` statements above no longer fail with `MISSING_ATTRIBUTES` and behave as
they do on a table without such columns. Statements that worked before are not
affected.
### How was this patch tested?
Pass the CIs with the newly added test cases.
- `UpdateTableSuiteBase`: `SPARK-60708: update char/varchar table with
metadata columns` and `SPARK-60708: update char/varchar table with an outer
metadata reference`
- `MergeIntoTableSuiteBase`: `SPARK-60708: merge into char/varchar table
with metadata columns` and `SPARK-60708: insert-only merge into char/varchar
table with metadata column in ON`
They run in the group-based, delta-based, split-update and no-metadata
suites (and the NoCodegen MERGE suites), and they fail without this fix. I also
checked manually that a `CHAR` column and a `MERGE` with multiple `NOT MATCHED`
clauses behave as they do on a table without `CHAR`/`VARCHAR` columns.
I also ran these suites locally, and all of them passed: the UPDATE, MERGE
(including schema evolution) and DELETE suites, `MetadataColumnSuite`,
`PlanResolutionSuite`, `AlignUpdateAssignmentsSuite`,
`AlignMergeAssignmentsSuite`, the row-level operation runtime filter suites,
`MergeIntoDataFrameSuite`, and the CHAR/VARCHAR test suites.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59918 from dongjoon-hyun/SPARK-60708.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit d443cc524904e47f0076923422f330cdf94ed4d9)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../ResolveRowLevelCommandAssignments.scala | 31 ++++++++---
.../sql/connector/MergeIntoTableSuiteBase.scala | 64 ++++++++++++++++++++++
.../spark/sql/connector/UpdateTableSuiteBase.scala | 44 +++++++++++++++
3 files changed, 131 insertions(+), 8 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
index a31ac092e279..229559d96325 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
@@ -17,8 +17,10 @@
package org.apache.spark.sql.catalyst.analysis
+import scala.collection.mutable
+
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, Cast}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap,
AttributeReference, Cast}
import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull
import org.apache.spark.sql.catalyst.plans.logical.{Assignment, DeleteAction,
InsertAction, LogicalPlan, MergeAction, MergeIntoTable, UpdateAction,
UpdateTable}
import org.apache.spark.sql.catalyst.rules.Rule
@@ -44,11 +46,11 @@ object ResolveRowLevelCommandAssignments extends
Rule[LogicalPlan] {
_.containsPattern(COMMAND), ruleId) {
case u: UpdateTable if !u.skipSchemaResolution && u.resolved &&
u.rewritable && !u.aligned =>
validateStoreAssignmentPolicy()
- val newTable = cleanAttrMetadata(u.table)
+ val (newTable, metadataAttrMap) = cleanAttrMetadata(u.table)
val newAssignments =
AssignmentUtils.alignUpdateAssignments(u.table.output, u.assignments,
fromStar = false, coerceNestedTypes = false,
ansiStoreAssignmentCastCheck = ansiStoreAssignmentCastCheck(u.table))
- u.copy(table = newTable, assignments = newAssignments)
+ u.copy(table = newTable, assignments =
newAssignments).rewriteAttrs(metadataAttrMap)
case u: UpdateTable if !u.skipSchemaResolution && u.resolved && !u.aligned
=>
resolveAssignments(u)
@@ -57,8 +59,9 @@ object ResolveRowLevelCommandAssignments extends
Rule[LogicalPlan] {
validateStoreAssignmentPolicy()
val coerceNestedTypes = conf.coerceMergeNestedTypes &&
m.withSchemaEvolution
val castCheck = ansiStoreAssignmentCastCheck(m.targetTable)
+ val (newTargetTable, metadataAttrMap) = cleanAttrMetadata(m.targetTable)
m.copy(
- targetTable = cleanAttrMetadata(m.targetTable),
+ targetTable = newTargetTable,
matchedActions = alignActions(
m.targetTable.output,
m.matchedActions,
@@ -73,7 +76,7 @@ object ResolveRowLevelCommandAssignments extends
Rule[LogicalPlan] {
m.targetTable.output,
m.notMatchedBySourceActions,
coerceNestedTypes,
- castCheck))
+ castCheck)).rewriteAttrs(metadataAttrMap)
case m: MergeIntoTable if shouldAlignAssignments(m) &&
containsFinalSchema(m) =>
resolveAssignments(m)
@@ -101,11 +104,23 @@ object ResolveRowLevelCommandAssignments extends
Rule[LogicalPlan] {
}
}
- private def cleanAttrMetadata(table: LogicalPlan): LogicalPlan = {
- table.transform {
+ // A cleaned relation is a new instance whose metadata output has new
expression IDs, so this
+ // also returns a map from the metadata output of the original relation,
which the command may
+ // reference, to the new one. A relation with nothing to clean is kept as is.
+ private def cleanAttrMetadata(table: LogicalPlan): (LogicalPlan,
AttributeMap[Attribute]) = {
+ val metadataAttrMapping = mutable.ArrayBuffer.empty[(Attribute, Attribute)]
+ val newTable = table.transform {
case r: DataSourceV2Relation =>
- r.copy(output = r.output.map(CharVarcharUtils.cleanAttrMetadata))
+ val newOutput = r.output.map(CharVarcharUtils.cleanAttrMetadata)
+ if (newOutput == r.output) {
+ r
+ } else {
+ val newRelation = r.copy(output = newOutput)
+ metadataAttrMapping ++=
r.metadataOutput.zip(newRelation.metadataOutput)
+ newRelation
+ }
}
+ (newTable, AttributeMap(metadataAttrMapping.toSeq))
}
private def resolveAssignments(p: LogicalPlan): LogicalPlan = {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
index 37983b75faf4..ea5d01c77180 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
@@ -2030,6 +2030,70 @@ abstract class MergeIntoTableSuiteBase extends
RowLevelOperationSuiteBase
}
}
+ test("SPARK-60708: merge into char/varchar table with metadata columns") {
+ withTempView("source") {
+ createTable("pk INT NOT NULL, id INT, dep STRING, name VARCHAR(10)")
+
+ append("pk INT NOT NULL, id INT, dep STRING, name STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr", "name": "a" }
+ |{ "pk": 2, "id": 2, "dep": "software", "name": "b" }
+ |{ "pk": 3, "id": 3, "dep": "hr", "name": "c" }
+ |{ "pk": 4, "id": 4, "dep": "software", "name": "d" }
+ |""".stripMargin)
+
+ Seq(1, 2, 5).toDF("pk").createOrReplaceTempView("source")
+
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED AND t._partition = 'hr' THEN
+ | UPDATE SET t.id = t.index + 10
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, id, dep, name) VALUES (s.pk, 0, 'hr', 'e')
+ |WHEN NOT MATCHED BY SOURCE AND t._partition = 'software' THEN
+ | DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, 10, "hr", "a"), // update (matched)
+ Row(2, 2, "software", "b"), // unchanged
+ Row(3, 3, "hr", "c"), // unchanged
+ Row(5, 0, "hr", "e"))) // insert
+ }
+ }
+
+ test("SPARK-60708: insert-only merge into char/varchar table with metadata
column in ON") {
+ withTempView("source") {
+ createTable("pk INT NOT NULL, id INT, dep STRING, name VARCHAR(10)")
+
+ append("pk INT NOT NULL, id INT, dep STRING, name STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr", "name": "a" }
+ |{ "pk": 2, "id": 2, "dep": "software", "name": "b" }
+ |""".stripMargin)
+
+ Seq(1, 2, 3).toDF("pk").createOrReplaceTempView("source")
+
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk AND t._partition = 'hr'
+ |WHEN NOT MATCHED THEN
+ | INSERT (pk, id, dep, name) VALUES (s.pk, 0, 'hr', 'new')
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, 1, "hr", "a"), // unchanged (matched)
+ Row(2, 2, "software", "b"), // unchanged
+ Row(2, 0, "hr", "new"), // insert (not matched in hr)
+ Row(3, 0, "hr", "new"))) // insert
+ }
+ }
+
test("merge with NOT NULL checks") {
withTempView("source") {
createAndInitTable("pk INT NOT NULL, s STRUCT<n_i: INT NOT NULL, n_l:
LONG>, dep STRING",
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
index c2b6c1f7ec94..f52ff2406755 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
@@ -915,6 +915,50 @@ abstract class UpdateTableSuiteBase extends
RowLevelOperationSuiteBase {
checkUpdateMetrics(numUpdatedRows = 2, numCopiedRows = 0)
}
+ test("SPARK-60708: update char/varchar table with metadata columns") {
+ createTable("pk INT NOT NULL, id INT, dep STRING, name VARCHAR(10)")
+
+ append("pk INT NOT NULL, id INT, dep STRING, name STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr", "name": "a" }
+ |{ "pk": 2, "id": 2, "dep": "software", "name": "b" }
+ |{ "pk": 3, "id": 3, "dep": "hr", "name": "c" }
+ |""".stripMargin)
+
+ sql(s"UPDATE $tableNameAsString SET id = -1 WHERE id IN (1, 2) AND
_partition = 'hr'")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr", "a") :: Row(2, 2, "software", "b") :: Row(3, 3, "hr",
"c") :: Nil)
+ checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 1)
+
+ sql(s"UPDATE $tableNameAsString t SET t.id = t.index + 10 WHERE t.pk = 2")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr", "a") :: Row(2, 10, "software", "b") :: Row(3, 3, "hr",
"c") :: Nil)
+ checkUpdateMetrics(numUpdatedRows = 1, numCopiedRows = 0)
+ }
+
+ test("SPARK-60708: update char/varchar table with an outer metadata
reference") {
+ createTable("pk INT NOT NULL, id INT, dep STRING, name VARCHAR(10)")
+
+ append("pk INT NOT NULL, id INT, dep STRING, name STRING",
+ """{ "pk": 1, "id": 1, "dep": "hr", "name": "a" }
+ |{ "pk": 2, "id": 2, "dep": "software", "name": "b" }
+ |{ "pk": 3, "id": 3, "dep": "hr", "name": "c" }
+ |""".stripMargin)
+
+ withTempView("deps") {
+ Seq("hr").toDF("d").createOrReplaceTempView("deps")
+ sql(s"UPDATE $tableNameAsString t SET id = -1 " +
+ "WHERE id IN (1, 2) AND EXISTS (SELECT 1 FROM deps WHERE deps.d =
t._partition)")
+ }
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr", "a") :: Row(2, 2, "software", "b") :: Row(3, 3, "hr",
"c") :: Nil)
+ }
+
test("update with NOT NULL checks") {
createAndInitTable("pk INT NOT NULL, s STRUCT<n_i: INT NOT NULL, n_l:
LONG>, dep STRING",
"""{ "pk": 1, "s": { "n_i": 1, "n_l": 11 }, "dep": "hr" }
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
