Conversation
Comet's memory model is spread across the JVM config path, the native pool decorators, the Arrow FFI ownership rules, and the container's cgroup, and none of it was written down in one place. The page covers: the allocator inventory (enabling Comet adds several consumers accounted by different parties, not one), where the native budget comes from in each Spark memory mode, the pool decorator stack and task-shared pool lifetime, how DataFusion consumes the pool, the asymmetric charging at the FFI boundary, why declared reservations structurally diverge from RSS, what a Kubernetes or YARN container actually counts, the open problems, and a triage checklist.
github-actions
Bot
added
the
documentation
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Correctness
This adds a contributor reference for memory budgeting and OOM diagnosis, plus its entry in the architecture navigation. The allocator inventory, JVM-to-native budget setup, task-shared registry lifetime, and container sizing are currently spread across several implementations. Bringing them together is useful, and the PR changes no executable behavior.
I checked the page against this head's Comet implementation, the locked DataFusion 55.1.0 and Arrow 59.3.0 sources, and the maintained Spark 3.5 and 4.0 branches. The on-heap and off-heap budget formulas, default pool choices, separate on-heap shuffle allocator, and weak-reference registry/drop race match those paths. Spark's resource calculation also includes configured off-heap memory in the executor container total.
Four statements need correction before this becomes the reference: ordinary JVM heap OOM can terminate the executor, Spark cannot trigger native spilling through the current callback, the fair pool checks aggregate task-pool usage against its divided limit, and FFI direction does not determine whether an operator reserves the buffers. The inline comments give the concrete paths and counterexamples. These are documentation findings, not new runtime regressions introduced by this PR. Expression results, types, null handling, ANSI behavior and fallback are unchanged.
Validation
At the September 15, 05:09 UTC refresh, GitHub reports 8 successful and 12 skipped checks. The native builds, Spark 3.4/3.5/4.0/4.1 SQL suites, Iceberg suites and benchmark job are skipped. I reviewed the complete two-file change, verified the cited source paths and local documentation links, and checked diff whitespace. No local Spark/JNI execution, product build or benchmark was run for this documentation-only change. The author's formatting check is an author claim, not a locally repeated result. Maintained Spark 3.4 and 4.1 sources were unavailable, so no source compatibility coverage is claimed for those versions.
Performance
This page adds no allocations, synchronization, copies or work to an execution path. A microbenchmark would not validate this change. The material performance concern is the accuracy of tuning guidance: the fair-pool example currently predicts more memory per task than its admission check permits, and the spill discussion promises reclamation that Spark cannot request from native operators. Correcting those descriptions will help contributors interpret early spilling and failed allocations without assuming the pool is enforcing a physical RSS ceiling.
Design
The sequence from allocators through budgets, ownership and OOM diagnosis provides a useful map of the existing design. Keeping the proposed allocator-level accounting work in a linked open issue also avoids turning this page into a design proposal. The important boundary to preserve is between a buffer's allocator, the operator's voluntary reservation, and the runtime holding the last reference. Those can differ in both FFI directions. The page should explain that independent relationship and describe the current one-way spill integration.
Abstraction & complexity
No runtime abstraction is introduced. The pool-stack diagram and task-registry explanation make the existing decorators and RAII ownership easier to follow. The main simplifications needed are conceptual: avoid treating the JVM heap like a voluntarily declared reservation pool, avoid calling the aggregate fair-pool threshold a per-consumer guarantee, and avoid deriving accounting from transfer direction. Correcting these specific claims is sufficient without adding a new accounting design to this PR.
There was a problem hiding this comment.
Correctness
[P2] Correct the executor outcome for a JVM heap OOM
Could we distinguish a real JVM heap OutOfMemoryError from SparkOutOfMemoryError here? In Spark 3.5 and 4.0, Executor.isFatalError exempts SparkOutOfMemoryError, but treats the ordinary JVM error as fatal and invokes SparkUncaughtExceptionHandler, which exits the process with SparkExitCode.OOM. The executor therefore does not normally survive the heap exhaustion described in this row. This distinction is central to the triage guidance: an executor loss does not imply a cgroup kill. Please describe heap exhaustion as potentially executor-fatal and reserve task-level recovery for Spark's managed-memory allocation error.
|
|
||
| - Comet competes with Spark's own off-heap consumers (Tungsten sorters, `BytesToBytesMap`, and so | ||
| on) for the same `spark.memory.offHeap.size`, and Spark's unified memory manager arbitrates. | ||
| - Spark can force _Spark's_ consumers to spill to satisfy Comet's request, and vice versa. |
There was a problem hiding this comment.
Correctness
[P2] Document that Spark cannot trigger native spilling
Could we remove and vice versa and call out the missing native spill callback? NativeMemoryConsumer.spill() always returns zero. A Comet allocation can make Spark spill eligible JVM consumers in the same task, but a Spark allocation cannot make a native sorter or aggregate release its reservations through that callback. Native spilling occurs when the native operator handles its own failed reservation. As written, this promises a recovery path that is unavailable when a JVM consumer is blocked behind native reservations.
| - A partial grant (`acquired < additional`) is released immediately and reported as | ||
| `ResourcesExhausted`, which is the signal DataFusion uses to spill. | ||
|
|
||
| `CometFairMemoryPool` additionally caps each registered consumer at `pool_size / num_consumers` |
There was a problem hiding this comment.
Correctness
[P2] Describe the fair pool's actual aggregate admission limit
Could we describe the implemented check rather than a per-consumer quota? CometFairMemoryPool::try_grow compares pool_size / num_consumers with state.used + additional, where state.used is the total for the shared pool, and does not use the requesting reservation's size. For example, with an 8 GiB pool and two registered consumers, if one has reserved 3 GiB, a 2 GiB request from the other is rejected even though neither consumer would exceed 4 GiB. The current wording materially overstates usable memory for multi-operator tasks and would mislead spill tuning. Please document this aggregate threshold and its task scope.
There was a problem hiding this comment.
Correctness
[P2] Separate FFI buffer ownership from operator reservations
Could we revise the direction-based accounting rule and the two examples above it? With the pinned DataFusion 55.1.0, ExternalSorter::reserve_memory_for_batch_and_maybe_spill and hash join's collect_left_input reserve the incoming batch's memory before retaining it, regardless of which allocator produced the buffers. Imported JVM batches can therefore be charged to Comet and, through a unified pool, to Spark. In the other direction, the sort output's ReservationStream::poll_next shrinks its reservation before returning each batch. Comet's prepare_output / move_to_spark exports buffer ownership without attaching a reservation to the release callback, so a JVM-held output need not remain pool-charged until close(). The real gap is that buffer and reservation lifetimes are independent, not that JVM-to-native buffers are always unaccounted. That distinction changes how contributors should diagnose and fix the gap.
2 participants
Which issue does this PR close?
Relates to #4576. It does not close it: this documents the memory model as it
exists today, including the gaps that issue is about, rather than changing any
behavior.
Rationale for this change
Comet's memory model is spread across four places that are individually
reasonable and collectively hard to hold in your head: the JVM config path that
computes the budget, the native memory-pool decorators, the Arrow FFI ownership
rules, and the container's cgroup. None of it was written down in one place.
The practical cost shows up when triaging an out-of-memory report. Answering
"which budget did this exceed, and who was supposed to be counting?" currently
means reading
CometExecIterator.getMemoryConfig,parse_memory_pool_config,the pool implementations, the FFI export paths, and Spark's
ResourceProfile.It also makes it hard to reason about proposed fixes, because the baseline they
would improve on is not written down.
I started this while reviewing #4582 and found that most of what I needed to
explain was existing behavior, not the prototype. The prototype is a separate
question; this page is useful either way.
What changes are included in this PR?
A single new contributor-guide page,
memory_management.md, registered underProject Architecture. No behavior changes.
Sections:
one, and they are not accounted by the same party. Notably
CometArrowAllocatoris a process-wide
RootAllocator(Long.MaxValue)that is unbounded and invisibleto both Spark's
TaskMemoryManagerand Comet's native pool; the JVM shuffleallocator switches between a Spark
MemoryConsumer(off-heap) and aself-bounded
UnsafeMemoryAllocator(on-heap); and on-heap mode sizes thenative pool and the JVM shuffle pool from the same
spark.comet.memoryOverhead,so it can occupy roughly twice that figure.
memory_limit/memory_limit_per_taskeach pool type is sized from.task-shared pool's RAII lifetime including the acquire/drop race.
produced a batch and the runtime that decides when it dies can be on opposite
sides. JVM-to-native batches are charged to nothing while native pins them;
native-to-JVM batches stay charged until the JVM closes them.
spark.comet.exec.memoryPool.fractionexists.executor.memory + memoryOverhead + offHeap.size + pyspark, sospark.memory.offHeap.sizeis inside the pod limit rather than headroom on topof it.
How are these changes tested?
Documentation only; there is no behavior to test. Every claim was checked against
the code on
mainor against the referenced Spark source rather than written frommemory, and
npx prettier --checkpasses.Marked as draft because I would like a sanity check on the accuracy of the
"Who allocates what" and "Crossing the FFI boundary" sections in particular
before this becomes the reference people cite.