You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/09/25 11:22:22 UTC
(spark) branch master updated: [SPARK-58966][SQL] Resolve SQL variables in UPDATE and MERGE INTO conditions
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new e985bd4b9cbf [SPARK-58966][SQL] Resolve SQL variables in UPDATE and MERGE INTO conditions
e985bd4b9cbf is described below
commit e985bd4b9cbfa02f937d7be2c585586a3375dda9
Author: Joel Robin P <jo...@gmail.com>
AuthorDate: Fri Sep 25 11:22:04 2026 +0000
[SPARK-58966][SQL] Resolve SQL variables in UPDATE and MERGE INTO conditions
### What changes were proposed in this pull request?
SQL variables declared with `DECLARE` cannot be referenced in the conditions of an
`UPDATE` or `MERGE INTO` statement. This passes `includeLastResort = true` at the nine
condition-resolution sites that make up those clauses, matching what SPARK-57260 did for
`OverwriteByExpression.deleteExpr`.
Variable resolution only runs from `resolveColsLastResort`, which is reached when
`resolveExpressionByPlanOutput` / `resolveExpressionByPlanChildren` are called with
`includeLastResort = true`. Both default the flag to `false`. Plans with no dedicated
resolution rule fall through to the generic operator case in `ResolveReferences`, which
does pass the flag -- which is why `DELETE ... WHERE` already works. `UPDATE` and
`MERGE INTO` each have a dedicated rule that omitted it:
- `UPDATE` condition (`ResolveReferencesInUpdate`)
- `MERGE` `ON` condition
- `MERGE` `WHEN MATCHED` `DELETE` / `UPDATE` conditions
- `MERGE` `UPDATE *` condition
- `MERGE` `WHEN NOT MATCHED` `INSERT` / `INSERT *` conditions
- `MERGE` `WHEN NOT MATCHED BY SOURCE` `DELETE` / `UPDATE` conditions
This covers every `MergeAction` condition form, so all `MERGE` conditions now resolve
variables consistently.
Assignment values (`SET col = var`, `INSERT VALUES (var)`) are affected by the same root
cause, but they resolve through `resolveExprInAssignment`, which sets
`includeLastResort = false` explicitly rather than by default. That is left unchanged
here pending a decision on whether the explicit `false` was deliberate, so this PR is
scoped to conditions only.
### Why are the changes needed?
Variables resolve in `SELECT`, `INSERT` (including `REPLACE WHERE`) and `DELETE`, but in
no condition of `UPDATE` or `MERGE INTO`, which fails analysis with:
```
[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column, variable, or function parameter with
name `target_dep` cannot be resolved. SQLSTATE: 42703
```
The message itself offers "A column, variable, or function parameter", so the analyzer
reports that variable resolution was attempted. Nothing in `error-conditions.json`, the
tests, or the docs records a restriction here. Where Spark does intentionally block
variables (SPARK-57360, generated columns) it uses an explicit validation and a dedicated
error class, so the inconsistency looks like an oversight rather than a deliberate
limitation.
This affects both session variables and SQL scripting local variables.
### Does this PR introduce _any_ user-facing change?
Yes, a bug fix.
Before, against a table using the built-in DSv2 row-level operation framework:
```sql
DECLARE OR REPLACE VARIABLE target_dep STRING DEFAULT 'hr';
UPDATE cat.ns1.test_table SET salary = 0 WHERE dep = target_dep;
-- [UNRESOLVED_COLUMN.WITH_SUGGESTION] ... `target_dep` cannot be resolved. SQLSTATE: 42703
```
After, the statement analyzes and executes, resolving `target_dep` to the declared
variable. Statements that previously succeeded are unaffected: the flag only enables a
last-resort resolution step that runs after normal column resolution fails, so column
references continue to take precedence over same-named variables.
### How was this patch tested?
Six new tests, plus a shared `withSessionVariable` helper in
`RowLevelOperationSuiteBase`:
- `UpdateTableSuiteBase` -- a session variable and a SQL scripting local variable in an
`UPDATE` condition.
- `MergeIntoTableSuiteBase` -- a variable in the `MERGE` `ON` condition, in the
`WHEN MATCHED` / `WHEN NOT MATCHED BY SOURCE` conditions, and in the
`WHEN NOT MATCHED` `INSERT` and `INSERT *` conditions.
Each test was verified to fail without the fix (raising `UNRESOLVED_COLUMN`) and to pass
with it. The not-matched tests keep a source row that the variable predicate excludes, so
they fail if the variable is resolved but evaluated incorrectly, not just if resolution
fails.
Suites run: `GroupBasedUpdateTableSuite`, `GroupBasedMergeIntoTableSuite`,
`DeltaBasedUpdateTableSuite` and `DeltaBasedMergeIntoTableSuite` -- 289 tests, all
passing. Reverting only the `Analyzer` change makes the two not-matched tests fail with
`UNRESOLVED_COLUMN` on the variable in the `insertaction` condition.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #58244 from joelrobin18/joel-robin/variable-resolution-in-update-merge.
Authored-by: Joel Robin P <jo...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 31 +++--
.../analysis/ResolveReferencesInUpdate.scala | 8 +-
...IntoSchemaEvolutionExtraSourceColumnTests.scala | 35 +++++
.../sql/connector/MergeIntoTableSuiteBase.scala | 148 +++++++++++++++++++++
.../spark/sql/connector/UpdateTableSuiteBase.scala | 35 +++++
5 files changed, 247 insertions(+), 10 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 19df2217568b..34eb76d26999 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -1824,17 +1824,24 @@ class Analyzer(
m
case _ =>
+ // Defer outer-reference and variable resolution until schema evolution has produced
+ // the final target schema so newly added target columns retain precedence.
+ val canResolveLastResort = !m.schemaEvolutionEnabled ||
+ (m.schemaEvolutionReady && m.pendingSchemaChanges.isEmpty)
+
def findAttrInTarget(name: String): Option[Attribute] = {
targetTable.output.find(targetAttr => conf.resolver(name, targetAttr.name))
}
val newMatchedActions = m.matchedActions.map {
case DeleteAction(deleteCondition) =>
val resolvedDeleteCondition = deleteCondition.map(
- resolveExpressionByPlanChildren(_, m))
+ resolveExpressionByPlanChildren(
+ _, m, includeLastResort = canResolveLastResort))
DeleteAction(resolvedDeleteCondition)
case UpdateAction(updateCondition, assignments, fromStar) =>
val resolvedUpdateCondition = updateCondition.map(
- resolveExpressionByPlanChildren(_, m))
+ resolveExpressionByPlanChildren(
+ _, m, includeLastResort = canResolveLastResort))
UpdateAction(
resolvedUpdateCondition,
// The update value can access columns from both target and source tables.
@@ -1857,7 +1864,9 @@ class Analyzer(
}
}
UpdateAction(
- updateCondition.map(resolveExpressionByPlanChildren(_, m)),
+ updateCondition.map(
+ resolveExpressionByPlanChildren(
+ _, m, includeLastResort = canResolveLastResort)),
// For UPDATE *, the value must be from source table.
resolveAssignments(assignments, m, MergeResolvePolicy.SOURCE, throws),
fromStar = true)
@@ -1868,7 +1877,8 @@ class Analyzer(
// The insert action is used when not matched, so its condition and value can only
// access columns from the source table.
val resolvedInsertCondition = insertCondition.map(
- resolveExpressionByPlanOutput(_, m.sourceTable))
+ resolveExpressionByPlanOutput(
+ _, m.sourceTable, includeLastResort = canResolveLastResort))
InsertAction(
resolvedInsertCondition,
resolveAssignments(assignments, m, MergeResolvePolicy.SOURCE, throws))
@@ -1876,7 +1886,8 @@ class Analyzer(
// The insert action is used when not matched, so its condition and value can only
// access columns from the source table.
val resolvedInsertCondition = insertCondition.map(
- resolveExpressionByPlanOutput(_, m.sourceTable))
+ resolveExpressionByPlanOutput(
+ _, m.sourceTable, includeLastResort = canResolveLastResort))
// Expand star to top level source columns. If source has less columns than target,
// assignments will be added by ResolveRowLevelCommandAssignments later.
val assignments = if (m.schemaEvolutionEnabled) {
@@ -1900,11 +1911,13 @@ class Analyzer(
val newNotMatchedBySourceActions = m.notMatchedBySourceActions.map {
case DeleteAction(deleteCondition) =>
val resolvedDeleteCondition = deleteCondition.map(
- resolveExpressionByPlanOutput(_, targetTable))
+ resolveExpressionByPlanOutput(
+ _, targetTable, includeLastResort = canResolveLastResort))
DeleteAction(resolvedDeleteCondition)
case UpdateAction(updateCondition, assignments, fromStar) =>
val resolvedUpdateCondition = updateCondition.map(
- resolveExpressionByPlanOutput(_, targetTable))
+ resolveExpressionByPlanOutput(
+ _, targetTable, includeLastResort = canResolveLastResort))
UpdateAction(
resolvedUpdateCondition,
// The update value can access columns from the target table only.
@@ -1913,7 +1926,9 @@ class Analyzer(
case o => o
}
- val resolvedMergeCondition = resolveExpressionByPlanChildren(m.mergeCondition, m)
+ val resolvedMergeCondition =
+ resolveExpressionByPlanChildren(
+ m.mergeCondition, m, includeLastResort = canResolveLastResort)
m.copy(mergeCondition = resolvedMergeCondition,
matchedActions = newMatchedActions,
notMatchedActions = newNotMatchedActions,
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveReferencesInUpdate.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveReferencesInUpdate.scala
index 92813a156988..9fe46a9ab771 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveReferencesInUpdate.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveReferencesInUpdate.scala
@@ -25,13 +25,16 @@ import org.apache.spark.sql.errors.QueryCompilationErrors
/**
* A virtual rule to resolve [[UnresolvedAttribute]] in [[UpdateTable]]. It's only used by the real
- * rule `ResolveReferences`. The column resolution order for [[UpdateTable]] is:
+ * rule `ResolveReferences`. Assignments and the condition share the first two steps below, then
+ * diverge: step 3 applies to assignments only, and steps 4 and 5 to the condition only.
* 1. Resolves the column to `AttributeReference` with the output of the child plan. This
* includes metadata columns as well.
* 2. Resolves the column to a literal function which is allowed to be invoked without braces, e.g.
* `SELECT col, current_date FROM t`.
* 3. Resolves the column to the default value expression, if the column is the assignment value
* and the corresponding assignment key is a top-level column.
+ * 4. Resolves the column to an outer reference.
+ * 5. Resolves the column to a SQL variable, which is always tried after outer references.
*/
class ResolveReferencesInUpdate(val catalogManager: CatalogManager)
extends SQLConfHelper with ColumnResolutionHelper {
@@ -67,7 +70,8 @@ class ResolveReferencesInUpdate(val catalogManager: CatalogManager)
val newUpdate = u.copy(
assignments = newAssignments,
- condition = u.condition.map(resolveExpressionByPlanChildren(_, u)))
+ condition = u.condition.map(
+ resolveExpressionByPlanChildren(_, u, includeLastResort = true)))
newUpdate.copyTagsFrom(u)
newUpdate
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraSourceColumnTests.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraSourceColumnTests.scala
index ac9c5dbf76b7..4312e4ffe71d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraSourceColumnTests.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoSchemaEvolutionExtraSourceColumnTests.scala
@@ -120,6 +120,41 @@ trait MergeIntoSchemaEvolutionExtraSourceColumnTests extends MergeIntoSchemaEvol
"`active` cannot be resolved"
)
+ test("schema evolution - evolved target column takes precedence over a SQL variable") {
+ withTable(tableNameAsString) {
+ withTempView("source") {
+ createAndInitTable(
+ "pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |{ "pk": 3, "salary": 300, "dep": "hr" }""".stripMargin)
+ Seq((1, 150, "hr", true)).toDF("pk", "salary", "dep", "active")
+ .createOrReplaceTempView("source")
+
+ withSessionVariable("active", "action_enabled") {
+ sql("DECLARE VARIABLE active BOOLEAN DEFAULT false")
+ sql("DECLARE VARIABLE action_enabled BOOLEAN DEFAULT true")
+ executeMerge(
+ withSchemaEvolution = true,
+ targetTableName = tableNameAsString,
+ sourceViewName = "source",
+ cond = "t.pk = s.pk",
+ clauses = Seq(
+ update(set = "salary = s.salary, active = s.active"),
+ updateNotMatched(
+ set = "salary = salary + 1",
+ condition = "active IS NULL AND action_enabled AND salary = 200"),
+ deleteNotMatched(
+ condition = "active IS NULL AND action_enabled AND salary = 300")))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 150, "hr", true) :: Row(2, 201, "software", null) :: Nil)
+ }
+ }
+ }
+ }
+
testEvolution("source has extra column with set all columns")(
targetData = Seq(
(1, 100, "hr"),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
index ea15f13c225c..93338ba043d9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala
@@ -3052,4 +3052,152 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
case None => fail(s"$metricName metric not found")
}
}
+
+ test("merge with a SQL variable in the merge condition") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+ Seq(1, 2).toDF("pk").createOrReplaceTempView("source")
+
+ withSessionVariable("target_dep") {
+ sql("DECLARE VARIABLE target_dep STRING DEFAULT 'hr'")
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk AND t.dep = target_dep
+ |WHEN MATCHED THEN UPDATE SET t.salary = 999
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 999, "hr") :: Row(2, 200, "software") :: Nil)
+ }
+ }
+ }
+
+ test("merge with a SQL variable in matched and not-matched-by-source conditions") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+ Seq(1).toDF("pk").createOrReplaceTempView("source")
+
+ withSessionVariable("salary_threshold") {
+ sql("DECLARE VARIABLE salary_threshold INT DEFAULT 150")
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED AND t.salary < salary_threshold THEN UPDATE SET t.salary = 999
+ |WHEN NOT MATCHED BY SOURCE AND t.salary > salary_threshold THEN DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 999, "hr") :: Nil)
+ }
+ }
+ }
+
+ test("merge with a SQL variable in matched delete condition") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+ Seq(1, 2).toDF("pk").createOrReplaceTempView("source")
+
+ withSessionVariable("salary_threshold") {
+ sql("DECLARE VARIABLE salary_threshold INT DEFAULT 150")
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED AND t.salary < salary_threshold THEN DELETE
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(2, 200, "software") :: Nil)
+ }
+ }
+ }
+
+ test("merge with a SQL variable in matched update-star condition") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+ Seq((1, 999, "finance"), (2, 888, "sales"))
+ .toDF("pk", "salary", "dep").createOrReplaceTempView("source")
+
+ withSessionVariable("pk_threshold") {
+ sql("DECLARE VARIABLE pk_threshold INT DEFAULT 2")
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN MATCHED AND s.pk < pk_threshold THEN UPDATE SET *
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 999, "finance") :: Row(2, 200, "software") :: Nil)
+ }
+ }
+ }
+
+ test("merge with a SQL variable in the not-matched insert condition") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |""".stripMargin)
+ Seq((2, 200, "software"), (3, 300, "hr"))
+ .toDF("pk", "salary", "dep").createOrReplaceTempView("source")
+
+ withSessionVariable("pk_threshold") {
+ sql("DECLARE VARIABLE pk_threshold INT DEFAULT 3")
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN NOT MATCHED AND s.pk < pk_threshold THEN
+ | INSERT (pk, salary, dep) VALUES (s.pk, s.salary, s.dep)
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 100, "hr") :: Row(2, 200, "software") :: Nil)
+ }
+ }
+ }
+
+ test("merge with a SQL variable in the not-matched insert-star condition") {
+ withTempView("source") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |""".stripMargin)
+ Seq((2, 200, "software"), (3, 300, "hr"))
+ .toDF("pk", "salary", "dep").createOrReplaceTempView("source")
+
+ withSessionVariable("pk_threshold") {
+ sql("DECLARE VARIABLE pk_threshold INT DEFAULT 3")
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING source s
+ |ON t.pk = s.pk
+ |WHEN NOT MATCHED AND s.pk < pk_threshold THEN INSERT *
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 100, "hr") :: Row(2, 200, "software") :: Nil)
+ }
+ }
+ }
+
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
index c122713c1873..3269081227e5 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/UpdateTableSuiteBase.scala
@@ -1370,4 +1370,39 @@ abstract class UpdateTableSuiteBase extends RowLevelOperationSuiteBase {
sql(s"SELECT * FROM $tableNameAsString"),
Row(1, -1, "hr") :: Row(2, 200, "software") :: Row(3, -1, "hr") :: Nil)
}
+
+ test("update with a SQL variable in the condition") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+
+ withSessionVariable("target_dep") {
+ sql("DECLARE VARIABLE target_dep STRING DEFAULT 'hr'")
+ sql(s"UPDATE $tableNameAsString SET salary = 999 WHERE dep = target_dep")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 999, "hr") :: Row(2, 200, "software") :: Nil)
+ }
+ }
+
+ test("update with a SQL scripting local variable in the condition") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+
+ sql(
+ s"""BEGIN
+ | DECLARE local_dep STRING DEFAULT 'hr';
+ | UPDATE $tableNameAsString SET salary = 999 WHERE dep = local_dep;
+ |END
+ |""".stripMargin)
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 999, "hr") :: Row(2, 200, "software") :: Nil)
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org