dongjoon-hyun opened a new pull request, #59893:
URL: https://github.com/apache/spark/pull/59893
### 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 metadata column such as `_partition` from the metadata output of the
target relation, it is not in that output, so the group filter of the second
read keeps the expression ID of the first read.
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.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
