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 2b5f4c080168 [SPARK-60664][SQL] Remap metadata columns in runtime
group filters of group-based UPDATE with subqueries
2b5f4c080168 is described below
commit 2b5f4c080168ad676c291c10dfa2570163659a8d
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 28 20:42:02 2026 -0700
[SPARK-60664][SQL] Remap metadata columns in runtime group filters of
group-based UPDATE with subqueries
### What changes were proposed in this pull request?
This PR aims to fix `RowLevelOperationRuntimeGroupFiltering` to remap the
metadata columns that the UPDATE condition references, in addition to the table
columns, when it builds a runtime group filter.
A group-based UPDATE whose condition has a subquery is rewritten as a
`Union` that reads the table twice
(`RewriteUpdateTable.buildReplaceDataWithUnionPlan`), and
`DeduplicateRelations` gives the second read new expression IDs, including its
metadata attributes. The group filter of each read is the UPDATE condition with
the attributes replaced by name with those of that read. However, only
`write.table.output` (the table columns) has been replaced. When the analyzer
resolves a metadat [...]
For example, `EXPLAIN EXTENDED` shows the following for the second read.
`_partition#30` belongs to the first read.
```
+- Filter dynamicpruningexpression(dep#45 IN (list#177 []))
: +- Aggregate [dep#174], [dep#174]
: +- Project [dep#174]
: +- Join LeftSemi, (pk#172 = pk#53)
: :- !Filter (_partition#30 = hr)
: : +- RelationV2[pk#172, dep#174] cat.ns1.test_table
: +- Project [value#52 AS pk#53]
: +- LocalRelation [value#52]
+- RelationV2[pk#43, salary#44, dep#45, _partition#46, index#47]
cat.ns1.test_table
```
With this PR, the group filter uses the metadata attribute of its own read.
```
+- Filter dynamicpruningexpression(dep#352 IN (list#484 []))
: +- Aggregate [dep#481], [dep#481]
: +- Project [dep#481]
: +- Join LeftSemi, (pk#479 = pk#360)
: :- Project [pk#479, dep#481]
: : +- Filter (_partition#482 = hr)
: : +- RelationV2[pk#479, dep#481, _partition#482]
cat.ns1.test_table
: +- Project [value#359 AS pk#360]
: +- LocalRelation [value#359]
+- RelationV2[pk#350, salary#351, dep#352, _partition#353, index#354]
cat.ns1.test_table
```
The table columns are mapped as before, including the
`_LEGACY_ERROR_TEMP_3075` check.
### Why are the changes needed?
The dangling reference makes such an UPDATE fail at execution with an
internal error. The scan of the group filter subquery prunes `_partition`, and
the post-scan `Filter` cannot bind the reference.
```sql
-- `cat.ns1.test_table` is a group-based table partitioned by `dep`
UPDATE cat.ns1.test_table
SET salary = -1
WHERE _partition = 'hr' AND pk IN (SELECT pk FROM updated_pk)
```
```
org.apache.spark.SparkException: [INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND] Could
not find _partition#26 in [pk#168,dep#170]. SQLSTATE: XX000
```
This happens when the UPDATE is resolved in a single analyzer iteration,
e.g., when no implicit cast is required. Otherwise, `AddMetadataColumns` adds
the metadata columns to the target relation in a later iteration, and they are
mapped as table columns.
### Does this PR introduce _any_ user-facing change?
Yes. Such UPDATE statements fail with an internal error without this PR.
The UPDATE branch of this rule has mapped only the table columns since
SPARK-43975 (Apache Spark 3.5.0).
### How was this patch tested?
Pass the CIs with the newly added test case. Without this fix, the test
fails like the following.
```
$ build/sbt 'sql/testOnly *GroupBasedUpdateTableSuite -- -z SPARK-60664'
...
[info] - SPARK-60664: update runtime group filtering with metadata column
and subquery *** FAILED *** (2 seconds, 467 milliseconds)
[info] org.apache.spark.SparkException: Exception thrown in awaitResult:
...
[info] Cause: org.apache.spark.SparkException:
[INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND] Could not find _partition#26 in
[pk#168,dep#170]. SQLSTATE: XX000
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59893 from dongjoon-hyun/SPARK-60664.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit ec018f385fd4aab72933658c976a1bd2c8f7fcb7)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../RowLevelOperationRuntimeGroupFiltering.scala | 5 ++++-
.../sql/connector/GroupBasedUpdateTableSuite.scala | 26 ++++++++++++++++++++++
2 files changed, 30 insertions(+), 1 deletion(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
index a26a0cea74f1..06d6d2686f4f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/RowLevelOperationRuntimeGroupFiltering.scala
@@ -111,7 +111,10 @@ class
RowLevelOperationRuntimeGroupFiltering(optimizeSubqueries: Rule[LogicalPla
// and must transform the runtime filter condition to use correct expr
IDs for each relation
// note this only applies to group-based row-level operations (i.e.
ReplaceData)
// see RewriteUpdateTable for more details
- val attrMap = buildTableToScanAttrMap(write.table.output,
relation.output)
+ // the condition may also reference metadata columns resolved from the
metadata output
+ // of the target relation, which are not in the table output and must
be mapped too
+ val metadataAttrs = (cond.references -- write.table.output).toSeq
+ val attrMap = buildTableToScanAttrMap(write.table.output ++
metadataAttrs, relation.output)
val transformedCond = cond transform {
case attr: AttributeReference if attrMap.contains(attr) =>
attrMap(attr)
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedUpdateTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedUpdateTableSuite.scala
index 61defff7ebf0..4ce568ed7f68 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedUpdateTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedUpdateTableSuite.scala
@@ -168,6 +168,32 @@ class GroupBasedUpdateTableSuite extends
UpdateTableSuiteBase {
}
}
+ test("SPARK-60664: update runtime group filtering with metadata column and
subquery") {
+ withTempView("updated_pk") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 300, "dep": "hr" }
+ |{ "pk": 2, "salary": 150, "dep": "software" }
+ |{ "pk": 3, "salary": 120, "dep": "hr" }
+ |""".stripMargin)
+
+ Seq(1, 2).toDF("pk").createOrReplaceTempView("updated_pk")
+
+ // the subquery makes the UPDATE read the table twice (UNION), so the
group filter
+ // of each read must refer to the metadata attributes of that read
+ sql(
+ s"""UPDATE $tableNameAsString
+ |SET salary = -1
+ |WHERE _partition = 'hr' AND pk IN (SELECT pk FROM updated_pk)
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr") :: Row(2, 150, "software") :: Row(3, 120, "hr") ::
Nil)
+
+ checkReplacedPartitions(Seq("hr"))
+ }
+ }
+
test("update with nondeterministic conditions") {
createAndInitTable("pk INT NOT NULL, id INT, dep STRING",
"""{ "pk": 1, "id": 1, "dep": "hr" }
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
