[
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||{Unknown macro: \\{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()}} 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||{Unknown macro: \{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]
> 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||{Unknown macro: \\{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]
