This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 6b9d033cd415 [SPARK-57681][SQL] Support dynamic table options for
UPDATE
6b9d033cd415 is described below
commit 6b9d033cd415a9492731d64440ef1e122ae5e97d
Author: Anurag Mantripragada <[email protected]>
AuthorDate: Tue Jul 7 17:20:28 2026 -0700
[SPARK-57681][SQL] Support dynamic table options for UPDATE
## What changes were proposed in this pull request?
This PR add the dynamic table options syntax to UPDATE statements.
## Why are the changes needed?
This is a continuation to adding options syntax in DMLs
[SPARK-49098](https://issues.apache.org/jira/browse/SPARK-49098)
Does this PR introduce any user-facing change?
Yes, it adds new SQL syntax. For example:
```sql
UPDATE catalog.db.table WITH (`k` = 'v' )
SET status = 'archived' WHERE created_at < '2024-01-01'.
```
## How was this patch tested?
Added tests to the following suites.
- Parser (DDLParserSuite)
- End-to-end (UpdateTableSuiteBase)
- LogicalWriteInfo (InMemoryRowLevelOperationTable)
## Was this patch authored or co-authored using generative AI tooling?
I used Claude Code (Claude Opus 4.8) to generate the code and tests and
verified manually.
Closes #56792 from anuragmantri/delete-update-sql-options.
Authored-by: Anurag Mantripragada <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/sql/catalyst/parser/SqlBaseParser.g4 | 2 +-
.../sql/catalyst/analysis/RewriteUpdateTable.scala | 3 +-
.../spark/sql/catalyst/parser/AstBuilder.scala | 3 +-
.../spark/sql/catalyst/parser/DDLParserSuite.scala | 33 +++++++++
.../catalog/InMemoryRowLevelOperationTable.scala | 18 +++--
.../spark/sql/connector/UpdateTableSuiteBase.scala | 78 +++++++++++++++++++++-
6 files changed, 127 insertions(+), 10 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 bf10656d6e77..38d4460965fb 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
@@ -745,7 +745,7 @@ dmlStatementNoWith
: insertInto (query | LEFT_PAREN query RIGHT_PAREN queryAlias=tableAlias)
#singleInsertQuery
| fromClause multiInsertQueryBody+
#multiInsertQuery
| DELETE FROM identifierReference tableAlias whereClause?
#deleteFromTable
- | UPDATE identifierReference tableAlias setClause whereClause?
#updateTable
+ | UPDATE identifierReference tableAlias optionsClause? setClause
whereClause? #updateTable
| MERGE (WITH SCHEMA EVOLUTION)? INTO target=identifierReference
targetAlias=tableAlias
USING (source=identifierReference |
LEFT_PAREN sourceQuery=query RIGHT_PAREN) sourceAlias=tableAlias
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
index 5664e4b3a24c..207b3e721704 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteUpdateTable.scala
@@ -26,7 +26,6 @@ import
org.apache.spark.sql.connector.write.{RowLevelOperationTable, SupportsDel
import org.apache.spark.sql.connector.write.RowLevelOperation.Command.UPDATE
import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
ExtractV2Table}
import org.apache.spark.sql.types.IntegerType
-import org.apache.spark.sql.util.CaseInsensitiveStringMap
/**
* A rule that rewrites UPDATE operations using plans that operate on
individual or groups of rows.
@@ -42,7 +41,7 @@ object RewriteUpdateTable extends RewriteRowLevelCommand {
EliminateSubqueryAliases(aliasedTable) match {
case r @ ExtractV2Table(tbl: SupportsRowLevelOperations) =>
checkNoGeneratedColumns(r, UPDATE)
- val table = buildOperationTable(tbl, UPDATE,
CaseInsensitiveStringMap.empty())
+ val table = buildOperationTable(tbl, UPDATE, r.options)
val updateCond = cond.getOrElse(TrueLiteral)
table.operation match {
case _: SupportsDelta =>
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 5832b14f72bc..8389e79331a9 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
@@ -1242,7 +1242,8 @@ class AstBuilder extends DataTypeAstBuilder
override def visitUpdateTable(ctx: UpdateTableContext): LogicalPlan =
withOrigin(ctx) {
val table = createUnresolvedRelation(
- ctx.identifierReference, writePrivileges =
Set(TableWritePrivilege.UPDATE))
+ ctx.identifierReference, Option(ctx.optionsClause()),
+ writePrivileges = Set(TableWritePrivilege.UPDATE))
val tableAlias = getTableAliasWithoutColumnAlias(ctx.tableAlias(),
"UPDATE")
val aliasedTable = tableAlias.map(SubqueryAlias(_, table)).getOrElse(table)
val assignments = withAssignments(ctx.setClause().assignmentList())
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 4edeb3176798..f8d16cb0097c 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
@@ -31,6 +31,7 @@ import
org.apache.spark.sql.connector.expressions.{ApplyTransform, BucketTransfo
import org.apache.spark.sql.connector.expressions.LogicalExpressions.bucket
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataType, Decimal, IntegerType, LongType,
StringType, StructType, TimestampLTZNanosType, TimestampNTZNanosType,
TimestampType, TimeType}
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.storage.StorageLevelMapper
import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
@@ -2253,6 +2254,38 @@ class DDLParserSuite extends AnalysisTest {
stop = 70))
}
+ test("SPARK-57681: update table: with options") {
+ parseCompare(
+ """
+ |UPDATE testcat.ns1.ns2.tbl WITH (`write.split-size` = 10)
+ |SET a='Robert', b=32
+ """.stripMargin,
+ UpdateTable(
+ UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl"),
+ new CaseInsensitiveStringMap(
+ java.util.Map.of("write.split-size", "10"))),
+ Seq(Assignment(UnresolvedAttribute("a"), Literal("Robert")),
+ Assignment(UnresolvedAttribute("b"), Literal(32))),
+ None))
+ }
+
+ test("SPARK-57681: update table: with options and alias") {
+ parseCompare(
+ """
+ |UPDATE testcat.ns1.ns2.tbl AS t WITH (`k` = 'v')
+ |SET t.a='Robert', t.b=32
+ |WHERE t.c=2
+ """.stripMargin,
+ UpdateTable(
+ SubqueryAlias("t",
+ UnresolvedRelation(Seq("testcat", "ns1", "ns2", "tbl"),
+ new CaseInsensitiveStringMap(
+ java.util.Map.of("k", "v")))),
+ Seq(Assignment(UnresolvedAttribute("t.a"), Literal("Robert")),
+ Assignment(UnresolvedAttribute("t.b"), Literal(32))),
+ Some(EqualTo(UnresolvedAttribute("t.c"), Literal(2)))))
+ }
+
test("merge into table: basic") {
parseCompare(
"""
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
index 3b94f19f6fcd..f62f1b0dce25 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
@@ -32,6 +32,14 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.unsafe.types.UTF8String
import org.apache.spark.util.ArrayImplicits._
+/**
+ * Test helper trait mixed into the in-memory row-level operations so tests
can verify that
+ * per-statement SQL options reach the operation via
[[RowLevelOperationInfo#options]].
+ */
+trait RowLevelOperationWithOptions {
+ def options: CaseInsensitiveStringMap
+}
+
class InMemoryRowLevelOperationTable private (
name: String,
columns: Array[Column],
@@ -108,13 +116,14 @@ class InMemoryRowLevelOperationTable private (
override def newRowLevelOperationBuilder(
info: RowLevelOperationInfo): RowLevelOperationBuilder = {
if (properties.getOrDefault(SUPPORTS_DELTAS, "false") == "true") {
- () => DeltaBasedOperation(info.command)
+ () => DeltaBasedOperation(info.command, info.options)
} else {
- () => PartitionBasedOperation(info.command)
+ () => PartitionBasedOperation(info.command, info.options)
}
}
- case class PartitionBasedOperation(command: Command) extends
RowLevelOperation {
+ case class PartitionBasedOperation(command: Command, options:
CaseInsensitiveStringMap)
+ extends RowLevelOperation with RowLevelOperationWithOptions {
var configuredScan: InMemoryBatchScan = _
override def requiredMetadataAttributes(): Array[NamedReference] = {
@@ -183,7 +192,8 @@ class InMemoryRowLevelOperationTable private (
}
}
- case class DeltaBasedOperation(command: Command) extends RowLevelOperation
with SupportsDelta {
+ case class DeltaBasedOperation(command: Command, options:
CaseInsensitiveStringMap)
+ extends RowLevelOperation with SupportsDelta with
RowLevelOperationWithOptions {
private final val PK_COLUMN_REF = FieldReference("pk")
override def requiredMetadataAttributes(): Array[NamedReference] = {
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 8eb314e00df8..c4c5627a12ea 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
@@ -20,9 +20,12 @@ package org.apache.spark.sql.connector
import org.apache.spark.SparkRuntimeException
import org.apache.spark.internal.config
import org.apache.spark.sql.{sources, AnalysisException, Row}
-import org.apache.spark.sql.connector.catalog.{Aborted, Column,
ColumnDefaultValue, Committed, InMemoryTable, TableChange, TableInfo}
+import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
+import org.apache.spark.sql.catalyst.plans.logical.{ReplaceData, WriteDelta}
+import org.apache.spark.sql.connector.catalog.{Aborted, Column,
ColumnDefaultValue, Committed, InMemoryTable, RowLevelOperationWithOptions,
TableChange, TableInfo}
import org.apache.spark.sql.connector.expressions.{GeneralScalarExpression,
LiteralValue}
-import org.apache.spark.sql.connector.write.UpdateSummary
+import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
UpdateSummary}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegerType, StringType}
@@ -1232,4 +1235,75 @@ abstract class UpdateTableSuiteBase extends
RowLevelOperationSuiteBase {
Row(1, 100, "hr"),
Row(2, 200, "software")))
}
+
+ // asserts the given SQL options reached every layer that should carry them:
the rewritten
+ // DataSourceV2Relation, the RowLevelOperationInfo passed to the operation
builder, and the
+ // write builder's LogicalWriteInfo
+ 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-57681: update 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)
+
+ checkRowLevelOperationOptions(
+ sql(s"UPDATE $tableNameAsString WITH (`write.split-size` = 10) SET
salary = -1 WHERE pk = 1"),
+ "write.split-size" -> "10")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr") :: Row(2, 200, "software") :: Nil)
+ }
+
+ test("SPARK-57681: update 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"UPDATE $tableNameAsString WITH (`write.split-size` = 10) SET
salary = -1 " +
+ s"WHERE pk IN (SELECT pk FROM $tableNameAsString WHERE dep = 'hr')"),
+ "write.split-size" -> "10")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr") :: Row(2, 200, "software") :: Row(3, -1, "hr") :: Nil)
+ }
+
+ test("SPARK-57681: update with dynamic options and a CTE 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"WITH hr_rows AS (SELECT pk FROM $tableNameAsString WHERE dep =
'hr') " +
+ s"UPDATE $tableNameAsString WITH (`write.split-size` = 10) SET salary
= -1 " +
+ s"WHERE pk IN (SELECT pk FROM hr_rows)"),
+ "write.split-size" -> "10")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr") :: Row(2, 200, "software") :: Row(3, -1, "hr") :: Nil)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
