Raviston6296 opened a new pull request, #59915:
URL: https://github.com/apache/spark/pull/59915
### What changes were proposed in this pull request?
Backport of #59821 to `branch-4.0`.
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 too. It described the check as
non-locking, which has not been true since SPARK-53435.
The `CacheManager` change applies cleanly. The test file needed a manual
resolution only because `master` has two later tests in `DatasetCacheSuite`
(SPARK-53418, SPARK-54812) that are not on `branch-4.0`; the added regression
test is identical to the one in #59821.
### Why are the changes needed?
Since SPARK-53435 (#52199), `isCachedColumnBuffersLoaded` 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, `buildBuffers` materializes the shuffle query stage before
returning, so the monitor is held for that stage's full duration.
`recacheByCondition` evaluated that check on every cache entry before the
plan match, so `Dataset.unpersist()` and `Catalog.dropTempView()` waited for
any in-progress shuffle-backed cache materialization on the driver, including
unrelated ones whose result is never used. One long-running materialization
therefore delays cache cleanup for all other work on the driver.
This affects 4.0.1+ (SPARK-53435 was backported to `branch-4.0`). 4.0.0 is
not affected.
### 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 from #59803 (spark-shell, `local[8]`; timings were measured there on
4.2.0):
```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() // before: ~25 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.
- On `branch-4.0` with JDK 17: `build/sbt 'sql/testOnly *DatasetCacheSuite
*CachedTableSuite'` passes (93/93).
- With the `CacheManager` change reverted, the new test fails with
`TimeoutException: Future timed out after [10 seconds]`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
