peter-toth commented on code in PR #57508:
URL: https://github.com/apache/spark/pull/57508#discussion_r3655890534
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -358,6 +364,13 @@ class RelationResolution(
CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident,
table, conf)
}
+ private def applyOptions(
+ cached: LogicalPlan,
+ options: CaseInsensitiveStringMap): LogicalPlan = cached transform {
Review Comment:
**Finding 5.** This `transform` rewrites *every* matching node in the cached
plan, with no check that the node is the relation for the identifier being
resolved — unlike the other cache-hit rewrite in this file,
`adaptCachedRelation(cached, ref)` at `:519-525`, which only touches a relation
passing `matchesReference` (same catalog and identifier).
It is safe today, but only because of an invariant that isn't written down
anywhere: all three writers to `relationCache` (`:313`, `:325`, `:493`) store a
plan with a single relation node, and that node belongs to the key. For a view
the cached plan is `SubqueryAlias(View(desc, child =
parser.parseQuery(viewText)))`, i.e. the body is still unresolved when it goes
into the cache, so it holds `UnresolvedRelation`s rather than the two types
matched here (same for `MetricViewPlaceholder`). If a cache entry ever held a
resolved sub-plan — a view body resolved before caching, say — then `SELECT ...
FROM v WITH ('x' = 1)` on a second reference would stamp the outer reference's
options onto the relations of unrelated tables inside it, silently and with no
error.
I'd just record the assumption rather than add a guard:
```scala
/**
* Re-applies `options` to the relation in a cached plan. Every
`relationCache` entry holds a
* single relation for its own identifier (a view's body is still
unresolved when it is cached),
* so this cannot reach another table's relation.
*/
private def applyOptions(
```
An identifier guard mirroring `matchesReference` is the tempting move but
it's awkward for `UnresolvedCatalogRelation`, which only carries a
`CatalogTable` — you'd be comparing a qualified `TableIdentifier` against the
v2 `Identifier` plus `catalog.name`, and getting that slightly wrong means
options quietly stop being re-applied, i.e. this PR's bug comes back on some
path. If you do want a structural guard instead of the comment, matching the
shapes (`case SubqueryAlias(id, r: DataSourceV2Relation) => ...`, `case r:
DataSourceV2Relation => ...`, `case other => other`) keeps it to the top
relation without any name comparison.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/RelationResolution.scala:
##########
@@ -358,6 +364,13 @@ class RelationResolution(
CatalogV2Util.lookupCachedRelation(sharedRelationCache, catalog, ident,
table, conf)
}
+ private def applyOptions(
+ cached: LogicalPlan,
+ options: CaseInsensitiveStringMap): LogicalPlan = cached transform {
+ case r: DataSourceV2Relation => r.copy(options = options)
+ case r: UnresolvedCatalogRelation => r.copy(options = options)
Review Comment:
**Finding 4.** This line also fixes the streaming *v1* path, since
`createRelation` wraps a session-catalog streaming table as `SubqueryAlias(_,
UnresolvedCatalogRelation(v1Table, options, isStreaming = true))`
(`RelationResolution.scala:398-401`). The v2 streaming shape is the one left
out: `createRelation` builds `StreamingRelationV2(..., options, ...)`
(`:438-450`), a `LeafNode` that holds its options in `extraOptions`, which
`applyOptions` doesn't match — so on a cache hit the second reference keeps the
first reference's options.
It is reachable from SQL: `optionsClause` is allowed on a stream relation
(`SqlBaseParser.g4:447-448`, `AstBuilder.visitStreamTableName` at
`AstBuilder.scala:3224`) and a stream-stream self join is expressible, e.g.
``SELECT * FROM STREAM t WITH ('a'='1') x JOIN STREAM t WITH ('a'='2') y ON
x.id = y.id`` (cf. `StreamRelationSuite.scala:58`). `StreamingRelationV2` is
already imported at `:31`, so it's one line:
```suggestion
case r: UnresolvedCatalogRelation => r.copy(options = options)
case r: StreamingRelationV2 => r.copy(extraOptions = options)
```
If you'd rather keep streaming out of this PR, a sentence in the method
comment saying the streaming v2 relation is not covered would do — this PR
removes the only comment that recorded the old limitation.
--
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]
