peter-toth commented on code in PR #57582:
URL: https://github.com/apache/spark/pull/57582#discussion_r3691370468
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -556,8 +545,10 @@ class RelationResolution(
private def toCacheKey(
catalog: CatalogPlugin,
ident: Identifier,
- timeTravelSpec: Option[TimeTravelSpec] = None): CacheKey = {
- ((catalog.name +: ident.namespace :+ ident.name).toImmutableArraySeq,
timeTravelSpec)
+ timeTravelSpec: Option[TimeTravelSpec] = None,
+ options: CaseInsensitiveStringMap = CaseInsensitiveStringMap.empty()):
RelationCacheKey = {
Review Comment:
**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`:
```scala
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`:
```scala
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):
```scala
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.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -280,7 +281,8 @@ class RelationResolution(
catalog,
ident,
finalTimeTravelSpec,
- Option(writePrivileges))
+ Option(writePrivileges),
+ finalOptions)
Review Comment:
**Finding 2.** The `Table` loaded here with `finalOptions` is discarded ~30
lines below whenever the table sits in the `CacheManager` --
`RelationResolution.scala:313-323`:
```scala
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 https://github.com/apache/spark/pull/57582#issuecomment-5134589603. 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:
```scala
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.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java:
##########
@@ -194,6 +195,45 @@ default Table loadTable(Identifier ident, long timestamp)
throws NoSuchTableExce
throw QueryCompilationErrors.noSuchTableError(name(), ident);
}
+ /**
+ * Load table metadata by {@link Identifier identifier} from the catalog,
forwarding all
+ * user-specified options.
+ * <p>
+ * 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.
Review Comment:
**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:
```java
* 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.
```
--
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]
