From 90596ef91570479e13daf650e95d0b98ba95f0b2 Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Fri, 24 Jul 2026 13:04:21 -0700 Subject: [PATCH 1/3] [SPARK-58330] [SQL] Prevent silent dropping of dynamic options on same table references in DML --- .../analysis/RelationResolution.scala | 14 ++- .../sql/catalyst/parser/AstBuilder.scala | 4 - .../connector/DataSourceV2OptionSuite.scala | 38 +++++++++ .../connector/MergeIntoTableSuiteBase.scala | 43 ++++++++++ .../sql/connector/UpdateTableSuiteBase.scala | 85 +++++++++++++++++++ 5 files changed, 179 insertions(+), 5 deletions(-) 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 22bed3fbe7699..88f2f693154dd 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 @@ -238,12 +238,18 @@ class RelationResolution( case CatalogAndIdentifier(catalog, ident) => val key = toCacheKey(catalog, ident, finalTimeTravelSpec) val planId = u.getTagValue(LogicalPlan.PLAN_ID_TAG) + val finalOptions = u.clearWritePrivileges.options relationCache .get(key) + // 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. + // `INSERT INTO t WITH (a) SELECT * FROM t`), 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 writePrivileges = u.options.get(UnresolvedRelation.REQUIRED_WRITE_PRIVILEGES) - 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 @@ -358,6 +364,12 @@ class RelationResolution( CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident, table, conf) } + private def applyOptions( + cached: LogicalPlan, + options: CaseInsensitiveStringMap): LogicalPlan = cached transform { + case r: DataSourceV2Relation => r.copy(options = 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 26d88c07af5c8..f28bb636de273 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 @@ -1283,10 +1283,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 803dd35513f45..edd43e70f8c9a 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 @@ -121,6 +121,44 @@ 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-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 9b625c1b81d24..de6c450069115 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 adaa3be4510fe..b54445466d7ef 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) + } } From d1b4a44e23d25e384697c46492a63ef4787586b1 Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Sun, 26 Jul 2026 21:23:08 -0700 Subject: [PATCH 2/3] Address review comments --- .../analysis/RelationResolution.scala | 1 + .../connector/DataSourceV2OptionSuite.scala | 20 +++++++ .../sql/execution/command/DDLSuite.scala | 52 +++++++++++++++++++ 3 files changed, 73 insertions(+) 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 88f2f693154dd..2ced6f40b9496 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 @@ -368,6 +368,7 @@ class RelationResolution( cached: LogicalPlan, options: CaseInsensitiveStringMap): LogicalPlan = cached transform { case r: DataSourceV2Relation => r.copy(options = options) + case r: UnresolvedCatalogRelation => r.copy(options = options) } private def adaptCachedRelation(cached: LogicalPlan, planId: Option[Long]): LogicalPlan = { 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 edd43e70f8c9a..01595cd6b491d 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 @@ -159,6 +159,26 @@ class DataSourceV2OptionSuite extends DatasourceV2SQLBase { } } + 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-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/execution/command/DDLSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLSuite.scala index a52baaeac42b9..2bcbe4d375c9b 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") From 6a7bfc42ef4b1e2da3f9db6c2359e011b45eed35 Mon Sep 17 00:00:00 2001 From: Anurag Mantripragada Date: Sun, 26 Jul 2026 21:30:00 -0700 Subject: [PATCH 3/3] Add migration guide --- docs/sql-migration-guide.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md index 423c2fdd0d885..6efed7224c0d5 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 carries a `null` value to drop a default). - 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