Memory Management

This page describes how memory is budgeted, accounted, and enforced across the JVM/native boundary. It is aimed at contributors working on memory pools, operators that reserve memory, or anyone debugging an out-of-memory report. For user-facing tuning advice, see the Memory Tuning guide.

This page covers off-heap mode (spark.memory.offHeap.enabled=true) only. Comet also has an on-heap mode, but it exists so that the Spark SQL test suite can run against Comet without changing Spark’s memory configuration. Comet performs no memory accounting in it: the native side gets DataFusion’s UnboundedMemoryPool and the JVM shuffle allocator (CometUnboundedShuffleMemoryAllocator) hands out Unsafe pages against no budget. Native memory is not on the JVM heap, so there is no Spark pool it could honestly be charged to, and the fixed-size pool that used to stand in for one bounded nothing the container cares about. On-heap mode must not be used in production, and it is not described further here.

Overview

A Comet executor has to satisfy three separate memory budgets at once, and they are enforced by three different parties:

Budget	Enforced by	What happens when it is exceeded
JVM heap	The JVM	java.lang.OutOfMemoryError; Spark treats it as fatal and exits the executor
Spark’s memory pools	Spark bookkeeping	A consumer is asked to spill, or the task fails with SparkOutOfMemoryError
Container RSS	The OS / cgroup	SIGKILL of the whole executor process (exit 137, OOMKilled)

The first two are accounting: a running total of bytes that consumers have voluntarily declared. The third is physical: the kernel measures resident pages and does not care what any accounting layer believes.

The first row is easy to get wrong. Spark’s Executor.isFatalError exempts SparkOutOfMemoryError, which is the error Spark’s own memory manager throws when a consumer cannot get the bytes it asked for, so that failure is confined to the task and the executor keeps running. An ordinary JVM OutOfMemoryError is not exempt: the executor hands it to SparkUncaughtExceptionHandler, which calls System.exit(SparkExitCode.OOM) (exit code 52). Real heap exhaustion therefore usually costs the whole executor, not one task. An executor loss on its own does not tell you which budget was exceeded; the exit code does.

Comet’s difficulty is that its allocations are made by Rust code, so no JVM allocator produces them and no JVM metric measures them, yet they land squarely in container RSS. Comet therefore maintains its own budget that is meant to shadow the physical one, and declares it to Spark so that the two compete for a single number. The accuracy of that shadow is the central problem this page is about.

The picture the Memory Tuning guide gives users is deliberately simple:

Most of this page is about the line between Comet’s share of the off-heap pool and its share of the memory overhead: which allocations are declared to the pool, and which land in the overhead with nothing tracking them.

Who allocates what

Enabling Comet does not add one new memory consumer, it adds several, and they are not all accounted by the same party. This inventory is worth internalizing before reading the rest of the page:

Allocator	Lives in	Bounded by	Visible to Spark?
Spark execution + storage (on-heap)	JVM heap	spark.executor.memory and the unified memory manager	Yes
Spark Tungsten (off-heap)	Off-heap	spark.memory.offHeap.size via TaskMemoryManager	Yes
Comet native (Rust global allocator)	Native heap	memory_limit (see below), enforced only via the memory pool	Reservations only
Comet JVM Arrow (CometArrowAllocator)	Off-heap	Nothing: a RootAllocator(Long.MaxValue)	No
Comet JVM shuffle pages	Off-heap	spark.memory.offHeap.size via TaskMemoryManager	Yes
Embedded Python and PyArrow (python-udf feature)	Native heap	Nothing: CPython’s and PyArrow’s own allocators	No

Several observations follow.

“Off-heap” and “native heap” are not the same thing. Both sit outside the JVM heap and both count toward container RSS, which is what makes them easy to conflate, but they differ in who allocates the bytes and in who is able to see them.

• Off-heap is allocated by JVM code: sun.misc.Unsafe for Spark’s Tungsten pages, and Arrow’s Unsafe-backed allocator for Comet’s Java Arrow buffers. A JVM-side allocator makes the call, so a JVM-side consumer is in a position to report it, and spark.memory.offHeap.size together with TaskMemoryManager arbitrates it.

• Native heap is allocated by Rust through its global allocator, in the same process. No JVM code makes the call, so no JVM-side allocator or metric ever measures these bytes. The only layer that sees any of them is Comet’s own memory pool, and then only the portion that operators explicitly reserve. That portion is still charged to Spark, as the next paragraph describes; what is never reserved is measured by nothing and budgeted by nobody.

The budget is shared even though the memory is not. memory_limit is derived from spark.memory.offHeap.size (see Where Comet’s budget comes from), and in both off-heap pool types every native reservation is forwarded to Spark’s off-heap execution pool over JNI. A native allocation therefore consumes the same accounting budget as a Tungsten page while occupying entirely different memory. Raising spark.memory.offHeap.size raises the ceiling for both at once, and raises the pod’s memory request by the same amount.

