You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/04/01 15:53:40 UTC

[I] Spark cannot reclaim memory from native operators (spill callback returns 0) [datafusion-comet]

andygrove opened a new issue, #3873:
URL: https://github.com/apache/datafusion-comet/issues/3873

   ## Describe the bug
   
   When Spark's memory manager is under pressure and calls `spill()` on Comet's `NativeMemoryConsumer`, it returns 0 — meaning Spark cannot reclaim any memory from Comet's native operators. This prevents cross-task memory eviction and causes Comet to require significantly more off-heap memory than necessary at scale.
   
   From `CometTaskMemoryManager.java`:
   
   ```java
   private class NativeMemoryConsumer extends MemoryConsumer {
       public long spill(long size, MemoryConsumer trigger) {
           return 0; // No spilling
       }
   }
   ```
   
   ## Impact
   
   When multiple concurrent tasks share a constrained off-heap pool (e.g., 16 tasks sharing 16GB), each task's shuffle writer greedily buffers data until `try_grow()` fails. Since Spark cannot reclaim memory from any of them, tasks compete for the shared pool without coordination, leading to OOM at lower memory settings.
   
   Benchmarking TPC-H SF100 Q9 with `local[4]` showed Comet's memory growing elastically with the offHeap budget (448 MB increase from 4g to 8g), while Spark's native operators stay flat because they participate in the spill protocol.
   
   ## Expected behavior
   
   `NativeMemoryConsumer.spill()` should signal the native memory pool to apply backpressure, causing DataFusion operators (Sort, Aggregate, Shuffle) to spill their internal state to disk. The actual bytes freed should be returned to Spark.
   
   ## Proposed approach
   
   Add a `SpillState` struct with atomics and a condvar for cross-thread coordination:
   
   1. When Spark calls `spill(size)`, JNI into native to set spill pressure
   2. The pool's `try_grow()` checks pressure and returns `ResourcesExhausted`
   3. DataFusion operators catch this and spill internally
   4. As operators call `shrink()`, freed bytes are tracked and returned to Spark
   
   See `docs/source/contributor-guide/memory-management.md` for full analysis including comparison with Gluten's approach.


-- 
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: github-unsubscribe@datafusion.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] Spark cannot reclaim memory from native operators (spill callback returns 0) [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.

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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org