This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 58e2554111f1 [SPARK-58007][SQL] Support dynamic table options for MERGE
58e2554111f1 is described below
commit 58e2554111f1311556dde32c7c26ec7d4fe96471
Author: Anurag Mantripragada <[email protected]>
AuthorDate: Tue Jul 21 08:50:27 2026 -0700
[SPARK-58007][SQL] Support dynamic table options for MERGE
## What changes were proposed in this pull request?
This PR adds the dynamic table options syntax to MERGE statements. MERGE
requires options on both the target table and the source table.
## Why are the changes needed?
This is a continuation to adding options syntax in DMLs SPARK-49098
## Does this PR introduce any user-facing change?
Yes, it adds new SQL syntax. For example:
```sql
MERGE INTO catalog.db.target t WITH (`write.split-size` = 10)
USING catalog.db.source WITH (`split-size` = 5) s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.status = 'archived'
WHEN NOT MATCHED THEN INSERT *
```
## How was this patch tested?
Added tests to the following suites.
- Parser (DDLParserSuite)
- End-to-end (MergeIntoTableSuiteBase)
## 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 #57227 from anuragmantri/merge-sql-options.
Lead-authored-by: Anurag Mantripragada <[email protected]>
Co-authored-by: x <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 8a33fb73fa520ba04be46a1c350e46d79234b53c)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../spark/sql/catalyst/parser/SqlBaseParser.g4 | 3 +-
.../catalyst/analysis/RewriteMergeIntoTable.scala | 3 +-
.../spark/sql/catalyst/parser/AstBuilder.scala | 8 +-
.../spark/sql/catalyst/parser/DDLParserSuite.scala | 49 +++++++
.../sql/connector/MergeIntoTableSuiteBase.scala | 144 ++++++++++++++++++++-
.../sql/connector/RowLevelOperationSuiteBase.scala | 23 +++-
.../spark/sql/connector/UpdateTableSuiteBase.scala | 28 +---
7 files changed, 225 insertions(+), 33 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 e64d88cce1ec..ca5de63ecb0b 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
@@ -750,7 +750,8 @@ dmlStatementNoWith
| DELETE FROM identifierReference tableAlias whereClause?
#deleteFromTable
| UPDATE identifierReference tableAlias optionsClause? setClause
whereClause? #updateTable
| MERGE (WITH SCHEMA EVOLUTION)? INTO target=identifierReference
targetAlias=tableAlias
- USING (source=identifierReference |
+ targetOptions=optionsClause?
+ USING (source=identifierReference sourceOptions=optionsClause? |
LEFT_PAREN sourceQuery=query RIGHT_PAREN) sourceAlias=tableAlias
ON mergeCondition=booleanExpression
matchedClause*
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
index e11a985d85dd..714f815161ae 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RewriteMergeIntoTable.scala
@@ -31,7 +31,6 @@ import
org.apache.spark.sql.connector.write.RowLevelOperation.Command.MERGE
import org.apache.spark.sql.errors.QueryCompilationErrors
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 MERGE operations using plans that operate on
individual or groups of rows.
@@ -128,7 +127,7 @@ object RewriteMergeIntoTable extends RewriteRowLevelCommand
with PredicateHelper
case r @ ExtractV2Table(tbl: SupportsRowLevelOperations) =>
checkNoGeneratedColumns(r, MERGE)
validateMergeIntoConditions(m)
- val table = buildOperationTable(tbl, MERGE,
CaseInsensitiveStringMap.empty())
+ val table = buildOperationTable(tbl, MERGE, r.options)
table.operation match {
case _: SupportsDelta =>
buildWriteDeltaPlan(
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 dcb8d42153ff..2100241ce9db 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
@@ -1268,8 +1268,13 @@ class AstBuilder extends DataTypeAstBuilder
override def visitMergeIntoTable(ctx: MergeIntoTableContext): LogicalPlan =
withOrigin(ctx) {
val withSchemaEvolution = ctx.EVOLUTION() != null
+ // The target and source may each carry their own dynamic table options
via `WITH (...)`.
+ // Known limitation: if the same table is used as both the target and the
source
+ // (e.g. `MERGE INTO t WITH (a) USING t WITH (b) s`), the analyzer's
relation cache is keyed
+ // without options and reuses the first resolved relation, so the target's
options win and the
+ // source's are silently dropped.
val sourceTableOrQuery = if (ctx.source != null) {
- createUnresolvedRelation(ctx.source)
+ createUnresolvedRelation(ctx.source, Option(ctx.sourceOptions))
} else if (ctx.sourceQuery != null) {
visitQuery(ctx.sourceQuery)
} else {
@@ -1355,6 +1360,7 @@ class AstBuilder extends DataTypeAstBuilder
val targetTable = createUnresolvedRelation(
ctx.target,
+ Option(ctx.targetOptions),
writePrivileges = MergeIntoTable.getWritePrivileges(
matchedActions, notMatchedActions, notMatchedBySourceActions))
val targetTableAlias = getTableAliasWithoutColumnAlias(ctx.targetAlias,
"MERGE")
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 4870c118dbc7..f33c11c8a2d4 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
@@ -2317,6 +2317,55 @@ class DDLParserSuite extends AnalysisTest {
withSchemaEvolution = false))
}
+ test("SPARK-58007: merge into table: with target and source options") {
+ parseCompare(
+ """
+ |MERGE INTO testcat1.ns1.ns2.tbl AS target WITH (`write.split-size` =
10)
+ |USING testcat2.ns1.ns2.tbl WITH (`split-size` = 5) AS source
+ |ON target.col1 = source.col1
+ |WHEN MATCHED THEN UPDATE SET target.col2 = source.col2
+ |WHEN NOT MATCHED THEN INSERT (target.col1, target.col2) values
(source.col1, source.col2)
+ """.stripMargin,
+ MergeIntoTable(
+ SubqueryAlias("target",
+ UnresolvedRelation(Seq("testcat1", "ns1", "ns2", "tbl"),
+ new CaseInsensitiveStringMap(java.util.Map.of("write.split-size",
"10")))),
+ SubqueryAlias("source",
+ UnresolvedRelation(Seq("testcat2", "ns1", "ns2", "tbl"),
+ new CaseInsensitiveStringMap(java.util.Map.of("split-size",
"5")))),
+ EqualTo(UnresolvedAttribute("target.col1"),
UnresolvedAttribute("source.col1")),
+ Seq(UpdateAction(None,
+ Seq(Assignment(UnresolvedAttribute("target.col2"),
+ UnresolvedAttribute("source.col2"))))),
+ Seq(InsertAction(None,
+ Seq(Assignment(UnresolvedAttribute("target.col1"),
UnresolvedAttribute("source.col1")),
+ Assignment(UnresolvedAttribute("target.col2"),
UnresolvedAttribute("source.col2"))))),
+ Seq.empty,
+ withSchemaEvolution = false))
+ }
+
+ test("SPARK-58007: merge into table: with target options only") {
+ parseCompare(
+ """
+ |MERGE INTO testcat1.ns1.ns2.tbl AS target WITH (`k` = 'v')
+ |USING testcat2.ns1.ns2.tbl AS source
+ |ON target.col1 = source.col1
+ |WHEN MATCHED THEN UPDATE SET target.col2 = source.col2
+ """.stripMargin,
+ MergeIntoTable(
+ SubqueryAlias("target",
+ UnresolvedRelation(Seq("testcat1", "ns1", "ns2", "tbl"),
+ new CaseInsensitiveStringMap(java.util.Map.of("k", "v")))),
+ SubqueryAlias("source", UnresolvedRelation(Seq("testcat2", "ns1",
"ns2", "tbl"))),
+ EqualTo(UnresolvedAttribute("target.col1"),
UnresolvedAttribute("source.col1")),
+ Seq(UpdateAction(None,
+ Seq(Assignment(UnresolvedAttribute("target.col2"),
+ UnresolvedAttribute("source.col2"))))),
+ Seq.empty,
+ Seq.empty,
+ withSchemaEvolution = false))
+ }
+
test("merge into table: using subquery") {
parseCompare(
"""
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 1dbc1001b5c0..9b625c1b81d2 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
@@ -20,14 +20,16 @@ package org.apache.spark.sql.connector
import org.apache.spark.SparkRuntimeException
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.{AttributeReference, EqualTo,
In, Not}
import org.apache.spark.sql.catalyst.optimizer.BuildLeft
-import org.apache.spark.sql.connector.catalog.{Aborted, Column,
ColumnDefaultValue, Committed, InMemoryTable, TableInfo}
+import org.apache.spark.sql.catalyst.plans.logical.{ReplaceData, WriteDelta}
+import org.apache.spark.sql.connector.catalog.{Aborted, Column,
ColumnDefaultValue, Committed, InMemoryBaseTable, InMemoryTable, TableInfo}
import org.apache.spark.sql.connector.expressions.{GeneralScalarExpression,
LiteralValue}
import org.apache.spark.sql.connector.write.MergeSummary
import org.apache.spark.sql.execution.SparkPlan
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
-import org.apache.spark.sql.execution.datasources.v2.MergeRowsExec
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
DataSourceV2ScanRelation, InsertOnlyMergeExec, MergeRowsExec}
import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec,
BroadcastNestedLoopJoinExec, CartesianProductExec}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources
@@ -2851,6 +2853,144 @@ abstract class MergeIntoTableSuiteBase extends
RowLevelOperationSuiteBase
assert(catalog.lastTransaction.isClosed)
}
+ test("SPARK-58007: merge with dynamic options on the target") {
+ withTable(sourceNameAsString) {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+ sql(s"CREATE TABLE $sourceNameAsString (pk INT NOT NULL, salary INT, dep
STRING)")
+ sql(s"INSERT INTO $sourceNameAsString VALUES (1, 150, 'hr'), (3, 300,
'hr')")
+
+ // the target options become the row-level operation options: they reach
the rewritten
+ // DataSourceV2Relation, the RowLevelOperationInfo, and the write
builder's LogicalWriteInfo
+ checkRowLevelOperationOptions(
+ sql(
+ s"""MERGE INTO $tableNameAsString t WITH (`write.split-size` = 10)
+ |USING $sourceNameAsString s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN UPDATE SET t.salary = s.salary
+ |WHEN NOT MATCHED THEN INSERT *
+ |""".stripMargin),
+ "write.split-size" -> "10")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 150, "hr") :: Row(2, 200, "software") :: Row(3, 300, "hr") ::
Nil)
+ }
+ }
+
+ test("SPARK-58007: merge with dynamic options on the source") {
+ withTable(sourceNameAsString) {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+ sql(s"CREATE TABLE $sourceNameAsString (pk INT NOT NULL, salary INT, dep
STRING)")
+ sql(s"INSERT INTO $sourceNameAsString VALUES (1, 150, 'hr'), (3, 300,
'hr')")
+
+ // the source is read-only; its options must reach the source scan the
same way a
+ // SELECT ... WITH (...) does
+ val Seq(qe) = withQueryExecutionsCaptured(spark) {
+ sql(
+ s"""MERGE INTO $tableNameAsString t
+ |USING $sourceNameAsString WITH (`split-size` = 5) s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN UPDATE SET t.salary = s.salary
+ |WHEN NOT MATCHED THEN INSERT *
+ |""".stripMargin)
+ }
+ val sourceOptions = qe.optimizedPlan.collect {
+ case r: DataSourceV2Relation => r.options
+ case s: DataSourceV2ScanRelation => s.relation.options
+ }.filter(_.containsKey("split-size"))
+ assert(sourceOptions.nonEmpty, "source relation carrying the option was
not found")
+ sourceOptions.foreach(opts => assert(opts.get("split-size") === "5"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 150, "hr") :: Row(2, 200, "software") :: Row(3, 300, "hr") ::
Nil)
+ }
+ }
+
+ test("SPARK-58007: merge with dynamic options on both the target and the
source") {
+ withTable(sourceNameAsString) {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+ sql(s"CREATE TABLE $sourceNameAsString (pk INT NOT NULL, salary INT, dep
STRING)")
+ sql(s"INSERT INTO $sourceNameAsString VALUES (1, 150, 'hr'), (3, 300,
'hr')")
+
+ // the target and source each carry their own options bag; the two must
not cross-contaminate
+ // (the single-option cases above already verify each bag reaches its
destination)
+ val Seq(qe) = withQueryExecutionsCaptured(spark) {
+ sql(
+ s"""MERGE INTO $tableNameAsString t WITH (`write.split-size` = 10)
+ |USING $sourceNameAsString WITH (`split-size` = 5) s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN UPDATE SET t.salary = s.salary
+ |WHEN NOT MATCHED THEN INSERT *
+ |""".stripMargin)
+ }
+
+ // target write relation carries only its own option, not the source's
+ 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]
+ assert(writeRelation.options.get("write.split-size") === "10", "target
relation option")
+ assert(!writeRelation.options.containsKey("split-size"), "target must
not see source option")
+
+ // source scan carries only its own option, not the target's
+ val sourceOptions = qe.optimizedPlan.collect {
+ case r: DataSourceV2Relation => r.options
+ case s: DataSourceV2ScanRelation => s.relation.options
+ }.filter(_.containsKey("split-size"))
+ assert(sourceOptions.nonEmpty, "source relation carrying the option was
not found")
+ sourceOptions.foreach { opts =>
+ assert(opts.get("split-size") === "5", "source relation option")
+ assert(!opts.containsKey("write.split-size"), "source must not see
target option")
+ }
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 150, "hr") :: Row(2, 200, "software") :: Row(3, 300, "hr") ::
Nil)
+ }
+ }
+
+ test("SPARK-58007: merge with dynamic options on the target, insert-only
fast path") {
+ withTable(sourceNameAsString) {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+ sql(s"CREATE TABLE $sourceNameAsString (pk INT NOT NULL, salary INT, dep
STRING)")
+ sql(s"INSERT INTO $sourceNameAsString VALUES (1, 150, 'hr'), (3, 300,
'hr')")
+
+ // a MATCHED-free merge is rewritten to InsertOnlyMerge (a plain
append), which builds
+ // write from the target's options at a different site than the
row-level rewrite path.
+ val executedPlan = executeAndKeepPlan {
+ sql(
+ s"""MERGE INTO $tableNameAsString t WITH (`write.split-size` = 10)
+ |USING $sourceNameAsString s
+ |ON t.pk = s.pk
+ |WHEN NOT MATCHED THEN INSERT *
+ |""".stripMargin)
+ }
+ val write = executedPlan.collectFirst {
+ case e: InsertOnlyMergeExec => e.write
+ }.getOrElse(fail("expected an InsertOnlyMergeExec in the executed plan"))
+ val append = write.toBatch.asInstanceOf[InMemoryBaseTable#Append]
+ assert(append.info.options.get("write.split-size") === "10")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, 100, "hr") :: Row(2, 200, "software") :: Row(3, 300, "hr") ::
Nil)
+ }
+ }
+
private def assertMetric(
mergeExec: MergeRowsExec,
metricName: String,
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
index 199b9ecbe0a0..5d52325da818 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/RowLevelOperationSuiteBase.scala
@@ -28,7 +28,7 @@ import
org.apache.spark.sql.catalyst.expressions.{DynamicPruningExpression, Expr
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, ReplaceData,
WriteDelta}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.METADATA_COL_ATTR_KEY
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Delete,
Identifier, InMemoryRowLevelOperationTable,
InMemoryRowLevelOperationTableCatalog, Insert, MetadataColumn, Operation,
Reinsert, Table, TableInfo, Txn, TxnTable, Update, Write}
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Column, Delete,
Identifier, InMemoryRowLevelOperationTable,
InMemoryRowLevelOperationTableCatalog, Insert, MetadataColumn, Operation,
Reinsert, RowLevelOperationWithOptions, Table, TableInfo, Txn, TxnTable,
Update, Write}
import
org.apache.spark.sql.connector.expressions.LogicalExpressions.{identity,
reference}
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.write.RowLevelOperationTable
@@ -179,6 +179,27 @@ abstract class RowLevelOperationSuiteBase
}.getOrElse(fail("couldn't find row-level operation in optimized plan"))
}
+ // 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'")
+ }
+ }
+
protected def assertNoScanPlanning(plan: LogicalPlan): Unit = {
val relations = plan.collect { case r: DataSourceV2Relation => r }
assert(relations.nonEmpty, "plan must contain relations")
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 c4c5627a12ea..adaa3be4510f 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,12 +20,9 @@ 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.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.catalog.{Aborted, Column,
ColumnDefaultValue, Committed, InMemoryTable, TableChange, TableInfo}
import org.apache.spark.sql.connector.expressions.{GeneralScalarExpression,
LiteralValue}
-import org.apache.spark.sql.connector.write.{RowLevelOperationTable,
UpdateSummary}
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.connector.write.UpdateSummary
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegerType, StringType}
@@ -1236,27 +1233,6 @@ abstract class UpdateTableSuiteBase extends
RowLevelOperationSuiteBase {
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" }
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
