You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by vi...@apache.org on 2026/10/14 17:10:38 UTC
(spark) branch branch-4.3 updated: [SPARK-60228][SQL] Avoid quadratic cache lookups by table name in CacheManager
This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new cb6500150ee0 [SPARK-60228][SQL] Avoid quadratic cache lookups by table name in CacheManager
cb6500150ee0 is described below
commit cb6500150ee0ab7de22d4510e4891b5c2e47a63d
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Wed Oct 14 10:10:08 2026 -0700
[SPARK-60228][SQL] Avoid quadratic cache lookups by table name in CacheManager
### What changes were proposed in this pull request?
SPARK-54004 changed `CacheManager.isMatchedTableOrView` to match on
`EliminateSubqueryAliases(plan)`, which rewrites the whole tree. The check is evaluated on
every node of every cached plan (via `plan.exists`) when uncaching or recaching by table name,
so each scan rewrites every subtree of every cached plan.
This PR strips only the chain of `SubqueryAlias` nodes at the root, since all matched cases
inspect only the root node. `tryRefreshPlan` and `findCachedRelations`, which also only look
at the root, use the same helper. The matching result is unchanged.
### Why are the changes needed?
The by-name scan runs on every DSv2 write (`refreshCache` -> `recacheTableOrView`) and on DDL
such as DROP / CREATE OR REPLACE / ALTER VIEW. With deep cached plans, the cost grows
super-linearly. Measured locally for a by-name scan that matches no entry, with plans built
from chained filters:
| plan depth x entries | before | after |
|---|---|---|
| 100 x 10 | 26 ms | 0.2 ms |
| 300 x 10 | 720 ms | 0.1 ms |
| 1000 x 10 | ~40 s | 0.4 ms |
`findCachedRelations` is also on the DSv2 table resolution path, where it previously
rewrote each cached plan once per lookup.
### Does this PR introduce _any_ user-facing change?
No, other than the performance improvement.
### How was this patch tested?
New test in `CachedTableSuite` that counts plan copies during by-name cache lookups (fails
before this change with 185 copies, passes with 0), plus existing `CachedTableSuite`,
`DataSourceV2DataFrameSuite` and `DatasetCacheSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5.5)
Closes #59448 from viirya/cache-manager-root-alias-strip.
Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
(cherry picked from commit 5c5525dd33e7d1b0b8a7b26ea4c986fe3dcfeb49)
Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
.../apache/spark/sql/execution/CacheManager.scala | 38 ++++---
.../org/apache/spark/sql/CachedTableSuite.scala | 111 +++++++++++++++++++--
2 files changed, 128 insertions(+), 21 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
index 9f4ee12b8cec..e14a9a73aa5c 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
@@ -17,18 +17,18 @@
package org.apache.spark.sql.execution
+import scala.annotation.tailrec
import scala.util.control.NonFatal
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.internal.{Logging, MessageWithContext}
import org.apache.spark.internal.LogKeys._
-import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
import org.apache.spark.sql.catalyst.analysis.Resolver
import org.apache.spark.sql.catalyst.catalog.HiveTableRelation
import org.apache.spark.sql.catalyst.expressions.{Attribute, SubqueryExpression}
import org.apache.spark.sql.catalyst.optimizer.EliminateResolvedHint
-import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan, ResolvedHint, View}
+import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan, ResolvedHint, SubqueryAlias, View}
import org.apache.spark.sql.catalyst.trees.TreePattern.PLAN_EXPRESSION
import org.apache.spark.sql.catalyst.util.sideBySide
import org.apache.spark.sql.classic.{Dataset, SparkSession}
@@ -238,28 +238,21 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
name: Seq[String],
cascade: Boolean,
blocking: Boolean = false): Unit = {
+ val resolver = spark.sessionState.conf.resolver
uncacheByCondition(
spark,
- isMatchedTableOrView(_, name, spark.sessionState.conf, includeTimeTravel = true),
+ isMatchedTableOrView(_, name, resolver, includeTimeTravel = true),
cascade,
blocking)
}
- private def isMatchedTableOrView(
- plan: LogicalPlan,
- name: Seq[String],
- conf: SQLConf,
- includeTimeTravel: Boolean): Boolean = {
- isMatchedTableOrView(plan, name, conf.resolver, includeTimeTravel)
- }
-
private def isMatchedTableOrView(
plan: LogicalPlan,
name: Seq[String],
resolver: Resolver,
includeTimeTravel: Boolean): Boolean = {
- EliminateSubqueryAliases(plan) match {
+ stripRootSubqueryAliases(plan) match {
case LogicalRelationWithTable(relation, Some(catalogTable)) =>
isSameName(name, catalogTable.identifier.nameParts, resolver) &&
(includeTimeTravel || !isTimeTravelRelation(relation))
@@ -278,6 +271,20 @@ 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.
+ * Inside those `exists` traversals the strip is redundant, since the alias's child is visited
+ * anyway; it is needed only where the predicate is applied to an entry's root.
+ */
+ @tailrec
+ private def stripRootSubqueryAliases(plan: LogicalPlan): LogicalPlan = plan match {
+ case SubqueryAlias(_, child) => stripRootSubqueryAliases(child)
+ case other => other
+ }
+
private def isTimeTravelRelation(relation: BaseRelation): Boolean = relation match {
case hadoopFsRelation: HadoopFsRelation => hadoopFsRelation.location.isTimeTravel
case _ => false
@@ -366,8 +373,9 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
spark: SparkSession,
name: Seq[String],
includeTimeTravel: Boolean = true): Unit = {
+ val resolver = spark.sessionState.conf.resolver
def shouldInvalidate(entry: CachedData): Boolean = {
- entry.plan.exists(isMatchedTableOrView(_, name, spark.sessionState.conf, includeTimeTravel))
+ entry.plan.exists(isMatchedTableOrView(_, name, resolver, includeTimeTravel))
}
recacheByCondition(spark, shouldInvalidate)
}
@@ -429,7 +437,7 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
*/
private def tryRefreshPlan(spark: SparkSession, plan: LogicalPlan): Option[LogicalPlan] = {
try {
- EliminateSubqueryAliases(plan) match {
+ stripRootSubqueryAliases(plan) match {
case r @ ExtractV2CatalogAndIdentifier(catalog, ident)
if r.timeTravelSpec.isEmpty && !r.table.isInstanceOf[ChangelogTable] =>
val table = CatalogV2Util.getTable(catalog, ident, options = r.options)
@@ -477,7 +485,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)
plan match {
case r @ ExtractV2CatalogAndIdentifier(catalog, ident)
if isSameName(name, catalog, ident, resolver) && r.timeTravelSpec.isEmpty =>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
index 66732a9e52b0..015e71d39937 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
@@ -21,7 +21,7 @@ import java.io.{File, FilenameFilter}
import java.nio.file.{Files, Paths}
import java.time.{Duration, LocalDateTime, LocalTime, Period}
import java.util
-import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import scala.collection.mutable.HashSet
import scala.concurrent.duration._
@@ -34,8 +34,8 @@ import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent, SparkListe
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.AsOfVersion
import org.apache.spark.sql.catalyst.analysis.TempTableAlreadyExistsException
-import org.apache.spark.sql.catalyst.expressions.SubqueryExpression
-import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, Join, JoinStrategyHint, SHUFFLE_HASH}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, SubqueryExpression}
+import org.apache.spark.sql.catalyst.plans.logical.{BROADCAST, Join, JoinStrategyHint, LogicalPlan, SHUFFLE_HASH, SubqueryAlias, UnaryNode}
import org.apache.spark.sql.catalyst.util.DateTimeConstants
import org.apache.spark.sql.connector.catalog.BasicInMemoryTableCatalog
import org.apache.spark.sql.connector.catalog.CatalogPlugin
@@ -44,7 +44,7 @@ import org.apache.spark.sql.connector.catalog.Identifier
import org.apache.spark.sql.connector.catalog.InMemoryCatalog
import org.apache.spark.sql.connector.catalog.TableWritePrivilege
import org.apache.spark.sql.connector.catalog.TruncatableTable
-import org.apache.spark.sql.execution.{ColumnarToRowExec, ExecSubqueryExpression, RDDScanExec, SparkPlan, SparkPlanInfo}
+import org.apache.spark.sql.execution.{ColumnarToRowExec, ExecSubqueryExpression, RDDScanExec, SparkPlan, SparkPlanInfo, SparkStrategy}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanHelper, AQEPropagateEmptyRelation}
import org.apache.spark.sql.execution.columnar._
import org.apache.spark.sql.execution.command.CommandUtils
@@ -55,15 +55,39 @@ import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{StringType, StructField, StructType}
-import org.apache.spark.sql.util.PartitionKeyedAccumulator
+import org.apache.spark.sql.util.{CaseInsensitiveStringMap, PartitionKeyedAccumulator}
import org.apache.spark.storage.{RDDBlockId, StorageLevel}
-import org.apache.spark.storage.StorageLevel.{MEMORY_AND_DISK_2, MEMORY_ONLY}
+import org.apache.spark.storage.StorageLevel.{MEMORY_AND_DISK, MEMORY_AND_DISK_2, MEMORY_ONLY}
import org.apache.spark.tags.SlowSQLTest
import org.apache.spark.unsafe.types.CalendarInterval
import org.apache.spark.util.{AccumulatorContext, Utils}
private case class BigData(s: String)
+/**
+ * A pass-through node that counts how many times it is copied with a new child, i.e. how many
+ * times a plan containing it is rewritten.
+ */
+private case class CopyCountingNode(child: LogicalPlan) extends UnaryNode {
+ override def output: Seq[Attribute] = child.output
+
+ override protected def withNewChildInternal(newChild: LogicalPlan): CopyCountingNode = {
+ CopyCountingNode.numCopies.incrementAndGet()
+ copy(child = newChild)
+ }
+}
+
+private object CopyCountingNode {
+ val numCopies = new AtomicInteger(0)
+}
+
+private object CopyCountingNodeStrategy extends SparkStrategy {
+ override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+ case CopyCountingNode(child) => planLater(child) :: Nil
+ case _ => Nil
+ }
+}
+
@SlowSQLTest
class CachedTableSuite extends SharedSparkSession
with AdaptiveSparkPlanHelper {
@@ -1952,6 +1976,81 @@ 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)
+ // Without an alias in the cache key, the old full-tree rewrite would not copy anything
+ // either, and the assertions below would pass vacuously.
+ val cacheKey = cacheManager.lookupCachedData(spark, plan).get.plan
+ assert(cacheKey.exists(_.isInstanceOf[SubqueryAlias]))
+ CopyCountingNode.numCopies.set(0)
+
+ // Scanning the cache for a table that is not referenced must leave the entry alone.
+ val otherName = Seq("testcat", "other")
+ cacheManager.uncacheTableOrView(spark, otherName, cascade = true)
+ cacheManager.uncacheTableOrView(spark, otherName, cascade = false)
+ cacheManager.recacheTableOrView(spark, otherName)
+ // The cached plan is not rooted at the relation, so this finds nothing; it is called to
+ // check that it does not rewrite the plan either.
+ assert(cacheManager.lookupCachedTable(
+ catalog("testcat"),
+ Identifier.of(Array(), "tbl"),
+ tableId = None,
+ CaseInsensitiveStringMap.empty(),
+ spark.sessionState.conf.resolver).isEmpty)
+ assert(cacheManager.numCachedEntries == 1)
+ assert(CopyCountingNode.numCopies.get() == 0)
+
+ // The entry still refers to the table, so uncaching it with cascade must remove it.
+ cacheManager.uncacheTableOrView(spark, Seq("testcat", "tbl"), cascade = true)
+ assert(cacheManager.numCachedEntries == 0)
+ } finally {
+ spark.experimental.extraStrategies = originalStrategies
+ }
+ }
+ }
+
+ test("cache lookups by table name see through a chain of root aliases") {
+ 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')")
+
+ // The cache key is rooted at SubqueryAlias(x, SubqueryAlias(testcat.tbl, relation)).
+ val df = spark.table(t).as("x")
+ df.cache()
+ assert(df.queryExecution.analyzed.asInstanceOf[SubqueryAlias].child
+ .isInstanceOf[SubqueryAlias])
+
+ assert(cacheManager.lookupCachedTable(
+ catalog("testcat"),
+ Identifier.of(Array(), "tbl"),
+ tableId = None,
+ CaseInsensitiveStringMap.empty(),
+ spark.sessionState.conf.resolver).isDefined)
+
+ cacheManager.uncacheTableOrView(spark, Seq("testcat", "tbl"), cascade = false)
+ assert(cacheManager.lookupCachedData(df).isEmpty)
+ }
+ }
+
test("cache DSv2 table with time travel") {
val t = "testcat.tbl"
val version = "v1"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org