[SPARK-59163][SQL] Reuse projected table-state options

[yyanyy]

What changes were proposed in this pull request?

This is tracked by SPARK-59163. It follows
#57585 and completes the table-state option reuse requested in
this review comment. The
earlier PR introduced the projection and has already landed; this PR is its subsequent internal
cleanup.

RelationResolution already computed the projected table-state options for TableCacheKey, but
table loading projected the full option map again. The projected map passed to
CatalogV2Util.lookupCachedRelation was also projected a second time inside that method.

This PR:

• as a minor refactoring, adds consistently named *WithStateOptions entry points and makes the
  existing full-option helpers project once before delegating to them;
• reuses TableCacheKey.stateOptions for shared relation cache lookup in both persistent relation
  and V2TableReference resolution, without repeating the projection for table loading;
• reuses the single projection in V2TableRefreshUtil for refresh deduplication and shared cache
  lookup, without projecting the full option map again for catalog loading;
• passes catalog code a defensive copy of projected state options, so mutation through
  CaseInsensitiveStringMap's collection views cannot change Spark-owned cache or deduplication
  keys.

Why are the changes needed?

The previous flow repeatedly normalized the catalog's state-option key set and materialized a full
projection during an uncached relation resolution or execution refresh. The new flow reuses each
projection for Spark's table pin or refresh deduplication and shared cache lookup, while passing an
equivalent defensive copy to catalog loading. This removes repeated projection without exposing a
map whose hash and equality Spark relies on to connector mutation.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

CatalogV2UtilSuite, DataSourceV2OptionSuite, and PlanResolutionSuite cover table-state option
filtering, persistent relation resolution, V2TableReference resolution, shared relation cache
matching, and execution refresh. Focused call-count assertions verify that already-projected paths
do not project again, and a mutation regression test verifies that catalog code cannot change a
retained cache key through mutable collection views.

XDG_RUNTIME_DIR=/tmp/yan-sbt-runtime \
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 \
DEFAULT_ARTIFACT_REPOSITORY=https://maven-proxy.cloud.databricks.com \
MAVEN_MIRROR_URL=https://maven-proxy.cloud.databricks.com \
build/sbt \
  'catalyst/testOnly org.apache.spark.sql.connector.catalog.CatalogV2UtilSuite' \
  'sql/testOnly org.apache.spark.sql.connector.DataSourceV2OptionSuite' \
  'sql/testOnly org.apache.spark.sql.execution.command.PlanResolutionSuite'

CatalogV2UtilSuite ran 20/20 tests, DataSourceV2OptionSuite ran 56/56 tests, and
PlanResolutionSuite ran 96/96 tests, with no failures, cancellations, ignores, or pending tests.
No test-case filters were used within the suites; the run used SBT's incremental compilation cache.
DataSourceV2OptionSuite and PlanResolutionSuite emitted non-failing possible thread-leak
warnings.

XDG_RUNTIME_DIR=/tmp/yan-sbt-runtime \
JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 \
DEFAULT_ARTIFACT_REPOSITORY=https://maven-proxy.cloud.databricks.com \
MAVEN_MIRROR_URL=https://maven-proxy.cloud.databricks.com \
build/sbt \
  'catalyst / Compile / scalastyle' \
  'catalyst / Test / scalastyle' \
  'sql / Compile / scalastyle' \
  'sql / Test / scalastyle' \
  'catalyst / Compile / checkstyle'

The four scalastyle tasks processed 718 Catalyst main files, 440 Catalyst test files, 826 SQL core
main files, and 1,106 SQL core test files with no errors or warnings. Catalyst main checkstyle also
completed with no issues.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex

[cloud-fan]

Review summary

The new loading path passes Spark's mutable table-state cache-key map directly to connector code; a connector can mutate it through live collection views and break state-correct cache behavior. Pass a defensive copy at the external catalog boundary while retaining the original projection internally. Also add call-count coverage so the single-projection optimization cannot regress without a test failure.

Findings

2 total: 0 P0, 1 P1, 1 P2, 0 P3.

Blocking (P1)

• Keep the table-state cache key private from connector mutation — sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala:548 — see inline.

