[SPARK-58389][SQL][FOLLOWUP] Pin DSv2 table instances by state options during analysis

[yyanyy]

What changes were proposed in this pull request?

SPARK-58389 changed DSv2 relation caching to include all read options. This is required
to preserve each reference's complete option bag, but it also means references that differ
only in scan-specific options can independently call loadTable and receive different
concrete table versions within one query.

This PR separates relation reuse from table-state pinning:

• The existing relationCache remains keyed by all read options and reuses a finalized relation
  only when the complete option bags match.
• A new query-scoped tableCache is keyed by catalog, identifier, time travel, and table-state
  options. References with the same table-state key reuse one concrete Table, while retaining
  their own complete options in their relations.
• References with different table-state options do not share a table pin.
• sharedRelationCache and CacheManager reuse continue to require complete option matching.

This PR adds the evolving SupportsTableStateOptions catalog capability so a catalog can declare
which raw options may affect the table state selected by loadTable, such as a branch, tag,
snapshot, or version. Catalogs that do not implement the capability are handled conservatively:
all raw options are considered table-state-affecting.

While applying the same table-pinning model to cacheable V2TableReference resolution, this PR
also fixes two existing gaps in getOrLoadRelation:

• It previously called loadTable(identifier) without passing the options captured in the table
  reference. A table-cache miss now uses the options-aware catalog API with the reference's
  complete option bag.
• It previously did not consult sharedRelationCache. Temporary-view re-resolution now consults
  sharedRelationCache while establishing the initial table pin, allowing it to preserve a
  Table already pinned through CacheManager.

Transaction references still use the Table loaded through the transaction catalog and do not
consult sharedRelationCache. Write targets remain non-cacheable and bypass the query-scoped read
caches.

The resulting resolution flow is:

1. Check relationCache using the full-option relation key.
2. On a miss, check tableCache using the table-state key.
3. On a tableCache hit, construct a relation from the pinned Table and the current reference's
   complete options. Do not call loadTable or consult sharedRelationCache.
4. On a tableCache miss, load the current Table through the applicable options-aware path.
5. Where sharedRelationCache lookup applies, reuse its relation only when table identity and all
   options match. The shared cached Table establishes the initial pin on a match; otherwise, the
   newly loaded Table establishes it.
6. Store the pinned Table in tableCache and the finalized relation in relationCache.

Execution-time table refresh uses the same table-state option projection to preserve this
first-resolution-wins behavior.

Why are the changes needed?

A catalog may accept both table-state options and scan-specific options. Using the complete option
bag for relation reuse is necessary, but using it as the only level of caching can cause references
in the same table-state domain to load different concrete table versions during one query.

The new tableCache pins one concrete Table per state key without weakening full-option matching
for finalized relations, sharedRelationCache, or CacheManager.

Forwarding options from V2TableReference is also necessary because those options may select the
table state being reloaded. Temporary views additionally need the sharedRelationCache bridge to
preserve a CacheManager-pinned Table.

Different state domains, including different parsed time-travel specifications, continue to
resolve and pin independently.

Does this PR introduce any user-facing change?

Yes, for catalog implementors only.

This adds the evolving SupportsTableStateOptions catalog capability. There is no new SQL syntax
or configuration, and catalogs that do not implement it retain the conservative behavior where all
raw options are considered table-state-affecting.

The table-consistency fix otherwise addresses behavior introduced on the unreleased master branch
by SPARK-58389.

How was this patch tested?

Added regression coverage for table-state projection and pinning, conservative fallback,
sharedRelationCache and CacheManager matching, nested analysis, execution refresh, and
temporary-view, transaction, and write-target V2TableReference behavior.

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

Generated-by: OpenAI Codex

[yyanyy]

The other case here for now is TransactionContext that shouldn't use shared relation cache to avoid replacing the table earlier than the appropriate transaction check to decide if a cache reuse is safe (txn.registerScans?)

[uros-b]

@szehon-ho Could you PTAL at these changes?

[aokolnychyi]

Well, I think we historically treated it the opposite way. Specifically, we didn't support any read options that defined the state. Therefore, we shared the table instance across all of them. We can't silently switch this behavior in 4.3. In fact, I think if the catalog doesn't implement this method, it should mean it does NOT support any state defining options. Otherwise, we will UNPIN versions within one query UNLESS the catalog implements the new mix-in interface that it may not know about. For example, we will break Iceberg in 4.3 unless we also change the connector code.

[aokolnychyi]

What about this description?

/**
 * A catalog capability for identifying options that affect table state such as branch or tag.
 * <p>
 * Spark may need to resolve the same table more than once while analyzing or refreshing a query.
 * By default, Spark reuses a single table instance for all references to the same identifier,
 * treating options as unable to select a different table state. A catalog can implement this
 * interface to declare which raw options may cause the catalog to select a different table state,
 * such as a branch, tag, snapshot, or version. Spark then reuses one table instance only for
 * references whose table-state options match, while preserving every reference's complete option
 * map for scan planning.
 * <p>
 * Catalogs that do not implement this interface are assumed to have no state-affecting options, so
 * all references to the same identifier continue to share a single table instance regardless of
 * their read or write options.
 * <p>
 * Option key matching is case-insensitive. Option values remain case-sensitive. State that Spark
 * parses and handles independently, such as time travel, must not be included in the returned set.
 *
 * @since 4.3.0
 */
@Evolving
public interface SupportsTableStateOptions extends CatalogPlugin {

  /**
   * Returns the raw option keys that may affect the table state selected by {@code loadTable}. An
   * empty set is equivalent to not implementing this interface: no option affects table state.
   *
   * @return a non-null set of case-insensitive option keys
   */
  Set<String> tableStateOptionKeys();
}

[yyanyy]

Had some offline discussion on this and post here for posterity: making the default behavior to consider all options as table state option could prevent accidental correctness issue when options can dictate different table versions, but this introduced 2 problems:

• this will make all DSV2 connectors not implementing the interface to start missing cache when options are stated differently, causing performance problem
• it will break the table pinning behavior within one query when the table was specified twice with different options, and this could potentially result in correctness concern

[aokolnychyi]

Can we call it extractTableStateOptions?

[aokolnychyi]

I think this needs to be flipped to return an empty map to preserve the original 4.2 behavior.

[aokolnychyi]

I think the logic here needs to be lookup using state keys only, if found, copy with the final options.

[yyanyy]

Offline conversation for posterity: we were worried that taking all options into consideration for looking up shared relation cache could cause non-deterministic behavior when users specify queries against tables in one single spark session, where some queries have the exact option list matching shared relation cache, while some others don't; in this case user could observe different queries using different table versions and cause confusion. One query referring to the same table with different non-state-options also will be undeterministic and depend on the order of specifying such tables. (There's test cases covering scenarios mentioned above.) Also, we were not able to think of a concrete use case where the non-state options need to participate in the selection for the table from shared relation cache, as it was used for obtaining the table without a table reload; the actual reusing of the cache the data, do consider all options correctly.

[aokolnychyi]

We have to push down the lookup into RelationCache with something like:

  def lookup(
      catalog: CatalogPlugin,
      ident: Identifier,
      tableId: Option[String],
      stateOptions: CaseInsensitiveStringMap,
      resolver: Resolver): Option[DataSourceV2Relation] = {
  ...
}