[SPARK-59163][SQL] Reuse projected options and make CaseInsensitiveStringMap read-only

[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;
• makes CaseInsensitiveStringMap consistently read-only, including keySet, values, and
  entrySet, so catalog loading can safely consume the same projected instance used by Spark's
  cache and 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 catalog loading, Spark's table pin or refresh deduplication, and shared cache lookup.
Because that same instance may be retained as a cache or deduplication key, it must remain stable
after being passed to connector code. CaseInsensitiveStringMap already rejected direct mutation;
making its normalized map unmodifiable also rejects mutation through collection views. This keeps
the reuse safe without reconstructing the map, preserving its original-case entries and the
effective value already selected for keys that differ only by case.

Does this PR introduce any user-facing change?

Yes, narrowly for connector implementations. Mutation through CaseInsensitiveStringMap's
collection views now throws UnsupportedOperationException, consistent with its existing direct
mutators. Read behavior is unchanged.

How was this patch tested?

CaseInsensitiveStringMapSuite, CatalogV2UtilSuite, DataSourceV2OptionSuite, and
PlanResolutionSuite cover collection-view immutability, duplicate-case preservation, 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 catalog-side mutation assertions verify that retained cache keys
cannot be changed through collection views. Five additional SQL suites cover existing connector
option-map consumers.

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.util.CaseInsensitiveStringMapSuite org.apache.spark.sql.connector.catalog.CatalogV2UtilSuite' \
  'sql/testOnly org.apache.spark.sql.connector.DataSourceV2OptionSuite org.apache.spark.sql.execution.command.PlanResolutionSuite org.apache.spark.sql.connector.V1WriteFallbackSuite org.apache.spark.sql.connector.AppendDataTransactionSuite org.apache.spark.sql.connector.SupportsCatalogOptionsSuite org.apache.spark.sql.execution.datasources.v2.FileTableSuite org.apache.spark.sql.sources.ExternalCommandRunnerSuite'

The nine-suite run completed 243/243 tests: CaseInsensitiveStringMapSuite 9/9,
CatalogV2UtilSuite 21/21, DataSourceV2OptionSuite 56/56, PlanResolutionSuite 96/96,
V1WriteFallbackSuite 13/13, AppendDataTransactionSuite 19/19,
SupportsCatalogOptionsSuite 20/20, FileTableSuite 8/8, and ExternalCommandRunnerSuite 1/1.
There were no failures, cancellations, ignores, or pending tests. The run selected these nine suites
but used no test-case filters within them; SBT's incremental compilation cache was used. Several
SQL suites emitted non-failing possible thread-leak warnings.

After the final asCaseSensitiveMap entry-mutation assertion was added, the focused suite was
rerun:

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.util.CaseInsensitiveStringMapSuite'

CaseInsensitiveStringMapSuite ran 9/9 tests, with no failures, cancellations, ignores, or pending
tests. No test-case filter was used; the run used SBT's incremental compilation cache.

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.

[yyanyy]

Thanks, good catch. I first switched the defensive copy to new CaseInsensitiveStringMap(stateOptions) and added duplicate-case coverage. That preserves the effective normalized value, but revealed another regression: iterating stateOptions exposes lowercased keys, so rebuilding from it loses the original spellings that asCaseSensitiveMap() should retain. In short, copying from asCaseSensitiveMap() preserves casing but can change the effective duplicate-case value, while copying from stateOptions preserves the value but loses casing.

Rather than introduce special copy machinery to reconstruct both representations, I decided to fix the underlying mutability gap forCaseInsensitiveStringMap, so that we can safely pass the original stateOptions instance without copying, since the change for that seems to not be a lot.