[
https://issues.apache.org/jira/browse/SPARK-60584?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Samuel Shpuntoff updated SPARK-60584:
-------------------------------------
Description:
{\{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}
Stock 4.2.0, 30 s fill, unpersist issued 5 s in:
||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
was:
{{Dataset.unpersist()}} and {{Catalog.dropTempView()}} wait until every
in-progress cache materialization on the driver completes, including those
unrelated to the plan being uncached. A single long-running materialization
therefore delays cache cleanup for all other jobs on the driver.
h3. Repro
spark-shell, {{{}local[8]{}}}:
import org.apache.spark.sql.functions._val small = spark.range(10).cache();
small.count()val sleepy = udf((x: Long) => { Thread.sleep(60000); 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() // 4.0.0: ~9 ms. 4.2.0: ~55 s
h3. Cause
{{CacheManager.recacheByCondition}} evaluates {{isCachedColumnBuffersLoaded}}
on every cache entry before checking whether the entry depends on the uncached
plan. Since SPARK-53435, this method acquires the builder's monitor, which is
held for the duration of an in-progress materialization. For unrelated entries
the result is not used.
Affects 4.0.0+ and branch-3.5
([#52199|https://github.com/apache/spark/pull/52199]).
h3. Proposed fix
Evaluate the plan match first:
cd.plan.exists(isMatchedPlan) &&
!cd.cachedRepresentation.cacheBuilder.isCachedColumnBuffersLoaded
With this change the repro completes in ~10 ms and the existing cache test
suites pass. A regression test is included.
A PR will be added as a comment shortly.
----
Generated-by: Claude Code (Claude Opus 5.5)
Github Issue:
[https://github.com/apache/spark/issues/59803]
(This is my first issue, so apologies for any mistakes on the form template)
> 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
>
> {\{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}
> Stock 4.2.0, 30 s fill, unpersist issued 5 s in:
> ||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]
