dongjoon-hyun opened a new pull request, #59896:
URL: https://github.com/apache/spark/pull/59896
### What changes were proposed in this pull request?
This PR aims to treat the metadata columns of the target table as outer
references when `RewriteMergeIntoTable` converts the MERGE `ON` condition into
the runtime group filter condition.
`RewriteMergeIntoTable.toGroupFilterCondition` builds `EXISTS (SELECT * FROM
source WHERE <ON condition>)` and wraps the target table attributes of the `ON`
condition with `OuterReference`. Previously, it checked only
`relation.outputSet`, which has only the data columns. Since a metadata column
of the target table (e.g. `_partition`) is resolved from
`relation.metadataOutput`, it remained a plain attribute inside the subquery
whose only input is the source.
```scala
- val condWithOuterRefs = cond transformUp {
- case attr: Attribute if relation.outputSet.contains(attr) =>
OuterReference(attr)
+ // the condition may also refer to metadata columns of the target table
+ val relationAttrs = AttributeSet(relation.output ++
relation.metadataOutput)
+ val condWithOuterRefs = cond transformUp {
+ case attr: Attribute if relationAttrs.contains(attr) =>
OuterReference(attr)
```
`RowLevelOperationRuntimeGroupFiltering` needs no change. For MERGE, it
plans the group filter on top of the read relation, which already outputs the
metadata columns required by the row-level operation, and assigns new
expression IDs to all of them. So, the group filter scan reads its own
`_partition` column.
```
+- BroadcastHashJoin [pk#5187], [pk#5079], LeftSemi, BuildRight, false, false
:- Project [pk#5187, dep#5189]
: +- Filter (_partition#5190 = hr)
: +- BatchScan cat.ns1.test_table[pk#5187, dep#5189, _partition#5190]
...
```
### Why are the changes needed?
A DSv2 MERGE whose `ON` condition refers to a metadata column of the target
table fails at analysis with an internal error when the runtime group filter is
used, i.e. `spark.sql.optimizer.runtime.rowLevelOperationGroupFilter.enabled`
is `true` (default) and there is no `WHEN NOT MATCHED BY SOURCE` clause. This
happens in group-based MERGE since Apache Spark
[3.5.0](https://github.com/apache/spark/releases/tag/v3.5.0) (2023-09-09,
SPARK-43963) and in delta-based MERGE since Apache Spark
[4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0) (2026-07-11,
SPARK-56669).
```sql
MERGE INTO cat.ns1.test_table t
USING source s
ON t.pk = s.pk AND t._partition = 'hr'
WHEN MATCHED THEN
UPDATE SET t.salary = -1
```
```
org.apache.spark.sql.catalyst.ExtendedAnalysisException:
[MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT] Resolved
attribute(s) "_partition" missing from "pk" in operator !Filter ((outer(pk#21)
= pk#45) AND (_partition#25 = hr)). SQLSTATE: XX000; line 1 pos 0;
ReplaceData RelationV2[pk#21, salary#22, dep#23] cat.ns1.test_table
: +- !Filter ((outer(pk#21) = pk#45) AND (_partition#25 = hr))
: +- SubqueryAlias s
...
```
The rest of the MERGE plan already handles the column because the read
relation includes the required metadata columns. The same statement succeeds
when the runtime group filter is disabled or when the statement has a `WHEN NOT
MATCHED BY SOURCE` clause.
In addition, SPARK-60667 (apache/spark#59895) stops adding metadata columns
to the target table of a MERGE resolved in a later analyzer iteration (e.g. `ON
t.pk = s.pk AND t._partition = 'hr' AND t.salary < 600.5`). After that, such a
MERGE hits the same failure, which this PR fixes too.
### Does this PR introduce _any_ user-facing change?
Yes. Such MERGE statements succeed instead of failing with the internal
error above.
### How was this patch tested?
Pass the CIs with the newly added test cases. They fail without this fix.
```
$ build/sbt 'sql/testOnly *GroupBasedMergeIntoTableSuite
*DeltaBasedMergeIntoTableSuite -- -z SPARK-60668'
```
### 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]
