sshpuntoff opened a new pull request, #59821:
URL: https://github.com/apache/spark/pull/59821
### What changes were proposed in this pull request?
In `CacheManager.uncacheByCondition`, the condition passed to
`recacheByCondition` now checks whether the entry depends on the uncached plan
before calling `CachedRDDBuilder.isCachedColumnBuffersLoaded`:
```scala
cd.plan.exists(isMatchedPlan) &&
!cd.cachedRepresentation.cacheBuilder.isCachedColumnBuffersLoaded
```
The comment at that call site is updated as well: it described the check as
non-locking, which has not been true since SPARK-53435.
Fixes #59803. This is the `master` version of #59804 (which targets
`branch-4.2`).
### Why are the changes needed?
Since SPARK-53435, `isCachedColumnBuffersLoaded` acquires the builder's
monitor, which `cachedColumnBuffers` holds for the duration of an in-progress
cache materialization (with AQE, `buildBuffers` runs the shuffle stages of a
cached plan that contains a shuffle). `recacheByCondition` evaluated that check
on every cache entry before the plan match, so `Dataset.unpersist()` and
`Catalog.dropTempView()` waited for every in-progress materialization on the
driver, including unrelated ones. For unrelated entries the result is not used.
On a shared driver, one long-running cache materialization therefore delays
cache cleanup for all other jobs.
### Does this PR introduce _any_ user-facing change?
Yes. `unpersist()` and `dropTempView()` no longer wait for unrelated
in-progress cache materializations, restoring the 4.0.0 behavior. Results are
unchanged: the same entries are recached, and the SPARK-53435 locking is
unchanged for entries that depend on the uncached plan.
Repro (spark-shell, `local[8]`):
```scala
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() // before: ~55 s; after: ~10 ms
```
### How was this patch tested?
- New test in `DatasetCacheSuite`: "uncaching an unrelated Dataset does not
wait for an in-progress cache materialization". It blocks a shuffle-backed
cache materialization on a latch and requires `unpersist()` of an unrelated
cache to complete within 10 s. The latch is always released, so a regression
fails rather than hangs. It fails without this change and passes with it.
- On `master`: `CachedTableSuite`, `DatasetCacheSuite`,
`ConcurrentInMemoryRelationSuite`, `InMemoryColumnarQuerySuite` and
`CatalogSuite` pass (226/226).
- Ran the repro above on a 4.2.0 distribution with the patched `spark-sql`
jar: `unpersist()` went from 55,086 ms to 10 ms, and `dropTempView()` from
55,083 ms to 16 ms.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
--
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]