Comet’s JVM-side Arrow allocator is unbounded and accounted by nobody. CometArrowAllocator (spark/src/main/scala/org/apache/comet/package.scala) is a single process-wide new RootAllocator(Long.MaxValue). Child allocators are cut from it for FFI stream export (CometNativeArrowSource), broadcast coalescing, the decoded broadcasts that an executor’s tasks share (SharedDecode), and CometSparkToColumnarExec. These are real off-heap bytes in container RSS that neither Spark’s TaskMemoryManager nor Comet’s native memory pool sees. In practice the volume is modest, a batch at a time per stream, but there is no ceiling and no backpressure. A broadcast is the largest case: a task that reads one holds all of it decoded, and the tasks of an executor that read it at the same time share one such copy. The executor’s memory usage log reports the allocator’s total, and the part of it charged to the import allocator described below, so that the overhead can be sized for it.

Sharing lowers the bytes that are resident, not the bytes that are reserved. Each task’s native join reserves its whole build side, because DataFusion skips only the buffers that the same plan has already counted, so with N tasks sharing one copy the pool holds N times the bytes that are resident, and spilling behaves as it did before. The memory usage log’s untracked figure, allocated + jvmArrow.allocatedByJvm - reserved, subtracts all N reservations while counting the one copy in jvmArrow, so it undercounts by the difference while tasks share a broadcast. With 8 tasks sharing a 100 MB build side the shared copy is 100 MB and reserved includes 800 MB for it, and the figure falls by 700 MB, which can hide other untracked memory in the log and in the limit warning, or clamp it to zero.

Charging these buffers to Spark’s off-heap pool instead was tried in #5998 and dropped. Spark cannot make Comet’s native consumer release anything, and native operators fill a task’s share before they spill, so the JVM allocation that decodes their next input batch is refused first. Refusing it fails the task where native would have spilled; not refusing it bounds nothing when the pool is full. It needs native reclaim first; see #3873.

One further child, CometArrowImportAllocator (comet-ffi-imports), is what the Arrow C Data Interface import path allocates from, so that tracing can report those charges apart from the rest of Comet’s Arrow memory. Unlike the others it is process-wide and never closed, because imported buffers are reference counted and routinely outlive the task that imported them. Its reservation is zero, so every byte still escalates to the root and the inventory above is unchanged by it. Being charged there is not the same as having been allocated there; see the scaladoc on the allocator before reading anything into the split.

The JVM shuffle allocator is an ordinary Spark consumer. CometShuffleMemoryAllocator.getInstance returns CometUnifiedShuffleMemoryAllocator, a Spark MemoryConsumer drawing from spark.memory.offHeap.size, so shuffle pages are arbitrated against Spark’s other consumers in the same task like any other allocation. A page or pointer array allocation that waits in Spark for memory can wake up to find the task’s entry gone from Spark’s execution pool; see Constraints on a Comet memory consumer. The allocator retries it, and after three attempts throws SparkOutOfMemoryError, which the shuffle writers handle like any other refused page.

Which allocator each call site uses, and who ends up charged for the bytes:

Constraints on a Comet memory consumer

CometTaskMemoryManager is the one place where Comet code acts as a Spark MemoryConsumer, and the rules below are why it is written the way it is. Each is easy to break by accident.

getUsed and spill must stay lock-free. Spark calls both while already holding the TaskMemoryManager monitor, which is why NativeMemoryConsumer.getUsed reads an AtomicLong and spill returns without touching guarded state. A consumer that took a monitor of its own in either method would risk a deadlock: a native reservation arriving over JNI on a Comet Tokio worker holds Spark’s monitor and would wait for the consumer’s, while whatever held the consumer’s waited for Spark’s. For the same reason, nothing reachable from a spill callback may allocate memory that routes back through the same consumer.

scala.util.control.NonFatal does not contain an acquisition. acquireExecutionMemory fails in three ways and NonFatal catches only the first. It runs other consumers’ spill, where TaskMemoryManager turns an interrupted spill into a RuntimeException and an IOException into a SparkOutOfMemoryError, which extends OutOfMemoryError and is therefore an Error. Separately, ExecutionMemoryPool.acquireMemory parks in lock.wait() when a task is below its fair share, so killing a task raises a plain InterruptedException, which NonFatal excludes by name. Comet reaches this method from native over JNI, so whatever escapes crosses the JNI boundary.

Spark exposes no per-consumer usage figure. TaskMemoryManager.getMemoryConsumptionForThisTask is task-wide, and a consumer that reaches acquireExecutionMemory directly rather than through MemoryConsumer.acquireMemory keeps an inherited used of zero. That is why NativeMemoryConsumer overrides getUsed to report Comet’s own tally instead: without it, Spark’s spill-victim ordering and showMemoryUsage would believe the consumer held nothing.

