[SPARK-58389][SQL] Pass all options while loading tables

[yyanyy]

What changes were proposed in this pull request?

This is a follow-up to #56044 (which passed all options while loading changelogs). It does the same for table reads by adding TableCatalog.loadTable(Identifier, TableContext, CaseInsensitiveStringMap), where TableContext carries the parsed, Spark-recognized parameters (time travel, write privileges) and the CaseInsensitiveStringMap carries all raw user options.

• New public connector types TableContext and TimeTravel (a clean sealed interface with AsOfVersion/AsOfTimestamp records), mirroring how [SPARK-56961][SQL] Pass all options while loading changelog #56044 introduced ChangelogContext/ChangelogRange rather than leaking the catalyst-internal TimeTravelSpec.
• The new loadTable overload has a default implementation that delegates to the existing loadTable overloads based on TableContext, so existing connectors keep working unchanged.
• CatalogV2Util.getTable/loadTable now build a TableContext from the catalyst TimeTravelSpec + write-privileges string and forward the user options, making the Java default the single dispatch site.
• Options are threaded through the read paths in RelationResolution and DataSourceV2Utils, and the SupportsCatalogOptions write path in DataFrameWriter.
• The analyzer per-query relation cache is keyed on the user options (extracted into RelationCacheKey), so two references to the same table with different options each get their own Table instead of sharing the first one's. The shared relation cache (CACHE TABLE) is reused only when the options match.

Potential follow-ups (intentionally out of scope here):

• The RelationCatalog single-RPC loadRelation(Identifier) path is not updated to forward options; closing it cleanly would add a loadRelation(Identifier, CaseInsensitiveStringMap) overload.
• The V2TableReference re-resolution path (getOrLoadRelation/loadRelation, used by DataFrame temp views, transaction re-resolution, and write targets) still loads via the single-arg loadTable. It is transaction-aware (the transaction catalog intercepts loadTable to track reads), so forwarding options there is better done as a separate change.

Why are the changes needed?

To make the API usable in connectors like Iceberg and Delta, which need the user options while reading a table. The functionality hasn't been released yet.

Does this PR introduce any user-facing change?

Yes, minor: INSERT INTO t WITH ('versionAsOf' = ...) (time travel on a write target, reachable via the option form) now raises a user-facing UNSUPPORTED_FEATURE.TIME_TRAVEL analysis error instead of an internal AssertionError. Connectors that override the new loadTable overload also see one loadTable call per distinct option bag when the same table is referenced multiple times.

How was this patch tested?

