[
https://issues.apache.org/jira/browse/SPARK-60584?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated SPARK-60584:
-----------------------------------
Labels: patch pull-request-available (was: patch)
> unpersist() waits for unrelated cache materializations to complete
> ------------------------------------------------------------------
>
> Key: SPARK-60584
> URL: https://issues.apache.org/jira/browse/SPARK-60584
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.1.0, 4.0.1, 3.5.7, 4.2.0
> Environment: Spark 4.2 with Java 25 on ARM64
> Reporter: Samuel Shpuntoff
> Priority: Major
> Labels: patch, pull-request-available
>
> {{Dataset.unpersist()}} and {{Catalog.dropTempView()}} can block on an
> unrelated in-progress cache materialization: when another thread is filling a
> cache whose plan contains a shuffle (with AQE enabled), any non-cascade
> uncache on the driver waits until that fill's shuffle stage completes. A
> single long-running materialization therefore delays cache cleanup for all
> other work on the driver.
> h3. Repro
> spark-shell, {{{}local[8]{}}}:
> {code:scala}
> import org.apache.spark.sql.functions._
> val small = spark.range(10).cache(); small.count()
> val sleepy = udf((x: Long) => { Thread.sleep(30000); x })
> val slow = spark.range(0, 2, 1,
> 2).select(sleepy(col("id"))).repartition(4).cache()
> new Thread(() => slow.count()).start(); Thread.sleep(5000)
> small.unpersist()
> {code}
> ||Fill plan||AQE||small.unpersist()||
> |with repartition(4)|on|25,102 ms|
> |without repartition(4)|on|17 ms|
> |with repartition(4)|off|24 ms|
> h3. Cause
> {{CacheManager.recacheByCondition}} evaluates {{isCachedColumnBuffersLoaded}}
> on every cache entry before checking whether the entry depends on the
> uncached plan. Since SPARK-53435, that method is {{synchronized}} on the
> entry's {{{}CachedRDDBuilder{}}}. {{cachedColumnBuffers}} holds the same
> monitor while building the cached RDD; when the cached plan contains a
> shuffle and AQE is enabled, {{cachedPlan.execute()}} materializes the shuffle
> query stage before returning, so the monitor is held for that stage's full
> duration. For unrelated entries the result is not used.
> Affects 3.5.7+ and 4.0.1+ (introduced by SPARK-53435). 3.5.6 and 4.0.0 are
> not affected.
> h3. Proposed fix
> Evaluate the plan match first:
> {code:scala}
> cd.plan.exists(isMatchedPlan) &&
> !cd.cachedRepresentation.cacheBuilder.isCachedColumnBuffersLoaded
> {code}
> With this change the repro completes in ~10 ms and the existing cache test
> suites pass; a regression test is included.
> PR (master): [https://github.com/apache/spark/pull/59821]
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