A partial grant can be stranded. acquireExecutionMemory takes its first grant from the pool before asking other consumers to spill, so when a spill throws, the task has been charged for bytes the call never returns. Nothing releases them until Spark’s final task cleanup, so they are headroom nobody can use for the rest of the task. Any caller that swallows the exception has to reconcile that grant, and the only figure available for doing so is the task-wide one above.

A parked acquire can wake up to a missing task entry. Spark removes the task’s memoryForTask entry when its balance reaches zero, and an acquire that was waiting in lock.wait() and wakes afterwards throws a NoSuchElementException (“key not found” and the task id) instead of a grant (SPARK-59444). Releases take no task monitor, so another native thread or plan, or a JVM consumer of the same task, can empty the balance while an acquire waits. Both of Comet’s consumers that ask Spark directly, CometTaskMemoryManager and the JVM shuffle allocator, retry that one case through MissingTaskEntryRetry. A missing entry means the task held nothing from Spark at that moment, so the failed call has no partial grant to reconcile, and the retry registers the task again and waits for its share as the first call would have. After three attempts CometTaskMemoryManager returns a zero grant. try_grow treats that as a refusal and the operator spills, while grow records the request as overcommit. The shuffle allocator throws SparkOutOfMemoryError instead. Both refuse without retrying once the task has completed or been killed: releasing a native plan at the end of the task can empty the balance under a Tokio worker still waiting in Spark, and a retry would register the ended task again and could wait for other tasks while holding the TaskMemoryManager monitor that Spark’s final cleanup needs. Spark’s own operators in the same task, such as its sorters and aggregates, are not guarded and can still hit the exception until Spark re-registers a waiting task in ExecutionMemoryPool (apache/spark#58747).

A consumer whose spill returns zero takes budget it can never give back. NativeMemoryConsumer.spill returns 0, so Spark can select it as a spill victim and reclaim nothing from it; it is only ever a spill trigger. The bytes it holds are real, so other consumers in the same task see correspondingly less headroom and can spill earlier than they otherwise would.

Where Comet’s budget comes from

CometExecIterator.getMemoryConfig computes the budget once per executor and passes it across JNI to Java_org_apache_comet_Native_createPlan as memory_limit, along with spark.memory.offHeap.size itself as off_heap_size. Comet shares Spark’s off-heap pool rather than asking for a separate allocation:

memory_limit = spark.memory.offHeap.size * spark.comet.exec.memoryPool.fraction

spark.comet.exec.memoryPool.fraction defaults to 1.0 and is deprecated. It was documented as the workaround for Comet’s under-accounting (see The accounting gap), holding back a slice of the off-heap pool for the memory Comet does not reserve, but it cannot do that:

• greedy_unified ignores memory_limit. It asks Spark for every byte it reserves.

• fair_unified caps each task at it (see The unified pools). Spark’s execution pool already limits each of N running tasks to spark.memory.offHeap.size / N, which is the tighter cap unless the fraction is below 1 / N, and the tasks together can still acquire the whole pool.

• Spark’s own off-heap consumers, non-Comet operators and off-heap storage, draw on the same pool with no Comet limit at all.

The only room Spark leaves for memory outside the pool is spark.executor.memoryOverhead.

Resolving the pool type

parse_memory_pool_config (native/core/src/execution/memory_pools/config.rs) turns the pool-type string and the limit into a MemoryPoolConfig. Two pool types are valid in off-heap mode:

Pool type	Sized from	Notes
fair_unified (default)	memory_limit and off_heap_size	Delegates to Spark’s TaskMemoryManager; task-shared
greedy_unified	n/a (pool size 0)	Spark owns the limit entirely; task-shared

Any other pool type is rejected with a configuration error. In on-heap mode the pool-type string is ignored and the pool is always UnboundedMemoryPool.

The pool stack

create_memory_pool builds a base pool and createPlan then wraps it in decorators. Reading from the inside out, a Comet plan in the default configuration sees:

[LoggingMemoryPool]          <- only when spark.comet.debug.memory=true
  [TaskSharedMemoryPool]     <- RAII handle for the per-task registry
    [SpillReplayPool]        <- lets a spilled final aggregate read its spill files back
      [TrackConsumersPool]   <- DataFusion; names the top 10 consumers in error messages
        [CometFairMemoryPool]  <- delegates acquire/release to Spark over JNI

Each decorator forwards every MemoryPool method to its inner pool, so reserved() at any level reports the base pool’s number. SpillReplayPool also turns one kind of refused try_grow into a grow; see Final aggregates reading their spill files back.

The unified pools

CometUnifiedMemoryPool and CometFairMemoryPool (unified_pool.rs, fair_pool.rs) are the bridge to Spark. Their try_grow and grow both call CometTaskMemoryManager.acquireMemory over JNI, which goes through Spark’s ordinary TaskMemoryManager. That means:

• 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 in the same task to spill to satisfy Comet’s request.

• The reverse does not hold. Comet registers a NativeMemoryConsumer with the TaskMemoryManager so that Spark has something to charge, but its spill() always returns 0 (CometTaskMemoryManager.java). A Spark allocation can therefore never make a native sorter or aggregate release its reservations. Native operators spill only when their own try_grow fails, so a JVM consumer that is blocked behind native reservations has no way to reclaim them.

• In try_grow, a partial grant (acquired < additional) is released immediately and reported as ResourcesExhausted, which is the signal DataFusion uses to spill.

• grow cannot fail, so when Spark grants less than asked the pool records the full amount anyway and carries the shortfall as overcommit (spark_memory.rs). A later shrink repays the overcommit before returning anything to Spark, so Spark is never handed back more than it granted. While overcommit is outstanding, try_grow asks Spark for it on top of the request and is refused unless Spark can cover both, so operators spill until the debt is repaid. Both pools’ Display output and their try_grow errors report the current overcommit.

CometFairMemoryPool exists because Spark cannot ask a native operator to spill. Spark limits each task to an even share of its off-heap pool among the tasks that hold memory from it. Once one operator holds its task’s share, Spark refuses the task’s other operators, and one that cannot spill, such as a sort’s merge or the build side of a hash join, fails the task. The pool keeps memory for those operators by limiting the ones that can spill to a share of the task’s, as DataFusion’s FairSpillPool does over its whole pool (issue #7088).

The task’s share is off_heap_size divided by the number of tasks on the executor that are running native plans, that is, that hold a task-shared pool of either type, or pool_size if that is smaller, which it is only while the deprecated fraction is below 1 / N. A consumer that can spill is limited to an even share of what the consumers that cannot spill leave of that:

task_share = min(off_heap_size / running_tasks, pool_size)
limit      = (task_share - unspillable) / num_spill

Here unspillable is the bytes held by consumers that cannot spill, and num_spill is the number of registered consumers that can.

Like Spark’s own divisor, the number of running tasks follows tasks as they start and finish, so a task that runs alone has the whole pool. It is still not Spark’s number or Spark’s pool, and the differences go in these directions:

• Spark counts a task from its first acquire until its balance returns to zero, whether or not it runs a native plan. Tasks that run no native plan but hold off-heap memory, such as the JVM operators of a mixed stage, are in Spark’s count and not Comet’s. Comet then divides by fewer tasks than Spark does, the task’s share is more than Spark will grant it, and the shares of its consumers that can spill can add up to more than Spark grants. That is issue #7088 again, so look here first when a stage that mixes JVM and native operators still fails a sort’s merge.

• A task that holds a native pool but no memory, such as a native scan that feeds a JVM shuffle, is in Comet’s count and not Spark’s. Comet then divides by more tasks than Spark does, and its consumers that can spill spill earlier than Spark requires. Look here when a mixed workload spills more under fair_unified than under greedy_unified.

• Spark divides the pool minus the off-heap storage in use, up to the storage region, which is spark.memory.storageFraction of the pool. With StorageLevel.OFF_HEAP caching, Spark’s limit on a task can be about half of off_heap_size / running_tasks at the default fraction, while task_share divides all of off_heap_size, so the shares can add up to more than Spark grants.

Before it asks Spark for a consumer that can spill, the pool makes two checks, and refuses the request without calling Spark if either fails:

• The consumer against its share. The share is what the consumers that cannot spill leave of the task’s share, divided by the number of registered consumers that can spill. What the consumer already holds plus the request must not exceed it. That counts all of the consumer’s reservations, including the sibling reservations that new_empty(), split() and take() create, so an operator that holds several reservations still gets one share. A sort’s streaming merge, for example, creates one for each batch it reads. The pool keeps a running total for each consumer, because reservation.size() covers only one reservation. With eight running tasks, a 16 GiB pool and two registered consumers that can spill, the task’s share is 2 GiB and each consumer’s is 1 GiB, or 768 MiB while a merge holds 512 MiB.

• The task’s total against the task’s share. The shares alone do not bound the total, because a consumer keeps what it reserved while fewer consumers were registered or fewer tasks were running.

A consumer that cannot spill takes what Spark grants. The pool refuses it only when the task’s total plus the request would exceed pool_size, which Spark’s own limit on the task is at least as tight as unless the deprecated fraction is below 1 / N.

The pool goes by what a consumer declares, with one exception. DataFusion 55.2 registers OrderedFinalAggregateStream as able to spill even when its input is sorted on every grouping key, as a workaround for apache/datafusion#17334, but builds a spill path only for partially sorted input. Fully sorted, as in a CometSortAggregateExec over a native sort, a refusal fails the query, so the pool checks its requests only against the pool, as it does those of a consumer that cannot spill (limited_to_its_share in fair_pool.rs). It still counts as a consumer that can spill when the pool divides the task’s share, so a sort below it stays at half of the share and leaves the aggregate the other half. The partially sorted stream still spills when Spark refuses it.

Three details matter for tuning:

• The pool is task-shared (see below), so the consumers that divide the task’s share include those of every native plan in the task, not just the plan making the request. Registering a consumer that can spill lowers the share of every other one, but does not take back memory they already hold. Registering one that cannot spill changes no share until it holds memory.

• A consumer at its share is refused, and spills, even when the rest of the task’s share is free. That keeps the memory for the task’s other consumers, and is why fair_unified can spill earlier than greedy_unified.

• A consumer’s limit can be well below what Spark would grant the task, and some operators fail where a consumer that can spill is refused. In DataFusion 55.2 an ExternalSorter returns “Not enough memory to continue external sort” instead of spilling when it cannot reserve a batch while it holds nothing, when it cannot reserve it again after spilling, and when it cannot resize its reservation for a sorted batch. A sort can therefore fail when one batch is larger than its share, as can happen with wide rows, many cores, little off-heap memory and several consumers that can spill in the task. It can also fail on its first request when consumers that cannot spill, such as the build side of a shuffled hash join or a sort-merge join’s buffered rows, hold the task’s whole share, because the limit is then zero. Spark lets such a consumer hold more than off_heap_size / running_tasks while other tasks have not yet asked for memory. A larger spark.memory.offHeap.size, a smaller spark.comet.batchSize or spark.comet.exec.memoryPool=greedy_unified avoids it, and the migration guide says so too.

Task-shared pools and their lifetime

A single Spark task can run more than one native plan at a time. A native shuffle is not one of these cases, because its writer is planned together with the native operators that feed it. These operators do split a task’s native work into separate plans:

• CometUnionExec and CometCoalesceExec read their children through the JVM, so the native plan above them and the native plans below them are separate.

• CometCollectLimitExec and CometTakeOrderedAndProjectExec apply their limit in a native plan of their own, over the output of the plan below them.

• A native Parquet or Iceberg write runs its writer as a native plan of its own, over the output of the plan below it.

If each plan got its own pool, the per-task limit would be enforced once per plan rather than once per task.

acquire_task_shared_pool (task_shared.rs) keeps a process-wide HashMap<task_attempt_id, Weak<TaskSharedMemoryPool>>. Plans in the same task upgrade the existing Weak and share one pool; the returned Arc is the only lifetime handle, so the registry entry disappears when the last plan (and its last reservation) drops. There is no explicit release call to forget, and a createPlan that fails partway through cleans up on unwind.

TaskSharedMemoryPool::drop has to handle one race: an acquire can observe an expired Weak and insert a replacement before the dying pool reaches the registry lock. The drop therefore compares pointers and only removes an entry that is still its own.

The pool acquires memory from Spark through the CometTaskMemoryManager passed with the plan that created it, so the JVM side shares one manager per task as well: CometExecIterator.taskMemory hands every native plan in a task the same one and drops it when the task completes. Its getUsed covers the whole task, so CometExecIterator.close() warns about memory still in use only when the task’s last open native plan closes.

Final aggregates reading their spill files back

SpillReplayPool (spill_replay.rs) wraps both Comet pools to work around issue #6254. Once one of DataFusion 55’s final aggregates has spilled, it reads its spill files back through an aggregate that cannot spill, so a refused try_grow there fails the task. When the pool refuses such a request, SpillReplayPool records it with the pool’s grow instead, which skips CometFairMemoryPool’s local checks and carries what Spark does not grant as overcommit. Every other refusal is passed on unchanged. spill_replay.rs describes how the wrapper recognizes these requests and what that relies on in DataFusion. Issue #6583 tracks removing it.

How DataFusion consumes the pool

Native operators reserve through DataFusion’s MemoryConsumer / MemoryReservation API:

• try_grow(n) may fail. Spillable operators (ExternalSorter, the grouped hash aggregate, sort-merge join) respond to a ResourcesExhausted error by spilling to disk and retrying. This is the only mechanism that turns memory pressure into progress rather than failure.

• grow(n) is infallible. It is used for memory that already exists and the caller cannot spill, such as a spilled batch read back from disk. The Comet pools record it even when Spark grants less, as overcommit (see The unified pools).

• shrink(n) returns bytes to the pool.

An operator that never calls try_grow is invisible to the pool no matter how much memory it uses.

Native writers

Both native writers reserve what they hold between batches through a single consumer per task, ParquetWriterExec[N] or IcebergWriteExec[N], resized after every batch. Neither can spill. A fanout Iceberg write gives memory back instead, by closing partitions early when the pool refuses a resize, as described below, so a refusal never fails it. An exception Spark throws during a resize, such as a failed spill or a killed task, still fails the task. When the pool refuses any other write’s resize, the task fails with a CometNativeException whose message starts Additional allocation failed for and names the consumer. That is a task failure Spark can retry. Unreserved, the same memory would count only toward the container limit, where exceeding it kills the executor.

• ParquetWriterExec writes one file per task and reserves parquet-rs’s estimate of the file’s in-progress row group (ArrowWriter::memory_size). The estimate counts encoded pages, encoder buffers, dictionaries and Bloom filters. The writer keeps parquet-rs’s default row-group limit of 1Mi rows and sets no byte limit, so a wide schema can hold a large row group (#5304). A file that is neither local nor on HDFS, such as one the S3A magic committer hands out, is written through a Hadoop output stream, and the reservation then also includes the 1 MiB buffer that batches the writes to it. The stream’s own buffers, such as S3A’s upload buffers (fs.s3a.fast.upload.buffer), are not Comet’s and sit outside its pool, as they do for Spark’s writer.

• The Iceberg writer keeps one file open per partition in a fanout write, so its reservation grows with the number of partitions a task writes. iceberg-rust’s ParquetWriter does not expose parquet-rs’s memory_size, only every byte the file has written, its in-progress row group’s encoded size included. So each file’s writer counts the bytes that leave memory on their way to storage, and the file reports what it has written less those. A local file writes a flushed row group out at once, so it reports its in-progress row group and the few KiB parquet-rs buffers in front of storage. S3 and GCS take a file in parts of at least 5 MiB, and OpenDAL holds each part in memory until the next one is complete or the file closes, so there a file also reports what it has flushed since its last part was uploaded. At the default row-group size (write.parquet.row-group-size-bytes, 128 MiB) that is the last row group it flushed. The reservation also covers the rows each partition holds back, first for its dictionary choice and then until they fill the 1000-row unit the rolling writer is fed in. When the pool refuses a fanout write’s resize, the writer writes out and closes the partitions holding the most, file and held rows together, until the resize succeeds, so the write ends with more files rather than failing. Each file also reports to its partition’s total, which is how the writer finds them. The reservation does not cover what parquet-rs holds beyond the encoded size: dictionary hash tables, unencoded dictionary indices and buffer capacity. Native writes decline Bloom filters today, and neither figure would include them.

The writers register one consumer per task rather than one per open file. A fanout write registers it as a consumer that can spill, because the writer answers a refusal by closing partitions, so fair_unified holds it to a share and it ends with more files rather than take memory from the task’s sorts. The unpartitioned and clustered writers keep their one file open, so a refusal fails the task. They cannot spill, fair_unified does not hold them to a share, and what they hold comes out of the share that the task’s operators that can spill divide.

Crossing the FFI boundary

Batches move between the JVM and native over the Arrow C Data and C Stream interfaces, which are zero-copy. Nothing is copied, so the allocator that produced a batch and the runtime that decides when it dies can be on opposite sides of the boundary. See Arrow FFI for the mechanics; what matters here is who is charged and who controls the lifetime.

Two things are easy to conflate here. The allocator of a buffer determines which process-level accounting (if any) saw the bytes appear. Whether the buffer is reserved in Comet’s pool is a separate decision, made by whichever operator holds it, and that operator neither knows nor cares which side of the boundary the bytes came from.

JVM → native (ScanExec). The JVM allocates the Arrow buffers from a child of CometArrowAllocator and exports the whole per-partition iterator once as an ArrowArrayStream. ScanExec imports each batch through arrow-rs’s ArrowArrayStreamReader and keeps every column as an Arc clone of the imported buffers; the JVM decodes dictionaries before export, so there is nothing to unpack. Those bytes stay where Java Arrow put them and are pinned for as long as any native reference survives. They are invisible to Spark’s TaskMemoryManager, and CometArrowAllocator is unbounded, so nobody charged for them at allocation time. Whether they are charged later depends on who holds them. DataFusion’s ExternalSorter reserves get_reserved_bytes_for_record_batch for every batch it retains, and the hash join build side reserves get_record_batch_memory_size for each incoming batch; both read buffer sizes off the ArrayData and apply equally to imported buffers. So an imported batch that a sort or join has buffered is reserved in Comet’s pool, and through a unified pool is charged to Spark. The same batch held by an operator that does not reserve (a projection or filter, or ScanExec itself between polls) is charged to nothing.

Native → JVM (CometExecIterator). DataFusion produces the batch in Rust, and whether it is still reserved when it reaches the boundary depends on the producing operator. The sort’s output is wrapped in a ReservationStream that shrinks the reservation by the batch’s size as each batch is emitted, so by the time prepare_output sees a sorted batch its bytes are no longer reserved. prepare_output then calls move_to_spark, which writes an FFI_ArrowArray whose release callback drops the Rust ArrayData; no reservation travels with it. The JVM wraps the pointers in ArrowBufs and the memory is freed when the JVM calls close(). An exported batch is therefore usually not pool-charged while the JVM holds it, but it is resident until the JVM releases it. A slow or backed-up JVM consumer keeps native memory alive that no accounting layer is counting.

The point is not that one direction is charged and the other is not. Buffer lifetime and reservation lifetime are independent. A reservation is a number an operator chose to declare and later withdraw; a buffer lives until its last reference drops, on whichever side of the boundary that happens. The two are kept in step only inside operators written to do so, and only while that operator holds the batch. Once the batch leaves the operator, in either direction, the reservation stops and the bytes keep going. That is the shape of the gap to hold in mind when reading the next section, and it is what any fix at the allocator level has to close.

The accounting gap

The pool tracks declared reservations. Container RSS counts pages the process touched. The two diverge for several structural reasons:

• Undeclared allocations. Arrow array builders, expression kernels producing intermediate arrays, decompression buffers, Parquet metadata structures, the part of a native writer’s buffers its estimate misses (see Native writers), object_store request buffers, and tokio’s own machinery all allocate without reserving. Only operators that were explicitly written to reserve show up in the pool.

• Rounding and padding. Arrow buffers are padded to 64-byte boundaries and builders grow by doubling, so a reservation of exactly n bytes routinely corresponds to more than n bytes of heap.

• Allocator behavior. malloc-level fragmentation, size-class rounding, and jemalloc’s retained/dirty page cache all add resident bytes that no layer above the allocator can see. Freeing memory does not necessarily return pages to the OS.

• Non-Rust allocations. Memory allocated by C dependencies through libc malloc, and anything mmaped, never passes through Rust’s GlobalAlloc, so neither the memory pool nor the allocation counters (native_allocated, jemalloc_allocated) see it. In a default build the C dependencies are libzstd (zstd-sys, behind the Parquet zstd codec), libhdfs (hdfs-sys, pulled in by the default hdfs-opendal feature), and the TLS stack used for cloud object stores (aws-lc-sys). With the python-udf feature, the embedded Python interpreter and PyArrow also allocate outside GlobalAlloc; their memory is absent from the executor’s native allocated figure, and Python working allocations are not reserved in Comet’s pool. Building with the jemalloc or mimalloc feature adds the allocator itself (tikv-jemalloc-sys, libmimalloc-sys). It is worth knowing which dependencies are not C, because several names suggest otherwise: the other Parquet codecs are pure Rust in this build, snap for Snappy, lz4_flex for LZ4 and zlib-rs for gzip, as is libbz2-rs-sys despite its name, so those allocations do pass through GlobalAlloc and are counted.

• Batches in flight across the FFI boundary. Reservations stop at the operator that made them. Imported JVM batches are reserved only while a reserving operator holds them, and exported native batches have usually been released by the time the JVM receives them yet stay resident until the JVM closes them (see Crossing the FFI boundary).

• Overcommit. Unlike everything above, this is not accidental. When Spark grants less than a grow asked for, Comet holds the difference anyway and deliberately does not tell Spark about it (see The unified pools). reserved() includes it, but Spark’s memory manager does not, so until it is repaid Spark can hand the same bytes to another consumer or task. The overcommit figure in the pool’s Display output and try_grow errors shows how much is outstanding. The executor’s memory usage log leaves it out of the reserved figure it reports, so that the log counts it as untracked: like an undeclared allocation, it needs room beyond what Spark has handed out. Tracing’s comet_memory_reserved_total still includes it.

The practical consequence is that reserved() is a lower bound on Comet’s real footprint, except while tasks share a decoded broadcast, which they reserve once each, and the gap is workload-dependent. The margin that covers it has to come from spark.executor.memoryOverhead. The deprecated spark.comet.exec.memoryPool.fraction cannot provide one; see Where Comet’s budget comes from.

To measure the gap on a real workload, read the executor’s periodic memory usage log, which reports the bytes Rust’s allocator has handed out next to the pools’ reservations, and the Arrow memory Comet holds on the JVM side; see Sizing the Overhead from the Memory Usage Log. For a view per event rather than per interval, enable tracing and compare native_allocated against comet_memory_reserved_total; see Tracing.

What the container sees

On Kubernetes, Spark sizes the executor pod from ResourceProfile:

pod memory request = pod memory limit
                   = spark.executor.memory
                   + spark.executor.memoryOverhead   (default max(factor * executor.memory, 384 MiB))
                   + spark.memory.offHeap.size
                   + spark.executor.pyspark.memory   (Python applications in cluster mode only)

The factor is spark.executor.memoryOverheadFactor or, when that is unset, spark.kubernetes.memoryOverheadFactor. Both default to 0.1, but in cluster mode spark-submit sets the Kubernetes factor to 0.4 for a PySpark or SparkR application that did not set it.

Both the request and the limit are set to this same value, so the pod’s cgroup memory.max is a hard ceiling on the sum of everything in the container. That cgroup counts, among other things:

• the JVM heap (spark.executor.memory),

• JVM non-heap: metaspace, code cache, thread stacks, GC structures, Netty direct buffers,

• Spark’s own off-heap allocations,

• all of Comet’s native allocations,

• Comet’s JVM-side Arrow buffers (CometArrowAllocator),

• page cache charged to the cgroup by the container’s file I/O, including spill files.

Everything the cgroup counts, and who accounts for each part:

Spark’s accounting covers the first group, though not in the same sense throughout it. The JVM heap, Tungsten pages and Comet’s shuffle pages are allocated by JVM code that reports what it allocated. A native reservation is a number an operator declared before allocating: try_grow succeeds only once CometTaskMemoryManager has charged Spark’s off-heap execution pool over JNI, so the budget really is spent, but nothing measured the bytes and the reservation is only a lower bound on them. The second group is outside every accounting layer.

When the total crosses memory.max, the kernel OOM killer kills the process. The failure mode is significantly worse than a task-level OOM: every task running on that executor dies, every cached block it held is lost and must be recomputed, and the shuffle files it produced become unavailable to downstream fetches. Spark’s driver sees only ExecutorLostFailure with exit code 137.

Two facts follow that are easy to get wrong:

1. spark.memory.offHeap.size is part of the pod limit, not extra headroom on top of it. Raising Comet’s off-heap budget on Kubernetes raises the pod’s memory request by the same amount, so the scheduler will place fewer executors per node rather than silently giving Comet more room.

2. spark.executor.memoryOverhead is the only slack in the container, and the JVM’s own non-heap usage already consumes a large part of it. Comet’s overshoot beyond its declared reservations eats into the same allowance.

YARN behaves analogously. The container size is the same sum, and the NodeManager kills containers that exceed it, but the kill is done by the NodeManager’s monitor rather than the kernel, so it is somewhat less abrupt.

Open problems

Nothing described above prevents an executor from being OOM-killed. Every enforcement point Comet has bounds declared reservations, and the sections above describe several structural reasons why declared reservations are a lower bound on physical usage. The known gaps, roughly in order of how much they matter:

• Real native usage is observed but not acted on. Comet counts the bytes Rust’s allocator has handed out, and each executor logs that count next to the pools’ reservations. Nothing reads it at runtime, though: no operator, metric, or policy responds to it, so an executor that outgrows its container is still stopped only by the kill.

• The memory overhead is sized by hand. The gap has to fit in spark.executor.memoryOverhead, and the memory usage log measures it, but nothing sizes the overhead from it.

• CometArrowAllocator is unbounded and participates in no budget. The memory usage log reports it, but nothing bounds it.

• Buffer and reservation lifetimes are independent across the FFI boundary. A batch can be resident on either side with no reservation covering it, because reservations are made and withdrawn by individual operators while the bytes outlive them.

• Spark cannot trigger native spilling. NativeMemoryConsumer.spill() returns 0, so native reservations are released only when a native operator decides to spill on its own failed try_grow. JVM consumers in the same task can be starved behind them.

Issue #4576 tracks work on the first two.

Debugging memory issues

Tool	What it gives you
spark.comet.debug.memory=true	LoggingMemoryPool logs every register/grow/shrink with the consumer name
spark.comet.explain.native.enabled=true	Native plan with per-operator metrics, including spill counts
Memory usage log	Executor-wide native allocation vs pool reservations, and JVM Arrow memory, logged every 10 seconds by default
Tracing	native_allocated vs comet_memory_reserved_total per event; the accounting gap over time
TrackConsumersPool	Names the top 10 consumers in ResourcesExhausted messages (always on)
thresher	Third-party crate that dumps a jemalloc heap profile at a threshold

A checklist for triaging an executor OOM kill:

1. Confirm which budget was exceeded. Exit code 137 / OOMKilled on the pod is the cgroup. Exit code 52 with java.lang.OutOfMemoryError in the executor log is JVM heap exhaustion; Spark treats it as fatal, so the executor is lost either way and the exit code is what distinguishes them. A failed task with SparkOutOfMemoryError and a surviving executor is Spark’s managed memory pool, which is the only one of the three that is recoverable at task level.

2. Compare allocated against reserved in the executor’s Comet native memory usage log lines leading up to the kill, or nativeAllocated against poolsReserved in its CometExecutorMemoryUsage events when the event log records them, or native_allocated against comet_memory_reserved_total in a trace. A large excess points at undeclared native allocations; a small excess points at the budget simply being too small, or at the JVM side, which the same lines report as JVM Arrow allocated.

3. Check spark.comet.batchSize against the schema width. Peak memory scales with batch_size * columns, and wide or deeply nested schemas amplify it.

4. Check whether the operators involved can spill at all. ShuffledHashJoin cannot, so spark.comet.exec.forceShuffledHashJoin=true converts a spillable sort-merge join into one that is not, for build sides under spark.comet.exec.forceShuffledHashJoin.maxBuildSize.