From 212020ee10997e5600c6de9dc61358b8fd3fda78 Mon Sep 17 00:00:00 2001 From: brijrajk <22271048+brijrajk@users.noreply.github.com> Date: Thu, 9 Jul 2026 23:20:05 +0530 Subject: [PATCH] [SPARK-58008][SQL] Support dynamic table options for DELETE Extends the per-statement WITH (key=value) options clause to DELETE, consistent with SELECT (SPARK-36680), INSERT (SPARK-49098), and UPDATE (SPARK-57681). Changes: - Grammar: add optionsClause? to the deleteFromTable rule - AstBuilder: pass Option(ctx.optionsClause()) to createUnresolvedRelation - RewriteDeleteFromTable: use r.options instead of CaseInsensitiveStringMap.empty() - DDLParserSuite: two parser tests for DELETE WITH options (with and without alias) - DeleteFromTableSuiteBase: end-to-end tests verifying options reach the row-level operation; use salary < 200 (LessThan) rather than pk = 1 (EqualTo) to prevent OptimizeMetadataOnlyDeleteFromTable from replacing the row-level plan with a filter-only deleteWhere call Co-Authored-By: Claude Sonnet 4.6 --- .../sql/catalyst/parser/SqlBaseParser.g4 | 2 +- .../analysis/RewriteDeleteFromTable.scala | 3 +- .../sql/catalyst/parser/AstBuilder.scala | 3 +- .../sql/catalyst/parser/DDLParserSuite.scala | 27 ++++++++ .../connector/DeleteFromTableSuiteBase.scala | 63 +++++++++++++++++-- 5 files changed, 89 insertions(+), 9 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 38d4460965fb1..cfbfdc2028617 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 @@ -744,7 +744,7 @@ resource dmlStatementNoWith : insertInto (query | LEFT_PAREN query RIGHT_PAREN queryAlias=tableAlias) #singleInsertQuery | fromClause multiInsertQueryBody+ #multiInsertQuery - | DELETE FROM identifierReference tableAlias whereClause? #deleteFromTable + | DELETE FROM identifierReference tableAlias optionsClause? whereClause? #deleteFromTable | UPDATE identifierReference tableAlias optionsClause? setClause whereClause? #updateTable | MERGE (WITH SCHEMA EVOLUTION)? INTO target=identifierReference targetAlias=tableAlias USING (source=identifierReference | diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala index f8881e2077103..c8795b8ad9b1f 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteDeleteFromTable.scala @@ -25,7 +25,6 @@ import org.apache.spark.sql.connector.catalog.{SupportsDeleteV2, SupportsRowLeve import org.apache.spark.sql.connector.write.{RowLevelOperationTable, SupportsDelta} import org.apache.spark.sql.connector.write.RowLevelOperation.Command.DELETE import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, ExtractV2Table} -import org.apache.spark.sql.util.CaseInsensitiveStringMap /** * A rule that rewrites DELETE operations using plans that operate on individual or groups of rows. @@ -45,7 +44,7 @@ object RewriteDeleteFromTable extends RewriteRowLevelCommand { d case r @ ExtractV2Table(t: SupportsRowLevelOperations) => - val table = buildOperationTable(t, DELETE, CaseInsensitiveStringMap.empty()) + val table = buildOperationTable(t, DELETE, r.options) table.operation match { case _: SupportsDelta => buildWriteDeltaPlan(r, table, cond) 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 8389e79331a96..022ea27d4989b 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 @@ -1229,7 +1229,8 @@ class AstBuilder extends DataTypeAstBuilder override def visitDeleteFromTable( ctx: DeleteFromTableContext): LogicalPlan = withOrigin(ctx) { val table = createUnresolvedRelation( - ctx.identifierReference, writePrivileges = Set(TableWritePrivilege.DELETE)) + ctx.identifierReference, Option(ctx.optionsClause()), + writePrivileges = Set(TableWritePrivilege.DELETE)) val tableAlias = getTableAliasWithoutColumnAlias(ctx.tableAlias(), "DELETE") val aliasedTable = tableAlias.map(SubqueryAlias(_, table)).getOrElse(table) val predicate = if (ctx.whereClause() != null) { 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 f8d16cb0097c1..24fbade8f3494 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 @@ -2212,6 +2212,33 @@ class DDLParserSuite extends AnalysisTest { stop = 56)) } + test("SPARK-58008: delete from table: with options") { + parseCompare( + """ + |DELETE FROM testcat.ns1.ns2.tbl WITH (`write.split-size` = 10) + |WHERE a = 1 + """.stripMargin, + DeleteFromTable( + UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl"), + new CaseInsensitiveStringMap( + java.util.Map.of("write.split-size", "10"))), + EqualTo(UnresolvedAttribute("a"), Literal(1)))) + } + + test("SPARK-58008: delete from table: with options and alias") { + parseCompare( + """ + |DELETE FROM testcat.ns1.ns2.tbl AS t WITH (`k` = 'v') + |WHERE t.a = 2 + """.stripMargin, + DeleteFromTable( + SubqueryAlias("t", + UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl"), + new CaseInsensitiveStringMap( + java.util.Map.of("k", "v")))), + EqualTo(UnresolvedAttribute("t.a"), Literal(2)))) + } + test("update table: basic") { parseCompare( """ diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala index b894d5d75b3c8..5b444cd1b6d45 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DeleteFromTableSuiteBase.scala @@ -19,12 +19,12 @@ package org.apache.spark.sql.connector import org.apache.spark.internal.config import org.apache.spark.sql.{AnalysisException, Row} +import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured import org.apache.spark.sql.catalyst.expressions.CheckInvariant -import org.apache.spark.sql.catalyst.plans.logical.Filter -import org.apache.spark.sql.connector.catalog.{Aborted, Committed} -import org.apache.spark.sql.connector.catalog.InMemoryTable -import org.apache.spark.sql.connector.write.DeleteSummary -import org.apache.spark.sql.execution.datasources.v2.{DeleteFromTableExec, ReplaceDataExec, WriteDeltaExec} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, ReplaceData, WriteDelta} +import org.apache.spark.sql.connector.catalog.{Aborted, Committed, InMemoryTable, RowLevelOperationWithOptions} +import org.apache.spark.sql.connector.write.{DeleteSummary, RowLevelOperationTable} +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DeleteFromTableExec, ReplaceDataExec, WriteDeltaExec} import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.sources @@ -1037,6 +1037,59 @@ abstract class DeleteFromTableSuiteBase extends RowLevelOperationSuiteBase { } } + protected def checkRowLevelOperationOptions( + func: => Unit, + expectedOptions: (String, String)*): Unit = { + val Seq(qe) = withQueryExecutionsCaptured(spark)(func) + val writeRelation = qe.optimizedPlan.collectFirst { + case rd: ReplaceData => rd.table + case wd: WriteDelta => wd.table + }.getOrElse(fail("couldn't find row-level operation in optimized plan")) + .asInstanceOf[DataSourceV2Relation] + val operation = writeRelation.table.asInstanceOf[RowLevelOperationTable].operation + .asInstanceOf[RowLevelOperationWithOptions] + expectedOptions.foreach { case (key, value) => + assert(writeRelation.options.get(key) === value, s"relation option '$key'") + assert(operation.options.get(key) === value, s"row-level operation option '$key'") + assert(table.lastWriteInfo.options().get(key) === value, s"write option '$key'") + } + } + + test("SPARK-58008: delete with dynamic options") { + createAndInitTable("pk INT NOT NULL, salary INT, dep STRING", + """{ "pk": 1, "salary": 100, "dep": "hr" } + |{ "pk": 2, "salary": 200, "dep": "software" } + |""".stripMargin) + + // Use salary < 200 (LessThan) rather than pk = 1 (EqualTo) so the optimizer does not + // replace the row-level plan with a filter-only deleteWhere call via + // OptimizeMetadataOnlyDeleteFromTable (canDeleteWhere returns false for LessThan). + checkRowLevelOperationOptions( + sql(s"DELETE FROM $tableNameAsString WITH (`write.split-size` = 10) WHERE salary < 200"), + "write.split-size" -> "10") + + checkAnswer( + sql(s"SELECT * FROM $tableNameAsString"), + Row(2, 200, "software") :: Nil) + } + + test("SPARK-58008: delete with dynamic options and a subquery on the same table") { + 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) + + checkRowLevelOperationOptions( + sql(s"DELETE FROM $tableNameAsString WITH (`write.split-size` = 10) " + + s"WHERE pk IN (SELECT pk FROM $tableNameAsString WHERE dep = 'hr')"), + "write.split-size" -> "10") + + checkAnswer( + sql(s"SELECT * FROM $tableNameAsString"), + Row(2, 200, "software") :: Nil) + } + private def executeDeleteWithRewrite(query: String): Unit = { val executedPlan = executeAndKeepPlan { sql(query)