[SPARK-58330][SQL] Prevent silent dropping of dynamic options on same table references in DML

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 =>

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) {

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) {

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",

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)
+  }
 }