Non-blocking (P2)

• Add a regression assertion for the single-projection contract — sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala:650 — see inline.

[cloud-fan]

Blocking (P1): This passes the same stateOptions instance that Spark retains in TableCacheKey/currentTables to arbitrary catalog code. Although direct mutators throw, CaseInsensitiveStringMap exposes mutable live keySet, values, and entrySet views, so a connector can change the map during or after loading and corrupt the hash/key Spark relies on. That can produce cache misses or let a table loaded for one branch be reused for another. Please keep Spark's projected map private and pass loadTable a defensive copy.

Recommended change: Keep the caller-projected map as Spark-private state and construct a defensive CaseInsensitiveStringMap copy immediately before invoking TableCatalog.loadTable; add a focused mutation regression test.

Why this works: External catalog code receives an equivalent copy, so mutation through a collection view cannot alter the map used for table pinning, refresh deduplication, or shared-cache lookup.

Scope: CatalogV2Util.getTableWithStateOptions and a focused Catalyst catalog test.

Compatibility: Catalogs receive the same option contents and case-insensitive lookup behavior, public signatures remain unchanged, and callers still perform only one projection.

Risks: Adds one small map copy at the external catalog boundary.

Constraints: Retain the original projected map for every Spark-owned cache and deduplication key. Do not expose that original instance to the connector callback.

Success: A catalog that mutates the supplied options through a collection view cannot change Spark's retained state key, and subsequent same-state and different-state lookups remain correct.

[yyanyy]

Thanks for calling this out! I noticed this earlier but initially decided not to add a copy since it's not very easy to mutate CaseInsensitiveStringMap without some explicit tricks, as the direct mutators throw as called out, and also other callers passing maps to connectors do not do this defensive copy either. However, this path indeed is more delicate since Spark retains the map as a cache for deduplication key, so I agree it should be protected.

I now create a defensive copy immediately before TableCatalog.loadTable and added a test.
I also considered limiting the copy only to callers that retain the map, but that would require splitting or adding plumbing to the loading path merely to avoid a small map copy, which kind of defeat half of the purpose of this PR; so I decided to keep the uniform copy in the common helper for readability, despite that there are code paths that would perform an unnecessary copy; please let me know if this is fine, or a more limited copy approach is still preferred; or I can even follow up to make CaseInsensitiveStringMap to be truly immutable in all possible cases.

[cloud-fan]

Non-blocking (P2): The existing tests verify the projected contents and final cache/load results, but none counts tableStateOptionKeys calls. Replacing this already-projected path with a full-options wrapper would therefore keep every assertion green while reintroducing the repeated projection this PR is meant to remove. Please use a resettable counting catalog in PlanResolutionSuite and DataSourceV2OptionSuite to assert one projection per resolution or refresh boundary alongside the existing semantic checks.

[cloud-fan]

Review summary

The single-projection refactoring and the new non-aliasing boundary are sound for ordinary option maps, and the earlier mutation and call-count review items are addressed. One blocking edge case remains: replaying duplicate case variants through asCaseSensitiveMap() can make the connector observe a different effective state value from the cache key. Construct the defensive copy from stateOptions itself and add a duplicate-case regression.

Findings

1 total: 0 P0, 1 P1, 0 P2, 0 P3.

Blocking (P1)

• Copy the effective case-insensitive entries — sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala:551 — see inline.

Re-review status

2 addressed, 0 remaining, 1 new to this AI review.

New attribution: 1 newly introduced, 0 late catch, 0 previously raised, 0 unattributed.

Remaining findings

No prior AI findings remain.

[cloud-fan]

Blocking (P1): stateOptions.asCaseSensitiveMap() can contain both original spellings for a case-insensitive key, while lookup and equality use one normalized delegate value. Reconstructing from the case-sensitive map replays those duplicates in a fresh map order, so the copy can choose a different effective value. A Connect named-table read can carry both snapshot and SNAPSHOT; Spark may then load one snapshot but cache it under the other state key, allowing a later unambiguous reference to reuse the wrong table. Please copy from the effective entries (for example, new CaseInsensitiveStringMap(stateOptions)) and add a duplicate-case regression.