You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/07/29 15:48:09 UTC
(spark) branch master updated: [SPARK-58330][SQL] Prevent silent dropping of dynamic options on same table references.
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 f27705db432c [SPARK-58330][SQL] Prevent silent dropping of dynamic options on same table references.
f27705db432c is described below
commit f27705db432c76119a09c7ef822927c2822eedc8
Author: Anurag Mantripragada <am...@apple.com>
AuthorDate: Wed Jul 29 08:47:58 2026 -0700
[SPARK-58330][SQL] Prevent silent dropping of dynamic options on same table references.
## What changes were proposed in this pull request?
This PR fixes dynamic table options being silently dropped when the same table is referenced more than once in a single statement. Options are now re-applied on a per-query relation cache hit, so each reference keeps its own options.
## Why are the changes needed?
The per-query relation cache in RelationResolution is keyed by catalog, namespace, name, and time travel spec, but not by options. When the same table is referenced twice in one statement with different options, the second reference gets a cache hit and reuses the first reference's options, so the options the user wrote on the second reference are silently ignored. This affects all queries that self reference a table.
## Does this PR introduce any user-facing change?
Yes. Options that were previously silently dropped on a second reference to the same table now take effect, and options that previously leaked from one reference onto another no longer do. This affects any statement with two or more references to one table.
Added a note to `docs/sql-migration-guide.md` scoped to SELECT/INSERT since those are the only two of the four statements that have shipped in a released version. UPDATE/MERGE self-reference support hasn't reached a release yet, so there's no upgrade path to document for those two.
## How was this patch tested?
Added end-to-end tests, each asserting that both references keep their own options:
- `DataSourceV2OptionSuite`: INSERT selecting from the same table, a self-join, and a streaming self-join and streaming CTE (both analysis-only, no writeStream.start())
- `MergeIntoTableSuiteBase`: self-merge (target and source referencing the same table)
- `UpdateTableSuiteBase`: a subquery and a CTE referencing the same table as the update target
- `DDLSuite`: the same three shapes (self-join, INSERT self-select, CTE) for v1 (session-catalog/Hive) tables
## 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 #57508 from anuragmantri/self-ref-options.
Lead-authored-by: Anurag Mantripragada <am...@apple.com>
Co-authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
docs/sql-migration-guide.md | 1 +
.../sql/catalyst/analysis/RelationResolution.scala | 21 ++++-
.../spark/sql/catalyst/parser/AstBuilder.scala | 4 -
.../sql/connector/DataSourceV2OptionSuite.scala | 104 ++++++++++++++++++++-
.../sql/connector/MergeIntoTableSuiteBase.scala | 43 +++++++++
.../spark/sql/connector/UpdateTableSuiteBase.scala | 85 +++++++++++++++++
.../spark/sql/execution/command/DDLSuite.scala | 52 +++++++++++
7 files changed, 304 insertions(+), 6 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 423c2fdd0d88..6efed7224c0d 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -37,6 +37,7 @@ license: |
- Since Spark 4.3, the new `COMMENT ON COLUMN ... IS NULL` syntax removes a column comment by passing a `null` comment to `TableChange.updateColumnComment(String[], String)`, so a `UpdateColumnComment` table change may now carry a `null` `newComment()`. Previously `newComment()` was always non-null. DataSource V2 catalogs that handle `UpdateColumnComment` should null-check `newComment()` and treat `null` as "remove the column comment" (mirroring how `UpdateColumnDefaultValue` already car [...]
- Since Spark 4.3, `unix_seconds`, `unix_millis`, and `unix_micros` accept `TIMESTAMP_NTZ` and the nanosecond-precision timestamp types directly, reading them with no time-zone shift. Previously these functions accepted only `TIMESTAMP_LTZ`; a `TIMESTAMP_NTZ` or nanosecond-timestamp argument was rejected with a `DATATYPE_MISMATCH` error. This is a new capability and does not change the result of any query that previously succeeded.
- Since Spark 4.3, `hash()` and `xxhash64()` include the `days` field of `CalendarInterval` when computing the hash, so their output for interval values differs from earlier releases. Previously the codegen path dropped `days`, disagreeing with interpreted evaluation.
+- Since Spark 4.3, when a `SELECT` or `INSERT` statement references the same table more than once with different `WITH (...)` options (for example a self-join, or `INSERT INTO t WITH (...) SELECT * FROM t WITH (...)`), each reference now uses its own options instead of the second reference silently inheriting the first reference's options via the analyzer's relation cache.
## Upgrading from Spark SQL 4.1 to 4.2
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
index 5bd1256b631f..0a085fcc2971 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala
@@ -239,6 +239,7 @@ class RelationResolution(
val key = toCacheKey(catalog, ident, finalTimeTravelSpec)
val planId = u.getTagValue(LogicalPlan.PLAN_ID_TAG)
val writePrivileges = u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES)
+ val finalOptions = u.clearWritePrivileges.options
// A reference that requires write privileges is never served from the per-query relation
// cache. The catalog authorizes the write in `loadTable(ident, writePrivileges)` below, and
// a cache hit would skip that call entirely. The hit happens whenever the write target is
@@ -247,9 +248,14 @@ class RelationResolution(
// for `INSERT INTO t SELECT * FROM t`.
val cached = if (writePrivileges == null) relationCache.get(key) else None
cached
+ // The per-query relation cache is not keyed by options. When the same table is referenced
+ // more than once in a single statement with different dynamic options (e.g. a self-join,
+ // or a second reference sharing the target's cache entry), a cache hit would otherwise
+ // reuse the first reference's options and silently drop this reference's. Re-apply this
+ // reference's options to the cached relation so each reference honors its own bag.
+ .map(applyOptions(_, finalOptions))
.map(adaptCachedRelation(_, planId))
.orElse {
- val finalOptions = u.clearWritePrivileges.options
// For a `RelationCatalog` with no time-travel / write privileges, the single-RPC
// `loadRelation` answers both "is there a table?" and "is there a view?" in one
// call. Time-travel and write privileges apply to tables only, so for those the
@@ -364,6 +370,19 @@ class RelationResolution(
CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident, table, conf)
}
+ /**
+ * Re-applies `options` to the relation in a cached plan. Every `relationCache` entry holds a
+ * single relation for its own identifier (a view's body is still unresolved when it is cached),
+ * so this cannot reach another table's relation.
+ */
+ private def applyOptions(
+ cached: LogicalPlan,
+ options: CaseInsensitiveStringMap): LogicalPlan = cached transform {
+ case r: DataSourceV2Relation => r.copy(options = options)
+ case r: UnresolvedCatalogRelation => r.copy(options = options)
+ case r: StreamingRelationV2 => r.copy(extraOptions = options)
+ }
+
private def adaptCachedRelation(cached: LogicalPlan, planId: Option[Long]): LogicalPlan = {
val plan = cached transform {
case multi: MultiInstanceRelation =>
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 950a22e1d840..636bee327730 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
@@ -1284,10 +1284,6 @@ class AstBuilder extends DataTypeAstBuilder
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, Option(ctx.sourceOptions))
} else if (ctx.sourceQuery != null) {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
index 803dd35513f4..f785cac9c012 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala
@@ -20,7 +20,8 @@ package org.apache.spark.sql.connector
import org.apache.spark.sql.{AnalysisException, Row}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.plans.logical._
-import org.apache.spark.sql.connector.catalog.InMemoryBaseTable
+import org.apache.spark.sql.catalyst.streaming.StreamingRelationV2
+import org.apache.spark.sql.connector.catalog.{InMemoryBaseTable, InMemoryRowLevelOperationTableCatalog}
import org.apache.spark.sql.execution.CommandResultExec
import org.apache.spark.sql.execution.datasources.v2._
import org.apache.spark.sql.functions.lit
@@ -121,6 +122,107 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase {
}
}
+ test("SPARK-58330: dynamic options are not lost when INSERT selects from the same table") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+
+ // The target and the source are the same table, so they share one per-query relation-cache
+ // entry. Each reference must keep its own options: the target's write options must not be
+ // dropped by the source reference, and the source's read options must not leak to the target.
+ val df = sql(s"INSERT INTO $t1 WITH (`write.split-size` = 10) " +
+ s"SELECT id, data FROM $t1 WITH (`split-size` = 5)")
+
+ val appendData = df.queryExecution.optimizedPlan.collectFirst {
+ case CommandResult(_, a: AppendData, _, _) => a
+ }.getOrElse(fail("expected an AppendData in the optimized plan"))
+
+ // target: the write relation keeps its own option and does not pick up the source's
+ val targetOptions = appendData.table.asInstanceOf[DataSourceV2Relation].options
+ assert(targetOptions.get("write.split-size") === "10", "target write option")
+ assert(!targetOptions.containsKey("split-size"), "target must not see source option")
+
+ // source: the read relation under the query keeps its own option and does not pick up
+ // the target's
+ val sourceOptions = appendData.query.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 read option")
+ assert(!opts.containsKey("write.split-size"), "source must not see target option")
+ }
+
+ checkAnswer(sql(s"SELECT * FROM $t1"),
+ Seq(Row(1, "a"), Row(2, "b"), Row(1, "a"), Row(2, "b")))
+ }
+ }
+
+ test("SPARK-58330: each reference in a self-join keeps its own dynamic options") {
+ val t1 = s"${catalogAndNamespace}table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+ sql(s"INSERT INTO $t1 VALUES (1, 'a'), (2, 'b')")
+
+ // Both references are reads of the same table with different options, so they share one
+ // per-query relation-cache entry. Neither scan should inherit the other's options.
+ val df = sql(s"SELECT a.id FROM $t1 WITH (`split-size` = 5) a " +
+ s"JOIN $t1 WITH (`split-size` = 9) b ON a.id = b.id")
+
+ val splitSizes = df.queryExecution.optimizedPlan.collect {
+ case s: DataSourceV2ScanRelation => s.relation.options.get("split-size")
+ }
+ assert(splitSizes.sorted === Seq("5", "9"))
+
+ checkAnswer(df, Seq(Row(1), Row(2)))
+ }
+ }
+
+ test("SPARK-58330: each reference in a streaming self-join keeps its own dynamic options") {
+ withSQLConf(
+ "spark.sql.catalog.streamcat" -> classOf[InMemoryRowLevelOperationTableCatalog].getName) {
+ val t1 = "streamcat.ns.table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+
+ // Both references are streaming reads of the same table with different options, so they
+ // share one per-query relation-cache entry. Neither should inherit the other's options.
+ // The plan is only analyzed here, never executed (no writeStream.start()), so no actual
+ // streaming query runs.
+ val df = sql(s"SELECT a.id FROM STREAM $t1 WITH (`split-size` = 5) a " +
+ s"JOIN STREAM $t1 WITH (`split-size` = 9) b ON a.id = b.id")
+
+ val splitSizes = df.queryExecution.analyzed.collect {
+ case r: StreamingRelationV2 => r.extraOptions.get("split-size")
+ }
+ assert(splitSizes.sorted === Seq("5", "9"))
+ }
+ }
+ }
+
+ test("SPARK-58330: a streaming CTE referencing the same table keeps its own dynamic options") {
+ withSQLConf(
+ "spark.sql.catalog.streamcat" -> classOf[InMemoryRowLevelOperationTableCatalog].getName) {
+ val t1 = "streamcat.ns.table"
+ withTable(t1) {
+ sql(s"CREATE TABLE $t1 (id bigint, data string)")
+
+ // The CTE's inner scan of `STREAM t WITH (...)` and the outer `STREAM t WITH (...)`
+ // reference resolve to the same per-query relation-cache entry; CTE substitution must
+ // not let one leak into the other. Analysis-only, as above.
+ val df = sql(s"WITH x AS (SELECT id FROM STREAM $t1 WITH (`split-size` = 5)) " +
+ s"SELECT a.id FROM STREAM $t1 WITH (`split-size` = 9) a CROSS JOIN x")
+
+ val splitSizes = df.queryExecution.analyzed.collect {
+ case r: StreamingRelationV2 => r.extraOptions.get("split-size")
+ }
+ assert(splitSizes.sorted === Seq("5", "9"))
+ }
+ }
+ }
+
test("SPARK-50286: Propagate options for DataFrameWriter Append") {
val t1 = s"${catalogAndNamespace}table"
withTable(t1) {
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 9b625c1b81d2..de6c45006911 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
@@ -2960,6 +2960,49 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase
}
}
+ test("SPARK-58330: self-merge keeps each reference's own dynamic options") {
+ createAndInitTable("pk INT NOT NULL, salary INT, dep STRING",
+ """{ "pk": 1, "salary": 100, "dep": "hr" }
+ |{ "pk": 2, "salary": 200, "dep": "software" }
+ |""".stripMargin)
+
+ // Target and source are the same table, so they share one per-query relation-cache entry.
+ // Each reference must keep its own options: the target keeps its write options and the source
+ // keeps its read options; neither is dropped or inherited from the other.
+ val Seq(qe) = withQueryExecutionsCaptured(spark) {
+ sql(
+ s"""MERGE INTO $tableNameAsString t WITH (`write.split-size` = 10)
+ |USING $tableNameAsString WITH (`split-size` = 5) s
+ |ON t.pk = s.pk
+ |WHEN MATCHED THEN UPDATE SET t.salary = s.salary
+ |""".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, 100, "hr") :: Row(2, 200, "software") :: 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",
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 adaa3be4510f..b54445466d7e 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.QueryTest.withQueryExecutionsCaptured
+import org.apache.spark.sql.catalyst.plans.logical.{ReplaceData, WriteDelta}
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.UpdateSummary
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegerType, StringType}
@@ -1265,6 +1268,46 @@ abstract class UpdateTableSuiteBase extends RowLevelOperationSuiteBase {
Row(1, -1, "hr") :: Row(2, 200, "software") :: Row(3, -1, "hr") :: Nil)
}
+ test("SPARK-58330: update keeps its own options when a subquery references the same table " +
+ "with different options") {
+ 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)
+
+ // The target and the WHERE subquery reference the same table with different options; they
+ // share one per-query relation-cache entry. Each reference must keep its own options.
+ val Seq(qe) = withQueryExecutionsCaptured(spark) {
+ sql(s"UPDATE $tableNameAsString WITH (`write.split-size` = 10) SET salary = -1 " +
+ s"WHERE pk IN (SELECT pk FROM $tableNameAsString WITH (`split-size` = 5) WHERE dep = 'hr')")
+ }
+
+ // target write relation carries only its own option
+ 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 subquery option")
+
+ // the subquery scan carries only its own option
+ val subqueryOptions = qe.optimizedPlan.collect {
+ case r: DataSourceV2Relation => r.options
+ case s: DataSourceV2ScanRelation => s.relation.options
+ }.filter(_.containsKey("split-size"))
+ assert(subqueryOptions.nonEmpty, "subquery relation carrying the option was not found")
+ subqueryOptions.foreach { opts =>
+ assert(opts.get("split-size") === "5", "subquery relation option")
+ assert(!opts.containsKey("write.split-size"), "subquery must not see target option")
+ }
+
+ 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" }
@@ -1282,4 +1325,46 @@ abstract class UpdateTableSuiteBase extends RowLevelOperationSuiteBase {
sql(s"SELECT * FROM $tableNameAsString"),
Row(1, -1, "hr") :: Row(2, 200, "software") :: Row(3, -1, "hr") :: Nil)
}
+
+ test("SPARK-58330: update keeps its own options when a CTE references the same table " +
+ "with different options") {
+ 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)
+
+ // The target and the CTE reference the same table with different options; they share one
+ // per-query relation-cache entry. Each reference must keep its own options.
+ val Seq(qe) = withQueryExecutionsCaptured(spark) {
+ sql(s"WITH hr_rows AS (SELECT pk FROM $tableNameAsString WITH (`split-size` = 5) " +
+ s"WHERE dep = 'hr') " +
+ s"UPDATE $tableNameAsString WITH (`write.split-size` = 10) SET salary = -1 " +
+ s"WHERE pk IN (SELECT pk FROM hr_rows)")
+ }
+
+ // target write relation carries only its own option
+ 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 CTE option")
+
+ // the CTE scan carries only its own option
+ val cteOptions = qe.optimizedPlan.collect {
+ case r: DataSourceV2Relation => r.options
+ case s: DataSourceV2ScanRelation => s.relation.options
+ }.filter(_.containsKey("split-size"))
+ assert(cteOptions.nonEmpty, "CTE relation carrying the option was not found")
+ cteOptions.foreach { opts =>
+ assert(opts.get("split-size") === "5", "CTE relation option")
+ assert(!opts.containsKey("write.split-size"), "CTE must not see target option")
+ }
+
+ checkAnswer(
+ sql(s"SELECT * FROM $tableNameAsString"),
+ Row(1, -1, "hr") :: Row(2, 200, "software") :: Row(3, -1, "hr") :: Nil)
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
index a52baaeac42b..2bcbe4d375c9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala
@@ -27,6 +27,7 @@ import org.apache.hadoop.fs.permission.{AclEntry, AclStatus}
import org.apache.spark.{SparkClassNotFoundException, SparkException, SparkFiles, SparkRuntimeException, SparkUnsupportedOperationException}
import org.apache.spark.internal.config
import org.apache.spark.sql.{AnalysisException, QueryTest, Row, SaveMode}
+import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.{FunctionIdentifier, QualifiedTableName, TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.TempTableAlreadyExistsException
import org.apache.spark.sql.catalyst.catalog._
@@ -1472,6 +1473,57 @@ abstract class DDLSuite extends QueryTest with DDLSuiteBase {
}
}
+ test("SPARK-58330: self-reference to a v1 table keeps each reference's own dynamic options") {
+ withTable("t") {
+ spark.sql("CREATE TABLE t(a string, b string) USING CSV")
+ spark.sql("INSERT INTO TABLE t VALUES ('a;b', 'c')")
+
+ checkAnswer(
+ sql("SELECT x.a, x.b, y.a, y.b FROM t x CROSS JOIN t WITH ('delimiter' = ';') y"),
+ Row("a;b", "c", "a", "b,c") :: Nil
+ )
+ }
+ }
+
+ test("SPARK-58330: INSERT keeps its own options when selecting from the same v1 table") {
+ withTable("t") {
+ spark.sql("CREATE TABLE t(a string, b string) USING CSV")
+ spark.sql("INSERT INTO TABLE t VALUES ('a;b', 'c')")
+
+ // The target and the source are the same session-catalog table, so they share one
+ // per-query relation-cache entry. The target's write option must not be dropped by the
+ // source reference, and the source's read option must not leak to the target.
+ val Seq(qe) = withQueryExecutionsCaptured(spark) {
+ sql("INSERT INTO t WITH ('sep' = '|') SELECT a, b FROM t WITH ('sep' = ';')")
+ }
+
+ val targetOptions = qe.analyzed.collectFirst {
+ case cmd: InsertIntoHadoopFsRelationCommand => cmd.options
+ }.getOrElse(fail("target InsertIntoHadoopFsRelationCommand not found"))
+ val sourceOptions = qe.analyzed.collect {
+ case LogicalRelation(fsRelation: HadoopFsRelation, _, _, _, _) => fsRelation.options
+ }.find(_.get("sep").contains(";")).getOrElse(fail("source relation with sep=; not found"))
+ assert(targetOptions.get("sep").contains("|"), "target write option")
+ assert(sourceOptions("sep") === ";", "source read option")
+ }
+ }
+
+ test("SPARK-58330: a CTE referencing the same v1 table keeps its own dynamic options") {
+ withTable("t") {
+ spark.sql("CREATE TABLE t(a string, b string) USING CSV")
+ spark.sql("INSERT INTO TABLE t VALUES ('a;b', 'c')")
+
+ // `t` (no options) and the CTE's inner scan of `t WITH ('delimiter' = ';')` resolve to
+ // the same per-query relation-cache entry; CTE substitution must not let one leak into
+ // the other.
+ checkAnswer(
+ sql("WITH x AS (SELECT a, b FROM t WITH ('delimiter' = ';')) " +
+ "SELECT t.a, t.b, x.a, x.b FROM t CROSS JOIN x"),
+ Row("a;b", "c", "a", "b,c") :: Nil
+ )
+ }
+ }
+
test("SPARK-18009 calling toLocalIterator on commands") {
import scala.jdk.CollectionConverters._
val df = sql("show databases")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org