From 336b25494640a808b3c3eeb0abcac54e3f5265ec Mon Sep 17 00:00:00 2001 From: Sandy Ryza Date: Fri, 12 Dec 2025 20:08:25 -0800 Subject: [PATCH 1/4] INSERT INTO BY NAME REPLACE WHERE --- .../sql/catalyst/parser/SqlBaseParser.g4 | 2 +- .../sql/catalyst/parser/AstBuilder.scala | 14 ++- .../sql/catalyst/parser/DDLParserSuite.scala | 27 +++++ .../sql/connector/DataSourceV2SQLSuite.scala | 99 +++++++++++++++++++ 4 files changed, 136 insertions(+), 6 deletions(-) diff --git a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 index c29452cab1804..80e51d622c8ce 100644 --- a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 +++ b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 @@ -562,7 +562,7 @@ query insertInto : INSERT OVERWRITE TABLE? identifierReference optionsClause? (partitionSpec (IF errorCapturingNot EXISTS)?)? ((BY NAME) | identifierList)? #insertOverwriteTable | INSERT INTO TABLE? identifierReference optionsClause? partitionSpec? (IF errorCapturingNot EXISTS)? ((BY NAME) | identifierList)? #insertIntoTable - | INSERT INTO TABLE? identifierReference optionsClause? REPLACE whereClause #insertIntoReplaceWhere + | INSERT INTO TABLE? identifierReference optionsClause? (BY NAME)? REPLACE whereClause #insertIntoReplaceWhere | INSERT OVERWRITE LOCAL? DIRECTORY path=stringLit rowFormat? createFileFormat? #insertOverwriteHiveDir | INSERT OVERWRITE LOCAL? DIRECTORY (path=stringLit)? tableProvider (OPTIONS options=propertyList)? #insertOverwriteDir ; diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala index c3571ad709622..578f0fffe72eb 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala @@ -928,11 +928,15 @@ class AstBuilder extends DataTypeAstBuilder case ctx: InsertIntoReplaceWhereContext => val options = Option(ctx.optionsClause()) withIdentClause(ctx.identifierReference, Seq(query), (ident, otherPlans) => { - OverwriteByExpression.byPosition( - createUnresolvedRelation(ctx.identifierReference, ident, options, - Seq(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE), isStreaming = false), - otherPlans.head, - expression(ctx.whereClause().booleanExpression())) + val table = createUnresolvedRelation(ctx.identifierReference, ident, options, + Seq(TableWritePrivilege.INSERT, TableWritePrivilege.DELETE), isStreaming = false) + val deleteExpr = expression(ctx.whereClause().booleanExpression()) + val isByName = ctx.NAME() != null + if (isByName) { + OverwriteByExpression.byName(table, otherPlans.head, deleteExpr) + } else { + OverwriteByExpression.byPosition(table, otherPlans.head, deleteExpr) + } }) case dir: InsertOverwriteDirContext => val (isLocal, storage, provider) = visitInsertOverwriteDir(dir) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala index 0cb6f53f273ab..9e2def3072abb 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/DDLParserSuite.scala @@ -49,6 +49,11 @@ class DDLParserSuite extends AnalysisTest { case u: UnresolvedRelation => i.copy(table = u.clearWritePrivileges) case _ => i } + case o: OverwriteByExpression => + o.table match { + case u: UnresolvedRelation => o.copy(table = u.clearWritePrivileges) + case _ => o + } } comparePlans(parsed, expected, checkAnalysis = false) } @@ -1763,6 +1768,28 @@ class DDLParserSuite extends AnalysisTest { ) } + test("insert table: REPLACE WHERE with BY NAME") { + parseCompare( + "INSERT INTO testcat.ns1.ns2.tbl BY NAME REPLACE WHERE a > 5 SELECT * FROM source", + OverwriteByExpression.byName( + UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), + Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), + org.apache.spark.sql.catalyst.expressions.GreaterThan( + UnresolvedAttribute("a"), + Literal(5)))) + } + + test("insert table: REPLACE WHERE without BY NAME") { + parseCompare( + "INSERT INTO testcat.ns1.ns2.tbl REPLACE WHERE a > 5 SELECT * FROM source", + OverwriteByExpression.byPosition( + UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl")), + Project(Seq(UnresolvedStar(None)), UnresolvedRelation(Seq("source"))), + org.apache.spark.sql.catalyst.expressions.GreaterThan( + UnresolvedAttribute("a"), + Literal(5)))) + } + test("delete from table: delete all") { parseCompare("DELETE FROM testcat.ns1.ns2.tbl", DeleteFromTable( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala index 5a9494cf81730..9c45348ad385f 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala @@ -3534,6 +3534,105 @@ class DataSourceV2SQLSuiteV1Filter } } + test("Overwrite: REPLACE WHERE with BY NAME - column reordering") { + val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L, "c"))).toDF("id", "data") + df.createOrReplaceTempView("source") + val df2 = spark.createDataFrame(Seq(("d", 4L), ("e", 5L), ("f", 6L))).toDF("data", "id") + df2.createOrReplaceTempView("source2_reordered") + + val t = "testcat.tbl" + withTable(t) { + spark.sql( + s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)") + spark.sql(s"INSERT INTO TABLE $t SELECT * FROM source") + + checkAnswer( + spark.table(s"$t"), + Seq(Row(1L, "a"), Row(2L, "b"), Row(3L, "c"))) + + spark.sql(s"INSERT INTO $t BY NAME REPLACE WHERE id = 3 SELECT * FROM source2_reordered") + checkAnswer( + spark.table(s"$t"), + Seq(Row(1L, "a"), Row(2L, "b"), Row(4L, "d"), Row(5L, "e"), Row(6L, "f"))) + } + } + + test("Overwrite: REPLACE WHERE without BY NAME - positional matching") { + val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L, "c"))).toDF("id", "data") + df.createOrReplaceTempView("source") + val df2 = spark.createDataFrame(Seq((4L, "d"), (5L, "e"), (6L, "f"))).toDF("data", "id") + df2.createOrReplaceTempView("source2_names_reordered") + + val t = "testcat.tbl" + withTable(t) { + spark.sql( + s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)") + spark.sql(s"INSERT INTO TABLE $t SELECT * FROM source") + + checkAnswer( + spark.table(s"$t"), + Seq(Row(1L, "a"), Row(2L, "b"), Row(3L, "c"))) + + spark.sql(s"INSERT INTO $t REPLACE WHERE id = 3 SELECT * FROM source2_names_reordered") + checkAnswer( + spark.table(s"$t"), + Seq(Row(1L, "a"), Row(2L, "b"), Row(4L, "d"), Row(5L, "e"), Row(6L, "f"))) + } + } + + test("Overwrite: REPLACE WHERE without BY NAME - different order, compatible types") { + val df = spark.createDataFrame(Seq((1L, 11L), (2L, 12L), (3L, 13L))).toDF("id", "data") + df.createOrReplaceTempView("source") + val df2 = spark.createDataFrame(Seq((14L, 4L), (15L, 5L), (16L, 6L))).toDF("data", "id") + df2.createOrReplaceTempView("source2_reordered") + + val t = "testcat.tbl" + withTable(t) { + spark.sql( + s"CREATE TABLE $t (id bigint, data bigint) USING foo PARTITIONED BY (id)") + spark.sql(s"INSERT INTO TABLE $t SELECT * FROM source") + + checkAnswer( + spark.table(s"$t"), + Seq(Row(1L, 11L), Row(2L, 12L), Row(3L, 13L))) + + spark.sql(s"INSERT INTO $t REPLACE WHERE id = 3 SELECT * FROM source2_reordered") + checkAnswer( + spark.table(s"$t"), + Seq(Row(1L, 11L), Row(2L, 12L), Row(14L, 4L), Row(15L, 5L), Row(16L, 6L))) + } + } + + test("Overwrite: REPLACE WHERE without BY NAME - incompatible types") { + val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L, "c"))).toDF("id", "data") + df.createOrReplaceTempView("source") + val df2 = spark.createDataFrame(Seq(("d", 4L), ("e", 5L), ("f", 6L))).toDF("data", "id") + df2.createOrReplaceTempView("source2_reordered") + + val t = "testcat.tbl" + withTable(t) { + spark.sql( + s"CREATE TABLE $t (id bigint, data string) USING foo PARTITIONED BY (id)") + spark.sql(s"INSERT INTO TABLE $t SELECT * FROM source") + + checkAnswer( + spark.table(s"$t"), + Seq(Row(1L, "a"), Row(2L, "b"), Row(3L, "c"))) + + checkError( + exception = intercept[AnalysisException] { + spark.sql(s"INSERT INTO $t REPLACE WHERE id = 3 SELECT * FROM source2_reordered") + }, + condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST", + parameters = Map( + "tableName" -> "`testcat`.`tbl`", + "colName" -> "`id`", + "srcType" -> "\"STRING\"", + "targetType" -> "\"BIGINT\"") + ) + } + } + test("SPARK-46144: Fail overwrite statement if the condition contains subquery") { val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L, "c"))).toDF("id", "data") df.createOrReplaceTempView("source") From a931b0686e9063141cf046ec9fd0d4e5652ed29d Mon Sep 17 00:00:00 2001 From: Sandy Ryza Date: Fri, 2 Jan 2026 16:07:20 -0800 Subject: [PATCH 2/4] Fix formatting in SqlBaseParser.g4 for insertInto rule --- .../org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 index 80e51d622c8ce..21b8781f43cea 100644 --- a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 +++ b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4 @@ -562,7 +562,7 @@ query insertInto : INSERT OVERWRITE TABLE? identifierReference optionsClause? (partitionSpec (IF errorCapturingNot EXISTS)?)? ((BY NAME) | identifierList)? #insertOverwriteTable | INSERT INTO TABLE? identifierReference optionsClause? partitionSpec? (IF errorCapturingNot EXISTS)? ((BY NAME) | identifierList)? #insertIntoTable - | INSERT INTO TABLE? identifierReference optionsClause? (BY NAME)? REPLACE whereClause #insertIntoReplaceWhere + | INSERT INTO TABLE? identifierReference optionsClause? (BY NAME)? REPLACE whereClause #insertIntoReplaceWhere | INSERT OVERWRITE LOCAL? DIRECTORY path=stringLit rowFormat? createFileFormat? #insertOverwriteHiveDir | INSERT OVERWRITE LOCAL? DIRECTORY (path=stringLit)? tableProvider (OPTIONS options=propertyList)? #insertOverwriteDir ; From c2fdab8e48f22a9ed812ef6ce49b933edec43467 Mon Sep 17 00:00:00 2001 From: Sandy Ryza Date: Fri, 2 Jan 2026 16:07:50 -0800 Subject: [PATCH 3/4] Apply suggestions from code review Co-authored-by: Thang Long Vu <107926660+longvu-db@users.noreply.github.com> --- .../org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala index 9c45348ad385f..6e415d8d28bab 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala @@ -3534,7 +3534,7 @@ class DataSourceV2SQLSuiteV1Filter } } - test("Overwrite: REPLACE WHERE with BY NAME - column reordering") { + test("Selective Overwrite: REPLACE WHERE with BY NAME - column reordering") { val df = spark.createDataFrame(Seq((1L, "a"), (2L, "b"), (3L, "c"))).toDF("id", "data") df.createOrReplaceTempView("source") val df2 = spark.createDataFrame(Seq(("d", 4L), ("e", 5L), ("f", 6L))).toDF("data", "id") @@ -3580,7 +3580,7 @@ class DataSourceV2SQLSuiteV1Filter } } - test("Overwrite: REPLACE WHERE without BY NAME - different order, compatible types") { + test("Overwrite: REPLACE WHERE without BY NAME - different column order between 2 tables, compatible types") { val df = spark.createDataFrame(Seq((1L, 11L), (2L, 12L), (3L, 13L))).toDF("id", "data") df.createOrReplaceTempView("source") val df2 = spark.createDataFrame(Seq((14L, 4L), (15L, 5L), (16L, 6L))).toDF("data", "id") From da1351dd76d84330e53a9c1cd8d4223507262fa5 Mon Sep 17 00:00:00 2001 From: Sandy Ryza Date: Sun, 4 Jan 2026 09:41:14 -0800 Subject: [PATCH 4/4] fix line length --- .../org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala index 6e415d8d28bab..d1255da0590c0 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala @@ -3580,7 +3580,8 @@ class DataSourceV2SQLSuiteV1Filter } } - test("Overwrite: REPLACE WHERE without BY NAME - different column order between 2 tables, compatible types") { + test("Overwrite: REPLACE WHERE without BY NAME - different column order between 2 tables, " + + "compatible types") { val df = spark.createDataFrame(Seq((1L, 11L), (2L, 12L), (3L, 13L))).toDF("id", "data") df.createOrReplaceTempView("source") val df2 = spark.createDataFrame(Seq((14L, 4L), (15L, 5L), (16L, 6L))).toDF("data", "id")