peter-toth commented on code in PR #57582:
URL: https://github.com/apache/spark/pull/57582#discussion_r3693371645
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -419,9 +419,9 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
try {
EliminateSubqueryAliases(plan) match {
case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if
r.timeTravelSpec.isEmpty =>
- val table = catalog.loadTable(ident)
+ val table = CatalogV2Util.getTable(catalog, ident, options =
r.options)
Review Comment:
**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`:
```scala
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:
```scala
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:
```scala
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
