andygrove commented on issue #3873: URL: https://github.com/apache/datafusion-comet/issues/3873#issuecomment-5804752757
I prototyped the approach in this issue, a per-task registry of spillable native operators driven over JNI from `NativeMemoryConsumer.spill()`, and I don't think it is worth pursuing in this form. Recording what I found so the next attempt can start from here. **Corrections to the issue description** - Spark does not evict memory across tasks. `TaskMemoryManager.acquireExecutionMemory` only calls `spill()` on consumers of the requesting task (the same in Spark 3.4 through 4.1). Between tasks, `ExecutionMemoryPool` just waits for memory to be released. So the only possible benefit is within a single task. - The proposed design (make `try_grow` fail, then wait on a condvar until operators spill) deadlocks. A native plan with JVM inputs is polled by `block_on` on the task thread, and that is the same thread that is blocked inside `acquireExecutionMemory` waiting for the spill, so nothing is left to poll the operators. - There is no cheap "shrink" half (item 2 of #5997). `CometUnifiedMemoryPool` and `CometFairMemoryPool` acquire exactly what an operator reserves and release it on shrink, so the pool never holds unused capacity to give back. **What can actually be spilled on request** DataFusion's sort, aggregate, hash join, sort-merge join and window operators keep their state private and have no external spill entry point. The only native memory holder that Comet owns and can spill is the native shuffle writer's `MultiPartitionShuffleRepartitioner`. So the prototype could only reclaim: - the native shuffle writer's buffered batches, - for multi-partition shuffles written to local disk (the Celeborn writer reserves memory while it writes, so spilling it from inside Spark's memory manager would reserve from the pool being relieved), - only while the writer is idle waiting for input. A spill must not block, because Spark calls it while holding the task's memory manager monitor, so a writer that is mid-insert frees nothing. That includes the insert whose own reservation triggered the request. It does work end to end: with 32 MB off-heap and five concurrent tasks, a partial aggregate feeding a native shuffle triggered about 80 spills of roughly 6 MB each from the writer. But the realistic benefit is narrow: - A stage whose only memory consumer is the shuffle writer gains nothing: the writer is always the requester and already spills itself. - With a partial aggregate upstream, the gain is roughly zero. Without this change a refused partial aggregate emits early, those rows land in the writer, and the writer spills anyway. The spill just moves. - With a spillable operator upstream (sort, final aggregate), it trades that operator's spill for the writer's. Possibly cheaper, not measured. - The one case with a clear payoff is a reservation that cannot spill (the sort merge phase, a hash join build side, sort-merge join buffering) failing with `ResourcesExhausted` while the writer holds memory in the same task. - With the default `fair_unified` pool, the per-consumer fair limit can refuse a request before it ever reaches Spark, in which case Spark never asks. The writer is often the largest holder because `spark.comet.shuffle.native.maxBufferBytes` defaults to unlimited, but setting that limit already bounds it without a cross-language protocol. **A hazard for any future attempt** No pool may hold its own lock across a JNI call to Spark. `CometFairMemoryPool` holds its `state` mutex across Spark `acquire` and `release` calls. That is harmless today, but once `spill()` can shrink the pool it deadlocks: thread A holds the task memory manager monitor and spills, which needs the pool lock, while thread B holds the pool lock and waits for the monitor in `acquireExecutionMemory`. **What would make this worthwhile** Reaching the operators that hold most of the memory. That needs an external spill API for sort and aggregate in DataFusion, as suggested earlier in this thread, or a DataFusion-level memory arbitrator. Until then I don't plan to open a PR for the prototype. -- 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]
