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 565bb29cd963 [SPARK-60668][SQL] Treat target metadata columns as outer
references in the runtime group filter of DSv2 MERGE
565bb29cd963 is described below
commit 565bb29cd9631c2c9685170c5b3d30c58fe4c044
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Wed Oct 28 21:44:35 2026 -0700
[SPARK-60668][SQL] Treat target metadata columns as outer references in the
runtime group filter of DSv2 MERGE
### 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 Spar [...]
```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
Closes #59896 from dongjoon-hyun/SPARK-60668.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 94ad54d90a5d5bd72fedc2d4fe66fdc41031f7f9)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../catalyst/analysis/RewriteMergeIntoTable.scala | 6 ++--
.../connector/DeltaBasedMergeIntoTableSuite.scala | 35 ++++++++++++++++++++++
.../connector/GroupBasedMergeIntoTableSuite.scala | 34 +++++++++++++++++++++
3 files changed, 73 insertions(+), 2 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
index 1f8e9df4a110..496514a839db 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.catalyst.analysis
import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
AttributeReference, Exists, Expression, IsNotNull, Literal, MetadataAttribute,
MonotonicallyIncreasingID, OuterReference, PredicateHelper, SubqueryExpression}
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
AttributeReference, AttributeSet, Exists, Expression, IsNotNull, Literal,
MetadataAttribute, MonotonicallyIncreasingID, OuterReference, PredicateHelper,
SubqueryExpression}
import org.apache.spark.sql.catalyst.expressions.Literal.{FalseLiteral,
TrueLiteral}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, JoinType,
LeftAnti, LeftOuter, RightOuter}
@@ -248,8 +248,10 @@ object RewriteMergeIntoTable extends
RewriteRowLevelCommand with PredicateHelper
source: LogicalPlan,
cond: Expression): Expression = {
+ // 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 relation.outputSet.contains(attr) =>
OuterReference(attr)
+ case attr: Attribute if relationAttrs.contains(attr) =>
OuterReference(attr)
case other => other
}
val outerRefs = condWithOuterRefs.collect {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoTableSuite.scala
index 53073628f274..1a477514dd10 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeltaBasedMergeIntoTableSuite.scala
@@ -136,6 +136,41 @@ class DeltaBasedMergeIntoTableSuite extends
DeltaBasedMergeIntoTableSuiteBase {
}
}
+ test("SPARK-60668: merge runtime group filtering with metadata column in ON
condition") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "hr" }
+ |{ "pk": 3, "salary": 300, "dep": "software" }
+ |{ "pk": 4, "salary": 400, "dep": "software" }
+ |""".stripMargin)
+
+ val sourceDF = Seq(1, 3).toDF("pk")
+ sourceDF.createOrReplaceTempView("source")
+
+ executeAndCheckScans(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk AND t._partition = 'hr'
+ |WHEN MATCHED THEN
+ | UPDATE SET t.salary = -1
+ |""".stripMargin,
+ primaryScanSchema = "pk INT, dep STRING, _partition STRING",
+ groupFilterScanSchema = Some("pk INT, dep STRING, _partition STRING"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, -1, "hr"), // update
+ Row(2, 200, "hr"), // unchanged
+ Row(3, 300, "software"), // unchanged
+ Row(4, 400, "software"))) // unchanged
+
+ checkLastWriteLog(
+ updateWriteLogEntry(id = 1, metadata = Row("hr", null), data = Row(1,
-1, "hr")))
+ }
+ }
+
test("merge does not double plan table (group filter enabled)") {
withSQLConf(SQLConf.RUNTIME_ROW_LEVEL_OPERATION_GROUP_FILTER_ENABLED.key
-> "true") {
withTempView("source") {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoTableSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoTableSuite.scala
index 95936a749270..1636e88a1dcc 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoTableSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/GroupBasedMergeIntoTableSuite.scala
@@ -184,6 +184,40 @@ class GroupBasedMergeIntoTableSuite extends
MergeIntoTableSuiteBase {
}
}
+ test("SPARK-60668: merge runtime group filtering with metadata column in ON
condition") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "hr" }
+ |{ "pk": 3, "salary": 300, "dep": "software" }
+ |{ "pk": 4, "salary": 400, "dep": "software" }
+ |""".stripMargin)
+
+ val sourceDF = Seq(1, 3).toDF("pk")
+ sourceDF.createOrReplaceTempView("source")
+
+ executeAndCheckScans(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk AND t._partition = 'hr'
+ |WHEN MATCHED THEN
+ | UPDATE SET t.salary = -1
+ |""".stripMargin,
+ primaryScanSchema = "pk INT, salary INT, dep STRING, _partition
STRING, index INT",
+ groupFilterScanSchema = Some("pk INT, dep STRING, _partition STRING"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Seq(
+ Row(1, -1, "hr"), // update
+ Row(2, 200, "hr"), // unchanged
+ Row(3, 300, "software"), // unchanged
+ Row(4, 400, "software"))) // unchanged
+
+ checkReplacedPartitions(Seq("hr"))
+ }
+ }
+
test("merge does not double plan table (group filter enabled)") {
withSQLConf(SQLConf.RUNTIME_ROW_LEVEL_OPERATION_GROUP_FILTER_ENABLED.key
-> "true") {
withTempView("source") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