New tests in CatalogV2UtilSuite (the default-dispatch mapping for each TableContext shape, and the time-travel/write-privileges mutual-exclusion invariant), DataSourceV2OptionSuite (end-to-end option forwarding via the DataFrame API, SQL, streaming, time travel, and the write path; per-option-bag loads on a self-join; a temp view's options not leaking to a later reference; CACHE TABLE results not reused for a differing-options read), PlanResolutionSuite (shared relation cache reused only when options match), and SupportsCatalogOptionsSuite (read-path option forwarding). Existing ChangelogResolutionSuite, ChangelogEndToEndSuite, and the DataSourceV2 SQL/DataFrame suites pass.

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

Generated-by: Claude code Opus 4.8

[peter-toth]

Heads-up on an overlap: @anuragmantri has #57508 (SPARK-58330) open, and it changes the same lines this PR touches.

#57508 fixes the case where the same table appears more than once in a statement with different WITH (...) options. The per-query relation cache is keyed on (catalog, identifier, timeTravelSpec) only, so the second reference gets a cache hit and silently inherits the first reference's options. Its fix re-applies each reference's own options to the cached relation on a hit.

Mechanically the two will conflict: #57508 restructures the .orElse { val finalOptions = ... } block that this PR reads finalOptions from. (Minor and separate: this PR and #57585 conflict with each other on the import line in DataSourceV2OptionSuite, though their RelationResolution edits merge fine.)

The design question. A cache hit never calls the catalog. So once loadTable can produce a different Table depending on the options, a second reference with different options keeps the Table built for the first one — its schema, partitioning, capabilities, and whatever the catalog selected from those options. #57508 can correct the options field on the cached relation, but it cannot re-derive the Table. Which reference wins is analyzer traversal order — for INSERT the target resolves after its query.

Spark already has a worked example of an option that affects the load, and it's instructive: time travel can be given purely as an option, versionAsOf / timestampAsOf (key names configurable via spark.sql.timeTravelVersionKey / timeTravelTimestampKey). TimeTravelSpec.fromOptions parses it out of the read options, and then Spark does three things with it — puts it in the relation cache key, dispatches at load time to loadTable(ident, version), and forbids it on a write target. It deliberately does not leave it to newScanBuilder(options), and it can't: DataSourceV2Relation.output comes from the Table returned at load time, so a snapshot selector that only reached the scan builder would have the query analyzed against the current schema and scanned at the old one.

So the question for this API is which kind of option it is meant to carry:

• Options that change how the table is read (split size and the like): DataSourceV2Relation.options is the right home, that is already per-reference, and [SPARK-58330][SQL] Prevent silent dropping of dynamic options on same table references. #57508 is the complete fix — nothing needed here.
• Options that change what the table is (a snapshot, a branch, a schema): they need the versionAsOf treatment — cache key, load-time dispatch, rejected on write targets. Today a connector's own selector of this kind is invisible to fromOptions, so it gets none of the three, and this PR is what makes it actionable at load time.

The javadoc currently says both — "take them into account when producing the Table" and "customize the scan" — which is what leaves it ambiguous.

The two treatments are not interchangeable, either. Putting options in the cache key means one load per distinct option bag, which gives up resolve-once-per-query: MERGE INTO t WITH (a) USING t WITH (b) would call loadTable twice, and each call resolves the table's state independently, so whether the target and the source observe the same state becomes the connector's problem rather than Spark's guarantee. #57508's approach keeps one Table for the statement and varies only the options. Worth picking deliberately.

Side note on a problem this creates regardless of which way it goes: the single-pass resolver keys on RelationId(multipartIdentifier, options, isStreaming, timeTravelSpec), so it already resolves once per option bag while the fixed-point relation cache does not. Once a connector overrides the new overload, the same query would resolve differently under the two.

Separate point on the same refactor. CatalogV2Util.getTable currently has:

assert(writePrivilegesString.isEmpty, "Should not write to a table with time travel")

This PR removes it, and the new default overload replaces it with silent precedence:

if (context.timeTravel().isPresent()) { ... }
else if (!context.writePrivileges().isEmpty()) { ... }

If both are ever set, time travel wins and the write privileges are dropped, so the catalog is never asked to authorize the write — the same shape as SPARK-58370. I couldn't find an earlier analyzer guard, so as far as I can tell that assert was the enforcement. It may be unreachable from SQL via the grammar today, in which case keeping the assert seems better than silently choosing one. A connector overriding the new method also receives a TableContext that can hold both, with no word on what the combination means.

Last thing: neither this PR nor #57585 has a test with the same table referenced twice, and the lastLoadTableOptions fixture records only the last call, so it could not express one.

The same argument applies to #57585 (RelationCatalog.loadRelation) — no need to repeat it there.

cc @anuragmantri

[aokolnychyi]

4.3.0?

[aokolnychyi]

Is my understanding correct that every option the user actually provided surfaces in options, even if it makes into a structured representation like time travel? The only thing removed is Spark's internal write-privilege marker, which the user never provided?

[yyanyy]

That's right. Every user-provided option is forwarded, including keys we also parse into context (e.g. versionAsOf). The only thing stripped is the internal REQUIRED_WRITE_PRIVILEGES marker, which users never set.

[aokolnychyi]

Optional: Is there a good utility method to put this conversion?

[aokolnychyi]

4.3.0?

[aokolnychyi]

4.3.0?

[aokolnychyi]

Should we prefer to use Spark exceptions? I believe we do use them in a few other places.

if (timeTravel != null && !privileges.isEmpty()) {
  throw new SparkIllegalArgumentException(
      "INTERNAL_ERROR",
      Map.of("message", "Cannot set both time travel and write privileges"));
}

[aokolnychyi]

This should be a Spark exception with INTERNAL_ERROR too. No need to define a new error class, just switch to Spark exceptions.

[aokolnychyi]

If you go with switch, you probably don't need the exception.

[aokolnychyi]

Can this be simplified with switch?

return switch (context.timeTravel().get()) {
  case TimeTravel.Version v -> loadTable(ident, v.version());
  case TimeTravel.Timestamp ts -> loadTable(ident, ts.micros());
};

[yyanyy]

I think we also compile with java 17 that switch isn't available

[aokolnychyi]

I am a bit on the fence with the name here.
I wonder if AsOfVersion/AsOfTimestamp would be better to match the SQL and internal values.

@cloud-fan, thoughts?

[yyanyy]

AsOfVersion/AsOfTimestamp sounds more reasonable to me too, updated

[aokolnychyi]

Do we need () around writePrivileges == null?

[aokolnychyi]

Can we use Set.of()? Will it simplify it?

[aokolnychyi]

Optional: I'd consider flipping the order of the check, but this is optional.

this.timeTravel = timeTravel;
this.writePrivileges = privileges == null ? Set.of() : Set.copyOf(privileges);
if (timeTravel != null && !writePrivileges.isEmpty()) {
  throw new SparkIllegalArgumentException(...)
}

[aokolnychyi]

This looks good to me. I left some minor comments.

[aokolnychyi]

I just saw @peter-toth's comment. Let me take a deeper look.

[aokolnychyi]

@peter-toth @anuragmantri, historically we didn't let Spark pass options to table loads. This created problems for many formats, including Iceberg. For example, Spark doesn't standardize branching, so Iceberg has custom extensions to resolve the branching later to work around this. We won't be able to standardize all table state properties across connectors, meaning we have to be able to pass these custom options to the connector. This does mean the analyzer cache must change too. Let me look into that a bit more.

[peter-toth]

Thanks @aokolnychyi for the details. Cache change makes sense to me and I'm not against it. It was just a heads-up that #57508 and #57563 landed recently and might cause some conflicts.

[aokolnychyi]

@yyanyy, can you rebase this one and then look into adapting the the analyzer and shared relation cache to respect the options? It seems we currently handle this inconsistently in the legacy analyzer. The new analyzer already keys on the options?

[aokolnychyi]

I am not sure you want to be this specific, there may be more usages in the future.

A time-travel specification for reading a table as of a specific version or point in time.

This seems good enough for me.

[aokolnychyi]

Question: Is there a good place where we can put a helper container? I see RelationResolution has the same triple dependency.

Also, is it feasible to promote RelationId from the single pass analyzer to AnalysisContext and use it as cache key in both analyzers?

[yyanyy]

Extracted the triple dependency to RelationCacheKey in a new file following examples of RelationCache and RelationId as neither AnalysisContext nor RelationResolution owns the key.

For promoting RelationId to be the shared key - will follow up with another PR on this since it's not a trivial change, as the two keys in both analyzers have some overlap but also diverge on the rest, e.g. isStreaming, and didn't want to increase the scope of this PR by too much

[aokolnychyi]

The updated PR seems reasonable to me (I did a quick pass).

@peter-toth @anuragmantri @cloud-fan, could you please review too?

[anuragmantri]

Thanks for rebasing the PR @yyanyy. I reviewed the relation cache and the resolution and it looks good to me.

@aokolnychyi - Should we make a change in Iceberg to use this new API? I have some use-cases for passing in arbitrary options through SQL.

[cloud-fan]

0 blocking, 0 non-blocking, 0 nits.
The options-aware API, analyzer cache identity, compatibility fallback, and test coverage are coherent; I found no actionable defects.

Verification

I traced the option map from DataFrame and SQL relation resolution through CatalogV2Util into the new catalog overload. I also verified that RelationCacheKey compares CaseInsensitiveStringMap by normalized map content, that write targets bypass the per-query cache, and that the default overload maps each TableContext state to the corresponding existing loadTable method.

[peter-toth]

Thanks for the PR, @yyanyy!

Follow-up to #56044 that adds TableCatalog.loadTable(ident, TableContext, options) with a default implementation dispatching to the existing overloads, and -- per @aokolnychyi's ask -- keys the per-query relation cache on the options so two references with different bags each get their own Table instead of sharing the first one's. The API shape and the per-query cache change both look right to me, and the mutual-exclusion guard plus the new UNSUPPORTED_FEATURE.TIME_TRAVEL on a write target close the dropped-assert hole I raised earlier. Three things I'd fix before merge: an option leak that removing applyOptions opened up on the V2TableReference cache path, the shared relation cache still handing back the option-free Table (the other half of @aokolnychyi's "analyzer and shared relation cache" ask), and the new javadoc not saying that an override takes over write-privilege authorization.

Blocking

• 1. option leak on a per-query cache hit: getOrLoadRelation stores its relation under a key with empty options while the relation carries ref.options, so an option-free read of the same table in the same statement hits it and inherits them -- the applyOptions re-apply masked this on master. Verified: a DataFrame temp view carrying split-size=5 joined to a plain read of the same table gives List(5, 5). [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:549]
• 2. the shared relation cache still discards the options-aware Table: when the table is in the CacheManager, the table just loaded with finalOptions is thrown away and the cached read's table is reused with only options re-stamped, so a connector's option-tailored table never reaches the plan. [inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:285]
• 3. new javadoc doesn't state the write-authorization contract: the default body is also the dispatch point for loadTable(ident, writePrivileges) and for time travel, so an override written just to grab the options silently drops both. [inline: sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java:204]

Non-blocking

• 4. V2TableReference re-resolution still loads without options: RelationResolution.scala:510 calls loadTable(ref.identifier) even though ref.options is right there, so DataFrame temp views, transaction re-resolution and write targets never reach the new overload. Either forward them, or list this next to the RelationCatalog.loadRelation gap in the description.
• 5. SupportsCatalogOptions write path missed: DataFrameWriter.scala:179 still does catalog.loadTable(ident) while its read counterpart in DataSourceV2Utils now forwards dsOptions -- same one-line change, dsOptions is already in scope.
• 6. description drift: "Does this PR introduce any user-facing change?" says No, but INSERT INTO t WITH ('versionAsOf' = 'v1') goes from an internal AssertionError to UNSUPPORTED_FEATURE.TIME_TRAVEL, and connectors now see one loadTable call per distinct option bag. The first bullet also still says TimeTravel has Version/Timestamp records -- they were renamed to AsOfVersion/AsOfTimestamp.
• 7. the DataSourceV2Utils change is untested: DataSourceV2Utils.scala:144 (the SupportsCatalogOptions read path, spark.read.format(...).option(...).load(...)) is the one production line with no new test; SupportsCatalogOptionsSuite looks like the place for it.

[peter-toth]

Finding 1. This default is what the other toCacheKey caller gets, and it makes the new "a hit means the options already match" invariant false for the entries that caller writes -- RelationResolution.scala:487-497:

private def getOrLoadRelation(ref: V2TableReference): LogicalPlan = {
  val key = toCacheKey(ref.catalog, ref.identifier)   // options = empty
  relationCache.get(key) match {
    ...
    case None =>
      val relation = loadRelation(ref)                // relation.options = ref.options
      relationCache.update(key, relation)             // key says "no options", value carries them

So a V2TableReference that carries options stores its relation under the empty-options key, and the next option-free UnresolvedRelation for the same table hits that entry and inherits them. On master the .map(applyOptions(_, finalOptions)) this PR removes forced the options on every hit, so it was masked; now it leaks. V2TableReference is reachable from a DataFrame temp view (views.scala:759-761), from transaction re-resolution (UnresolveRelationsInTransaction), and for write targets.

Repro I ran on this head, added to DataSourceV2OptionSuite:

spark.read.option("split-size", "5").table(t1).createOrReplaceTempView("v")
val df = sql(s"SELECT v.id FROM v JOIN $t1 b ON v.id = b.id")
df.queryExecution.analyzed.collect { case r: DataSourceV2Relation => r.options.get("split-size") }
// got List(5, 5) -- `b` inherited the view's option; expected List(5, null)

Order decides which way it goes: with the plain read on the left it comes out List(null, 5), because in that direction adaptCachedRelation(cached, ref) re-applies ref.options onto the hit.

Putting the ref's options in the key fixes it, and all 25 DataSourceV2OptionSuite tests stay green with it (verified locally):

    val key = toCacheKey(ref.catalog, ref.identifier, None, ref.options)

A test for it would fit next to the two new SPARK-58389 cache tests.

[yyanyy]

Thanks for the catch and the detailed reproduction! Updated

[peter-toth]

Finding 2. The Table loaded here with finalOptions is discarded ~30 lines below whenever the table sits in the CacheManager -- RelationResolution.scala:313-323:

val sharedRelationCacheMatch = for {
  t <- table
  if finalTimeTravelSpec.isEmpty && writePrivileges == null && !u.isStreaming
  cached <- lookupSharedRelationCache(catalog, ident, t)
} yield {
  val updatedRelation = cached.copy(options = finalOptions)   // keeps cached.table

Neither side of that lookup considers options: CacheManager.lookupCachedTable matches by name (CacheManager.scala:438, plus timeTravelSpec.isEmpty), and isSameTable compares rel.table.id == table.id (CatalogV2Util.scala:542). So for any table someone has cache()d, a read carrying options gets the Table that was built for the cached read's options, with this read's options merely stamped onto the relation -- the exact failure mode the per-query cache change fixes, and the "shared relation cache" half of #57582 (comment). The new PlanResolutionSuite test ("shared relation cache hit re-applies the current read's options", assert(resolved.table == cachedTable)) currently locks that behavior in.

Reusing the cached table also buys nothing once the options differ: the resulting plan carries them, so it no longer matches the cached entry's fingerprint -- your own DataSourceV2OptionSuite test asserts exactly that (lookupCachedData(spark.read.option("split-size", "5").table(t1)).isEmpty). So the connector pays for the load and Spark throws the result away for no cache reuse.

Suggest gating the reuse on the options matching and letting the freshly loaded table win otherwise:

val sharedRelationCacheMatch = for {
  t <- table
  if finalTimeTravelSpec.isEmpty && writePrivileges == null && !u.isStreaming
  cached <- lookupSharedRelationCache(catalog, ident, t)
  if cached.options == finalOptions
} yield {
  val nameParts = ident.toQualifiedNameParts(catalog)
  val aliasedRelation = SubqueryAlias(nameParts, cached)
  relationCache.update(key, aliasedRelation)
  adaptCachedRelation(aliasedRelation, planId)
}

That keeps SPARK-54022's guarantee for the case it was written for (same read, same options -> same table version as the cached plan) and drops it only where the cached plan could not have been reused anyway.

[yyanyy]

Thanks for the catch! I originally looked at this a little, but decided it might not be an issue because the table returned should have the options correctly reflected, and that would hit the cache correctly. But you're right that I didn't realize the same-table check inside the cache lookup is based only on the ID, and there's no guarantee the ID reflects the options; even if the options substantially change the table content, the ID might still be the same. So agreed, it's much safer to include the options in the lookup.

This may give up reuse in some edge cases where the cache could actually have been reused safely, e.g. purely read-tuning options that don't change which table is read or how it's structured; but that's still better than risking the wrong table from the cache.

[yyanyy]

Though I just realized that I need to make sure all the cache/refresh will include options, and in my current commit it's not guaranteed; will need to amend this PR one more time to include it

[yyanyy]

This is updated and ready to review, please take a look. Thank you!

[aokolnychyi]

This is actually more subtle than this, given that Spark relies on the shared relation cache to pin the versions of the tables. Let me think.

[aokolnychyi]

The only option I can think of right now is defining TableCatalog$tableStateOptions as a method that would list what options we have to pass to load. In case of Iceberg, this will be branch, for example. Then we will have to key our caching semantic on those keys alone.

[aokolnychyi]

I wonder if I tend to overcomplicate this... Still thinking.

[yyanyy]

Agreed that ideally we should separate the connector specific options from the options only matter to Spark to avoid unnecessary cache misses

[aokolnychyi]

I think my point can be addressed in a follow up, we need to discuss it a bit more.

[peter-toth]

Finding 9 (Non-blocking, and I think it belongs in the follow-up @aokolnychyi parked rather than in this PR): separate from the state-vs-cosmetic split, the lookup behind both new guards can fail to find a matching entry even when one exists.

CacheManager.lookupCachedTable matches cache entries by name only and returns just one of them -- the head of the matches (CacheManager.scala:438-449) -- and since cachedData is prepended on each cache (CacheManager.scala:165), that is the most recently cached entry. The new guards then compare that one candidate's options against the current read's (RelationResolution.scala:320, V2TableRefreshUtil.scala:93) and reuse its Table only when they are equal; nothing goes back to check the other name matches. A table has more than one entry because cachedData is keyed by the normalized plan and the plan carries options (CacheManager.scala:145), so spark.table(t).cache() and spark.read.option("split-size", "5").table(t).cache() are two entries. A read with split-size=5 therefore gets the pin only if that entry happens to be the newer of the two.

Reproduced on 3377cd99 -- same query both times, only the caching order swapped:

// cache both spark.table(t) and spark.read.option("split-size", "5").table(t)
val df = spark.read.option("split-size", "5").table(t).filter("id > 0")
df.queryExecution.analyzed
inMemoryCatalog.resetLoadTableCalls()
df.collect()
inMemoryCatalog.loadTableCalls.map(_._2.get("split-size")).filter(_ != null)

// split-size entry cached last  -> List()   pinned to the cached table, no reload
// option-free entry cached last -> List(5)  matching entry never offered, reloads latest

So today whether a query keeps its session-pinned version is decided by which of that table's cache entries was written most recently. It can also cost an outright cache hit, once the table has moved on from what was cached -- which is the situation the pin exists for. Cache spark.read.option("split-size", "5").table(t), let the table get a new version, then re-run that exact query with a newer cache entry for t present: the lookup offers the newer entry, the guard rejects it, the relation is loaded fresh at the new version, and sameResult then fails against the v=old entry that would have matched. With the caching order swapped, the same query is served from the cache.

TableCatalog#tableStateOptions would shrink the key but not close this: with two entries differing only in state options, the lookup still offers just one of them. The multi-reference case makes it concrete -- one query reading t on branch=b1 and t on branch=b2, with both branches cached, resolves each reference against the same single newest candidate (tryResolvePersistent runs per reference and there is no memo across them), so exactly one of the two gets pinned to its cached snapshot and which one flips with the caching order. Closing it means making the shared-cache lookup options-aware instead of filtering after the fact -- RelationCache.lookup returns Option[LogicalPlan] today, so it would need to expose all name matches (or take the options) and let the caller pick the one that matches.

On your split-size point: you're right, and it corrects something I asserted earlier in this thread. I argued the old cached.copy(options = finalOptions) branch "buys nothing" once the options differ, on the grounds that the resulting plan can't match the cached entry's fingerprint anyway. That holds for the useCachedData hit, but the branch was also providing version pinning, which is independent of it -- so for a purely cosmetic option the guard does give up something real.

[peter-toth]

Finding 3. This tells connectors to override the method to receive the options, but not what they take over by doing so. The default body right below is also the dispatch point for loadTable(ident, writePrivileges) -- the call a catalog uses to authorize a write, and the one SPARK-58370 hardened after an authorization bypass -- and for time travel. A connector that overrides this to grab the options and returns a table without looking at context silently loses both, and nothing in Spark will catch it: CatalogV2Util.getTable has no other dispatch site now.

Suggest spelling the contract out:

   * The default implementation ignores {@code options} and delegates to the existing
   * {@code loadTable} overloads based on {@code context}. Catalogs that want to receive the user
   * options while reading a table must override this method.
   * <p>
   * An override replaces that dispatch and must honor {@code context} itself: apply the time
   * travel in {@link TableContext#timeTravel()}, and authorize the requested
   * {@link TableContext#writePrivileges()} as it would in
   * {@link #loadTable(Identifier, Set)}. Spark does not re-check either afterwards.

[peter-toth]

Re-checked through 96b9ddab -- findings 1-7 all resolved (getOrLoadRelation's key carries ref.options, shared-cache reuse is gated on matching options, the javadoc states the override contract, DataFrameWriter's save path forwards dsOptions, the description is accurate, and SupportsCatalogOptionsSuite covers the read path). Picking up your own note that you wanted to make sure "all the cache/refresh will include options": CacheManager.tryRefreshPlan gets there for a bare cached relation, but the refresh path everything else goes through -- including the per-query refresh phase -- does not.

Blocking

• 8. refresh phase reloads the table option-blind (late catch): V2TableRefreshUtil.refresh loads via catalog.loadTable(ident), memoizes on (catalog, ident), and takes cached.table from the shared relation cache with no options check, so the options-aware Table is swapped out for an option-free one at execution time. Reproduced both on a plain spark.read.option(...).table(t).collect() and on recaching a cached plan that isn't a bare relation; a validated fix and its cost are in the inline comment. [inline: sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:422]

[peter-toth]

Finding 8. This branch now forwards r.options, but the case _ => Some(V2TableRefreshUtil.refresh(spark, plan)) at CacheManager.scala:429 doesn't -- and that is the path every cached plan except a bare relation takes, plus the per-query refresh phase at QueryExecution.scala:271-272 (refreshPhaseEnabled defaults to true). V2TableRefreshUtil.refresh, V2TableRefreshUtil.scala:84-102:

val currentTables = mutable.HashMap.empty[(TableCatalog, Identifier), Table]  // no options
...
  lookupCachedRelation(spark, catalog, ident, r.table) match {
    case Some(cached) => cached.table            // shared cache, no options check
    case None => catalog.loadTable(ident)        // r.options is right here, not passed
  }
...
r.copy(table = currentTable)                     // keeps r.options, swaps the Table

Three ways that undoes what this PR just fixed:

• the load ignores r.options, so the relation still says split-size=5 while its Table was built for no options -- the same split the per-query cache change removed;
• cached.table comes from the shared relation cache with no options check, i.e. the guard you just added at RelationResolution.scala:316-320 is not applied at this sibling site;
• the memo key is (catalog, ident), so a self-join at two different option bags shares one reloaded Table -- the RelationCacheKey fix undone one layer down.

The loop already has the concept of a relation whose identity the user pinned: it skips anything with a time-travel spec, because refreshing that to "latest" would be wrong. An option that selects a branch or a snapshot is the same situation, which is how the gap arose -- the difference being that time travel must not be refreshed at all, while an option-selected table should be refreshed with its options.

versionedOnly = true in the refresh phase is barely a limit: isVersioned is table.version != null (DataSourceV2Relation.scala:145), which holds for Iceberg/Delta and for the in-tree InMemoryBaseTable (version() returns tableVersion.toString). So it fires on ordinary reads, and validateDataColumns will surface a schema difference as columnsChangedAfterAnalysis rather than reading the option-selected table.

Reproduced on this head with a catalog that counts direct single-arg loads:

class LoadCountingCatalog extends InMemoryCatalog {
  val singleArgLoads = new AtomicInteger(0)
  override def loadTable(ident: Identifier): Table = {
    singleArgLoads.incrementAndGet()
    super.loadTable(ident)
  }
}

// plain read, no caching at all:
spark.read.option("split-size", "5").table(t1).collect()
// optionsAwareLoads=1 singleArgLoads=2  -> one load bypassed the options-aware overload

And for the recache case -- cache spark.read.option("split-size", "5").table(t1).filter("id > 0"), then spark.catalog.refreshTable(t1) -- the only options-aware load carries split-size=null, so "Preserve table options when recaching a cached table" holds for the bare-relation branch only.

What made both go green locally:

val currentTables =
  mutable.HashMap.empty[(TableCatalog, Identifier, CaseInsensitiveStringMap), Table]
...
val currentTable = currentTables.getOrElseUpdate((catalog, ident, r.options), {
  val tableName = V2TableUtil.toQualifiedName(catalog, ident)
  lookupCachedRelation(spark, catalog, ident, r.table) match {
    case Some(cached) if cached.options == r.options =>
      cached.table
    case _ =>
      CatalogV2Util.getTable(catalog, ident, options = r.options)
  }
})

One cost worth knowing before you take it: this makes the refresh phase a second options-aware load per relation, which two of your new tests then see, because they count loadTableCalls across the whole statement. "a self-join with different options loads the table once per option bag" becomes Seq(5, 5, 9, 9) and "repeated references with the same options load the table once" counts 2. Both need their counting scoped to analysis (reset the recorder after queryExecution.analyzed) or changed to count distinct option bags. The second load itself is inherent to the refresh phase, not something the fix introduces.

Same shape, also unhandled, if you want to sweep them or name them as out of scope: ResolveSchemaEvolution.scala:54 and the catalog.loadTable(ident, INSERT) fallbacks in WriteToDataSourceV2Exec (:101, :145, :208, :278) all reload a relation whose options are in hand.

While you are in the description: the bullet list doesn't mention the CacheManager recache change at all, so it would be worth covering that and the refresh paths there.

[yyanyy]

thanks for the catch! was in a rush to get this PR ready and trusted claude to tell me refresh coverage is enough without verifying myself... updating now

[yyanyy]

updated, thanks!

[peter-toth]

Re-checked through 3377cd99 -- finding 8 resolved: V2TableRefreshUtil.refresh now keys its memo on r.options, gates the shared-cache branch on matching options, and loads through CatalogV2Util.getTable. The four new refresh tests all fail on 96b9ddab and pass here, and CI is green. Nothing blocking left from my side.

Non-blocking

• 9. shared-cache pin depends on cache insertion order (late catch): Not a change request for this PR. @aokolnychyi already concluded that the state-vs-cosmetic options keying "can be addressed in a follow up", and this belongs in that same follow-up -- I'm recording it so the design accounts for it. CacheManager.lookupCachedTable returns only the newest name match, so both new cached.options == ... guards accept or reject whichever single entry it hands back; with two cached entries for one table, a read whose options exactly match the older one is rejected and loses its version pin. The part that matters for the follow-up: narrowing the key (e.g. via tableStateOptions) is not sufficient on its own, because the lookup still offers only one candidate. Reproduced by swapping only the caching order -- details and the repro are on @aokolnychyi's thread.

Minor

• 10. description misses two production files and a behaviour change (round 2): the bullet list still doesn't mention the CacheManager recache change or the V2TableRefreshUtil refresh change, both of which are now part of the PR. The user-facing-change section is also missing one: on master a read carrying options reuses a cached table's version unconditionally, and now it only does so when the options match -- so anyone combining CACHE TABLE with dynamic options can see a query stop being served from the cache, and get freshly loaded data instead of the cached snapshot.