dongjoon-hyun commented on code in PR #59448:
URL: https://github.com/apache/spark/pull/59448#discussion_r4267518428
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -259,7 +259,7 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
resolver: Resolver,
includeTimeTravel: Boolean): Boolean = {
- EliminateSubqueryAliases(plan) match {
+ stripRootSubqueryAliases(plan) match {
Review Comment:
With the whole-tree rewrite gone, most of the remaining per-node overhead in
these by-name scans is the resolver lookup. `uncacheTableOrView` (L243) and
`recacheTableOrView` (L382) pass `spark.sessionState.conf` into the `exists`
predicate, so `conf.resolver`, i.e. `getConf(CASE_SENSITIVE)` (a conf-map
lookup plus a boolean conversion), is evaluated on every visited node of every
cache entry. Since this PR targets the cost of these scans, how about resolving
it once per call?
```scala
val resolver = spark.sessionState.conf.resolver
... isMatchedTableOrView(_, name, resolver, includeTimeTravel)
```
The `SQLConf` overload of `isMatchedTableOrView` would then become unused.
##########
sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala:
##########
@@ -1952,6 +1976,53 @@ class CachedTableSuite extends SharedSparkSession
}
}
+ test("cache lookups by table name do not rewrite cached plans") {
+ val t = "testcat.tbl"
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id int, data string) USING foo")
+ sql(s"INSERT INTO $t VALUES (1, 'a'), (2, 'b')")
+
+ val originalStrategies = spark.experimental.extraStrategies
+ try {
+ // Set the strategy before caching, since caching plans the query in a
cloned session.
+ spark.experimental.extraStrategies = CopyCountingNodeStrategy +:
originalStrategies
+
+ // Stack several nodes above the aliased table relation. Rewriting the
subtree under every
+ // node that the by-name checks visit would copy each of these nodes
once for itself and
+ // once for every node above it, which is quadratic in the plan size.
+ val numNodes = 10
+ val plan = (1 to
numNodes).foldLeft(spark.table(t).queryExecution.analyzed) {
+ (child, _) => CopyCountingNode(child)
+ }
+ cacheManager.cacheQuery(spark, plan, tableName = None, MEMORY_AND_DISK)
+ assert(cacheManager.numCachedEntries == 1)
+ CopyCountingNode.numCopies.set(0)
Review Comment:
This test only proves something while the normalized cache key still has a
`SubqueryAlias` under the `CopyCountingNode`s. If plan normalization (or the
analysis of `spark.table(t)`) ever stops keeping that alias, `numCopies == 0`
would also hold with the old `EliminateSubqueryAliases` code, and the test
would pass vacuously. Could we assert the precondition before resetting the
counter?
```scala
val cacheKey = cacheManager.lookupCachedData(spark, plan).get.plan
assert(cacheKey.exists(_.isInstanceOf[SubqueryAlias]))
CopyCountingNode.numCopies.set(0)
```
(`SubqueryAlias` would need to be added to the `plans.logical` import.)
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -477,7 +489,7 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
name: Seq[String],
resolver: Resolver): Seq[LogicalPlan] = {
cachedData.flatMap { cd =>
- val plan = EliminateSubqueryAliases(cd.plan)
+ val plan = stripRootSubqueryAliases(cd.plan)
Review Comment:
Minor, since the PR description calls out this DSv2 table resolution path:
`findCachedRelations` builds a `Seq` of every matching relation with a strict
`flatMap`, while its only caller, `lookupCachedTable`, uses just the first one
that passes its `collectFirst`. Iterating lazily
(`cachedData.iterator.flatMap(...)`) or folding the extra predicates into a
single `collectFirst` over `cachedData` would stop at the first hit.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -278,6 +278,18 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
}
}
+ /**
+ * Strips the chain of [[SubqueryAlias]] nodes at the root of the given
plan. Callers that only
+ * inspect the root node use this instead of `EliminateSubqueryAliases`,
which rewrites the whole
+ * tree. That matters because [[isMatchedTableOrView]] is evaluated on every
node of every cached
Review Comment:
The strip is semantically needed only where the predicate is applied to an
entry's root: the non-cascade check in `uncacheByCondition` (`shouldRemove =
isMatchedPlan`, the SPARK-54004 case), `findCachedRelations`, and
`tryRefreshPlan`. Inside the `plan.exists(...)` traversals it is redundant,
because `exists` visits the alias's child anyway. Could the scaladoc mention
this, so that a heavier normalization does not come back into this per-node
predicate later?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala:
##########
@@ -278,6 +278,18 @@ class CacheManager extends Logging with
AdaptiveSparkPlanHelper {
}
}
+ /**
+ * Strips the chain of [[SubqueryAlias]] nodes at the root of the given
plan. Callers that only
+ * inspect the root node use this instead of `EliminateSubqueryAliases`,
which rewrites the whole
+ * tree. That matters because [[isMatchedTableOrView]] is evaluated on every
node of every cached
+ * plan, where a full rewrite per node makes cache lookups by name quadratic
in plan size.
+ */
+ @tailrec
+ private def stripRootSubqueryAliases(plan: LogicalPlan): LogicalPlan = plan
match {
Review Comment:
`CommandUtils.uncacheTableOrView` and `CommandUtils.recacheTableOrView` (the
same by-name flow, CommandUtils.scala L494 and L521) and several catalyst
callers (`v2Commands`, `LookupCatalog`, `RewriteDeleteFromTable`,
`RewriteUpdateTable`, `SparkStrategies`, `ddl.scala`) still run a full
`EliminateSubqueryAliases` only to match the root node. Those are not in hot
loops, so this is not blocking, but would it make sense to put this helper next
to `EliminateSubqueryAliases` in catalyst so that those callers can reuse it
later?
##########
sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala:
##########
@@ -1952,6 +1976,53 @@ class CachedTableSuite extends SharedSparkSession
}
}
+ test("cache lookups by table name do not rewrite cached plans") {
Review Comment:
This test covers only the no-match path. Could we also cover the cases that
the PR says stay unchanged?
- The recursive branch of `stripRootSubqueryAliases`: for example, cache
`spark.table(t).as("x")` (`SubqueryAlias(x, SubqueryAlias(testcat.tbl, ...))`)
and check that `uncacheTableOrView(..., cascade = false)` still removes it,
which is the SPARK-54004 case.
- `lookupCachedTable` still finding a relation that is cached behind the
root alias chain.
That would catch a later regression to stripping only one alias level.
--
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]
