Skip to content

Commit 4825905

Browse files
joelrobin18MaxGekk
authored andcommitted
[SPARK-57260][SQL] Fix variable resolution in REPLACE WHERE clause of INSERT INTO
### What changes were proposed in this pull request? This PR fixes variable resolution in the REPLACE WHERE clause of INSERT INTO statements. REPLACE WHERE is represented as OverwriteByExpression.deleteExpr during analysis. Previously, this expression was resolved only against the target table output because resolveExpressionByPlanOutput was called without includeLastResort = true. This PR enables last-resort resolution for OverwriteByExpression.deleteExpr, allowing SQL variables declared with DECLARE to be resolved in REPLACE WHERE predicates while preserving table-column precedence. ### Why are the changes needed? [SPARK-57260](https://issues.apache.org/jira/browse/SPARK-57260) reports that SQL variables can be used in the VALUES clause of INSERT INTO, but not in the REPLACE WHERE clause. For example, this previously failed during analysis: ``` BEGIN DECLARE x INT DEFAULT 1; INSERT INTO table_y REPLACE WHERE y = x VALUES (x); END ``` The predicate y = x could not resolve x as a SQL variable, resulting in an unresolved column/variable error. ### Does this PR introduce any user-facing change? Yes. Before this change, INSERT INTO ... REPLACE WHERE could not resolve SQL variables declared with DECLARE in the REPLACE WHERE predicate and failed during analysis with an unresolved column/variable error. After this change, INSERT INTO ... REPLACE WHERE can resolve SQL variables declared with DECLARE in the REPLACE WHERE predicate. ### How was this patch tested? Added test coverage for variable resolution in INSERT INTO ... REPLACE WHERE, including: session variables SQL scripting local variables table-column precedence over SQL scripting variables ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Cursor GPT-5.5 and Claude Code Opus 4.8 Closes #56321 from joelrobin18/SPARK-57260-fix-replace-where-variable-resolution. Authored-by: Joel Robin P <joelrobin1818@gmail.com> Signed-off-by: Max Gekk <max.gekk@gmail.com> (cherry picked from commit a32cda3) Signed-off-by: Max Gekk <max.gekk@gmail.com>
1 parent 5bee0db commit 4825905

3 files changed

Lines changed: 73 additions & 1 deletion

File tree

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1698,7 +1698,8 @@ class Analyzer(
16981698
case o: OverwriteByExpression if o.table.resolved =>
16991699
// The delete condition of `OverwriteByExpression` will be passed to the table
17001700
// implementation and should be resolved based on the table schema.
1701-
o.copy(deleteExpr = resolveExpressionByPlanOutput(o.deleteExpr, o.table))
1701+
o.copy(deleteExpr = resolveExpressionByPlanOutput(
1702+
o.deleteExpr, o.table, includeLastResort = true))
17021703

17031704
case u: UpdateTable => resolveReferencesInUpdate(u)
17041705

sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3923,6 +3923,29 @@ class DataSourceV2SQLSuiteV1Filter
39233923
}
39243924
}
39253925

3926+
test("Session variable in INSERT REPLACE WHERE") {
3927+
val t = "testcat.tbl"
3928+
withTable(t) {
3929+
spark.sql(s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)")
3930+
spark.sql(s"INSERT INTO TABLE $t VALUES (1, 'a'), (2, 'b'), (3, 'c')")
3931+
spark.sql("DECLARE OR REPLACE VARIABLE replacement_id BIGINT DEFAULT 2")
3932+
try {
3933+
spark.sql(
3934+
s"""
3935+
|INSERT INTO $t
3936+
| REPLACE WHERE id = replacement_id
3937+
| VALUES (2, 'bb')
3938+
|""".stripMargin)
3939+
3940+
checkAnswer(
3941+
spark.table(t),
3942+
Seq(Row(1L, "a"), Row(2L, "bb"), Row(3L, "c")))
3943+
} finally {
3944+
spark.sql("DROP TEMPORARY VARIABLE IF EXISTS replacement_id")
3945+
}
3946+
}
3947+
}
3948+
39263949
test("Selective Overwrite: REPLACE WHERE with BY NAME - column reordering") {
39273950
val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L, "c"))).toDF("id", "data")
39283951
df.createOrReplaceTempView("source")

sql/core/src/test/scala/org/apache/spark/sql/scripting/SqlScriptingE2eSuite.scala

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -348,6 +348,54 @@ class SqlScriptingE2eSuite extends SharedSparkSession {
348348
}
349349
}
350350

351+
test("variable in REPLACE WHERE clause of INSERT INTO") {
352+
withCatalog("cat") { _ =>
353+
withTable("cat.ns1.t") {
354+
val sqlScript =
355+
"""
356+
|BEGIN
357+
| DECLARE x STRING DEFAULT 'hr';
358+
| CREATE TABLE cat.ns1.t (pk INT NOT NULL, salary INT, dep STRING)
359+
| PARTITIONED BY (dep);
360+
| INSERT INTO cat.ns1.t VALUES (1, 100, 'hr'), (2, 200, 'software');
361+
| INSERT INTO cat.ns1.t
362+
| REPLACE WHERE dep = x
363+
| VALUES (1, 150, 'hr');
364+
| SELECT * FROM cat.ns1.t ORDER BY pk;
365+
|END
366+
|""".stripMargin
367+
368+
verifySqlScriptResult(
369+
sqlScript,
370+
Seq(Row(1, 150, "hr"), Row(2, 200, "software")))
371+
}
372+
}
373+
}
374+
375+
test("REPLACE WHERE resolves table columns before SQL scripting variables") {
376+
withCatalog("cat") { _ =>
377+
withTable("cat.ns1.t") {
378+
val sqlScript =
379+
"""
380+
|BEGIN
381+
| DECLARE dep STRING DEFAULT 'software';
382+
| CREATE TABLE cat.ns1.t (pk INT NOT NULL, salary INT, dep STRING)
383+
| PARTITIONED BY (dep);
384+
| INSERT INTO cat.ns1.t VALUES (1, 100, 'hr'), (2, 200, 'software');
385+
| INSERT INTO cat.ns1.t
386+
| REPLACE WHERE dep = 'hr'
387+
| VALUES (1, 150, 'hr');
388+
| SELECT * FROM cat.ns1.t ORDER BY pk;
389+
|END
390+
|""".stripMargin
391+
392+
verifySqlScriptResult(
393+
sqlScript,
394+
Seq(Row(1, 150, "hr"), Row(2, 200, "software")))
395+
}
396+
}
397+
}
398+
351399
test("continue handler with transactional checks - handler DML runs in its own transaction") {
352400
withCatalog("cat") { catalog =>
353401
withTable("cat.ns1.t") {

0 commit comments

Comments
 (0)