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

[anuragmantri]

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 DML 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 INSERT, UPDATE, and MERGE.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added end-to-end tests to the following suites, each asserting that both references keep their own options:

• DataSourceV2OptionSuite (INSERT selecting from the same table)
• MergeIntoTableSuiteBase
• UpdateTableSuiteBase

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.

[anuragmantri]

@dongjoon-hyun @peter-toth - Please provide your review on this.

[dongjoon-hyun]

+1, LGTM. Thank you, @anuragmantri .

[peter-toth]

Thanks for the PR, @anuragmantri!

This is the holistic fix for the limitation we recorded in #57227: AnalysisContext.relationCache is keyed by (catalog + namespace + name, timeTravelSpec) and not by options, so the second reference to a table in one statement got a cache hit and silently inherited the first reference's WITH (...) options. Re-applying each reference's own options to the cached plan, instead of adding options to the cache key, is the right shape: the table is still loaded once and both references share one Table instance, so they still see the same snapshot, and it matches what the two other cache-hit paths already do (adaptCachedRelation(cached, ref) at RelationResolution.scala:518-524 and the shared-relation-cache branch at :310). The details check out - finalOptions is clearWritePrivileges'd so the internal privileges key can't leak into the relation and from there into the write options V2Writes builds, the cache entry itself is left untouched, and the newInstance() / tag handling is unchanged. Three things below.

Blocking

• 1. "Does this PR introduce any user-facing change?" is answered No: it does introduce one. Options that were silently dropped now take effect, and options that leaked onto a second reference no longer do, so existing queries can behave differently. With a connector where a read option picks a snapshot/version, SELECT ... FROM t WITH (`snapshot-id` = X) a JOIN t b used to read snapshot X on both sides and now reads the live table for b - different results, not just a different split size. Please flip it to Yes with a sentence on the delta, and say whether you think it needs a docs/sql-migration-guide.md line. Same paragraph: "This affects INSERT, UPDATE, and MERGE" is narrower than the change - any statement with two references to one table is affected, plain SELECT included.

Non-blocking

• 2. v1 relations still share the first reference's options: applyOptions only rewrites DataSourceV2Relation, so a v1 (session-catalog data-source / Hive) table keeps the cached UnresolvedCatalogRelation's options - and v1 dynamic options are live and tested (DDLSuite.scala:1394 asserts SELECT * FROM t WITH ('delimiter' = ';') changes the rows for a CSV table). One extra case covers it; otherwise the description should scope the fix to DSv2. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:370]
• 3. No test for two read references in one query: all three new tests pair a write reference with a read reference, so the base case of the bug - two reads with different options - is untested, and that is the case where the fix changes behavior in both directions. [inline: sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2OptionSuite.scala:124]

[peter-toth]

Finding 2. This only rewrites DataSourceV2Relation, so the same bug survives for v1 relations. A v1 table (session-catalog data-source or Hive table) resolves through createDataSourceV1Scan -> SessionCatalog.getRelation(v1Table, options) -> SubqueryAlias(_, UnresolvedCatalogRelation(metadata, options)) (SessionCatalog.scala:1140), that plan goes into the same relationCache, and the options on it are live: FindDataSourceTable hands them to readDataSourceTable -> DataSourceUtils.generateDatasourceOptions (DataSourceStrategy.scala:382, :259), which even rebuilds a cached HadoopFsRelation when the options differ. v1 dynamic options aren't theoretical - DDLSuite.scala:1394 ("SPARK-51747: Data source cached plan respects options if ignore conf disabled") asserts that SELECT * FROM t WITH ('delimiter' = ';') on a CSV table changes the returned rows. So INSERT INTO v1t WITH (a) SELECT * FROM v1t WITH (b), or a plain self-join on a v1 table, still ends up with one set of options for both references.

UnresolvedCatalogRelation is already imported here, so it's a one-liner:

Suggested change

	case r: DataSourceV2Relation => r.copy(options = options)
	case r: DataSourceV2Relation => r.copy(options = options)
	case r: UnresolvedCatalogRelation => r.copy(options = options)

If you'd rather keep this PR DSv2-only, could you say that in the description and leave a short note here? This PR removes the AstBuilder comment that was the only record of the limitation.

[peter-toth]

Finding 3. All three new tests pair a write reference with a read reference. The plainest instance of the bug - two read references in one query - isn't covered, and that's the case where the fix changes behavior in both directions (neither scan inherits the other's options). On master both scans come out with split-size = 5, because the second reference hits the cache. Roughly ten lines here:

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')")

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