dongjoon-hyun opened a new pull request, #59920:
URL: https://github.com/apache/spark/pull/59920
[SPARK-60710][SQL] Expose source metadata columns to the runtime group
filter of DSv2 MERGE
### What changes were proposed in this pull request?
This PR aims to make the runtime group filter of DSv2 MERGE read the source
metadata columns that the `ON` condition refers to.
`RewriteMergeIntoTable.toGroupFilterCondition` converts the `ON` condition
into `EXISTS (SELECT * FROM source WHERE <ON condition>)` on top of the
original source plan. A source metadata column (e.g. `s._partition` of a DSv2
table or `s._metadata` of a file source table) is resolved from
`source.metadataOutput`, so the source plan does not output it yet when the
MERGE is rewritten. For the main join, `Analyzer.AddMetadataColumns` adds the
column to the source in the next analyzer iteration, but the rule does not
reach the subquery plan. In addition, `DeduplicateRelations` assigns new
expression IDs to the subquery's copy of the source relation, so the subquery
keeps a reference to the `_partition` attribute of the main join's source.
```
ReplaceData RelationV2[pk#22, salary#23, dep#24] cat.ns1.test_table
: +- !Filter ((outer(pk#22) = pk#56) AND (_partition#30 = hr))
: +- SubqueryAlias s
: +- SubqueryAlias cat.ns1.source_table
: +- RelationV2[pk#56, dep#57] cat.ns1.source_table
+- MergeRows[...]
+- Join LeftOuter, ((pk#22 = pk#25) AND (_partition#30 = hr)), ...
:- ...
+- Project [pk#25, dep#26, true AS __row_from_source#34, _partition#30]
+- SubqueryAlias s
+- SubqueryAlias cat.ns1.source_table
+- RelationV2[pk#25, dep#26, index#29, _partition#30]
cat.ns1.source_table
```
This PR adds the referenced source metadata columns to the source plan of
the subquery in the same way as `AddMetadataColumns` does, i.e.
`ExposesMetadataColumns.withMetadataColumns()` for the relation and extra
project list entries for the `Project` nodes in between. The source plan is
unchanged when the `ON` condition does not refer to source metadata columns.
```scala
+ // the condition may refer to source metadata columns that are not in
the source output
+ // AddMetadataColumns adds them to the source in the join but does not
reach this subquery
+ val sourceMetadataAttrs =
+ AttributeSet(source.metadataOutput).intersect(cond.references) --
source.outputSet
+ val filterSource = if (sourceMetadataAttrs.isEmpty) {
+ source
+ } else {
+ addMetadataAttrs(source, sourceMetadataAttrs)
+ }
+ Exists(Filter(condWithOuterRefs, filterSource), outerRefs)
```
Since the subquery's source relation now outputs the column,
`DeduplicateRelations` renews it together with the other columns, and the group
filter applies the source predicate.
```
+- BroadcastHashJoin [pk#4009], [pk#3879], LeftSemi, BuildRight, false, false
:- Project [pk#4009, dep#4011]
: +- BatchScan cat.ns1.test_table[pk#4009, dep#4011] ...
+- BroadcastQueryStage 0
+- BroadcastExchange ...
+- *(1) Project [pk#3879]
+- *(1) Filter ((_partition#3882 = hr) AND isnotnull(pk#3879))
+- BatchScan cat.ns1.source_table[pk#3879, _partition#3882]
...
```
Two simpler alternatives were considered and not chosen.
- Skip the group filter when the `ON` condition refers to source metadata
columns. This is what disabling the group filter does today, and a group-based
MERGE reads and replaces every group of the target table.
- Drop the predicates on source metadata columns from the group filter. The
filter still selects every group with matching rows, but it can select more. In
the new test, the `software` partition would be replaced although none of its
rows changes.
### Why are the changes needed?
A DSv2 MERGE whose `ON` condition refers to a metadata column of the source
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.
```sql
CREATE TABLE cat.ns1.source_table (pk INT, dep STRING) PARTITIONED BY (dep);
INSERT INTO cat.ns1.source_table VALUES (1, 'hr'), (3, 'software'), (5,
'hr');
MERGE INTO cat.ns1.test_table t
USING cat.ns1.source_table s
ON t.pk = s.pk AND s._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", "dep" in operator !Filter
((outer(pk#22) = pk#56) AND (_partition#30 = hr)). SQLSTATE: XX000; line 1 pos
0;
```
The same statement succeeds when the runtime group filter is disabled. A
MERGE that is rewritten in a later analyzer iteration, e.g. because its `ON`
condition needs type coercion, does not fail either, because
`AddMetadataColumns` adds the column to the MERGE source before the rewrite.
The subquery has been built on top of the original source plan since the
runtime group filter of group-based MERGE was added in Apache Spark
[3.5.0](https://github.com/apache/spark/releases/tag/v3.5.0) (2023-09-09,
SPARK-43963), and delta-based MERGE uses it since Apache Spark
[4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0) (2026-07-11,
SPARK-56669).
This is the source-side counterpart of SPARK-60668 (apache/spark#59896),
which made the runtime group filter treat the target metadata columns in the
same condition as outer references.
### Does this PR introduce _any_ user-facing change?
Yes. Such MERGE statements succeed instead of failing with the internal
error above, and the runtime group filter honors the predicates on source
metadata columns.
### How was this patch tested?
Pass the CIs with the newly added test cases. They fail without this fix.
Each test runs the statement with the DSv2 source table and with a subquery on
top of it, which covers the `Project` case. The group-based test also checks
that only the `hr` partition is replaced.
```
$ build/sbt 'sql/testOnly *GroupBasedMergeIntoTableSuite
*DeltaBasedMergeIntoTableSuite -- -z SPARK-60710'
```
### 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]
