feat: always count native allocations and log executor native memory usage

.github/actions/rust-test/action.yaml

@@ -84,16 +84,13 @@ runs:
         export LD_LIBRARY_PATH=${JAVA_HOME}/lib/server:${LD_LIBRARY_PATH}
         RUST_BACKTRACE=1 cargo nextest run
 
-    # The `alloc-accounting` feature is off by default, so nothing else in CI compiles the
-    # allocator wrapper, its backend selection, or the benchmark's liveness guards. Lint them with
-    # the accounting wrapper over jemalloc, run the accounting tests with the wrapper installed, and
-    # check the system-allocator arm.
-    - name: Check and test the alloc-accounting feature
+    # The steps above lint and test the accounting allocator over the system allocator. Lint and
+    # test it over jemalloc too, since that is a different allocator backend and nothing else in CI
+    # builds it.
+    - name: Check and test the jemalloc allocator backend
       shell: bash
       run: |
         cd native
         export LD_LIBRARY_PATH=${JAVA_HOME}/lib/server:${LD_LIBRARY_PATH}
-        cargo clippy --color=never -p datafusion-comet --all-targets --features jemalloc,alloc-accounting -- -D warnings
-        RUST_BACKTRACE=1 cargo nextest run -p datafusion-comet --lib --features jemalloc,alloc-accounting alloc_accounting
-        cargo check -p datafusion-comet --features alloc-accounting
-
+        cargo clippy --color=never -p datafusion-comet --all-targets --features jemalloc -- -D warnings
+        RUST_BACKTRACE=1 cargo nextest run -p datafusion-comet --lib --features jemalloc alloc_accounting

docs/source/contributor-guide/memory_management.md

@@ -355,14 +355,14 @@ diverge for several structural reasons:
   Freeing memory does not necessarily return pages to the OS.
 - **Non-Rust allocations.** Memory allocated by C dependencies through libc `malloc`, and anything
   `mmap`ed, never passes through Rust's `GlobalAlloc`, so neither the memory pool nor the
-  `jemalloc_allocated` metric sees 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`). 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.
+  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`). 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
@@ -372,9 +372,13 @@ The practical consequence is that `reserved()` is a lower bound on Comet's real
 gap is workload-dependent. `spark.comet.exec.memoryPool.fraction` exists purely so operators can
 hand-tune a margin that covers the gap for their workload.
 
-To measure the gap on a real query, enable tracing with the `jemalloc` feature and compare
-`jemalloc_allocated` against the summed `thread_NNN_comet_memory_reserved` values; see
-[Tracing](tracing.md#analyzing-memory-usage).
+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; see
+[Sizing the Overhead from the Memory Usage Log][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](tracing.md#analyzing-memory-usage).
+
+[memory-usage-log]: ../user-guide/latest/tuning.md#sizing-the-overhead-from-the-memory-usage-log
 
 ## What the container sees
 
@@ -452,9 +456,9 @@ has bounds _declared reservations_, and the sections above describe several stru
 declared reservations are a lower bound on physical usage. The known gaps, roughly in order of how
 much they matter:
 
-- **No signal for real native usage.** The only way to observe the gap today is to enable tracing
-  with the `jemalloc` feature and compare `jemalloc_allocated` against summed reservations after
-  the fact. There is no runtime value that an operator, a metric, or a policy could read.
+- **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.
 - **`spark.comet.exec.memoryPool.fraction` is a manual proxy for the gap.** It asks operators to
   guess a per-workload margin rather than measuring anything.
 - **`CometArrowAllocator` is unbounded** and participates in no budget.
@@ -469,13 +473,14 @@ much they matter:
 
 ## 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              |
-| [Tracing](tracing.md#analyzing-memory-usage)     | `jemalloc_allocated` vs summed pool reservations; the accounting gap       |
-| `TrackConsumersPool`                             | Names the top 10 consumers in `ResourcesExhausted` messages (always on)    |
-| [`thresher`](https://github.com/cetra3/thresher) | Third-party crate that dumps a jemalloc heap profile at a threshold        |
+| 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](../user-guide/latest/tuning.md#sizing-the-overhead-from-the-memory-usage-log) | Executor-wide native allocation vs pool reservations, logged every 10 seconds by default    |
+| [Tracing](tracing.md#analyzing-memory-usage)                                                     | `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`](https://github.com/cetra3/thresher)                                                 | Third-party crate that dumps a jemalloc heap profile at a threshold                         |
 
 A checklist for triaging an executor OOM kill:
 
@@ -484,9 +489,10 @@ A checklist for triaging an executor OOM kill:
    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 `jemalloc_allocated` against the summed pool reservations from 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.
+2. Compare `allocated` against `reserved` in the executor's `Comet native memory usage` log lines
+   leading up to the kill, 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.
 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

docs/source/contributor-guide/tracing.md

@@ -27,20 +27,17 @@ directory with the filename `comet-event-trace.json`.
 
 [Trace Event Format]: https://docs.google.com/document/d/1CvAClvFfyA5R-PhYUmn5OOQtYMH4h6I0nSsKchNAySU/preview?tab=t.0#heading=h.yr4qxyxotyw
 
-Additionally, enabling the `jemalloc` feature will enable tracing of native memory allocations.
+Native memory is traced as `native_allocated`. Comet wraps whichever global allocator the build
+selected and counts the bytes it has handed out. It counts only what Rust code allocated, so it can be compared against the
+memory pool's reservations without the allocator's own caching in the way. The same figure appears in
+the executor's periodic memory usage log, which does not need tracing; see
+[Sizing the Overhead from the Memory Usage Log](../user-guide/latest/tuning.md#sizing-the-overhead-from-the-memory-usage-log).
 
-```shell
-make release COMET_FEATURES="jemalloc"
-```
-
-The `alloc-accounting` feature adds a second, allocator-independent measure of native memory. It
-wraps whichever global allocator the build selected and reports the bytes it has handed out as
-`native_allocated`. Unlike `jemalloc_allocated` it does not require jemalloc, and it counts only
-what Rust code allocated, so it can be compared against the memory pool's reservations without the
-allocator's own caching in the way. The two features are independent and can be combined:
+Enabling the `jemalloc` feature adds a second measure, `jemalloc_allocated`, which also includes
+jemalloc's own metadata and fragmentation:
 
 ```shell
-make release COMET_FEATURES="jemalloc,alloc-accounting"
+make release COMET_FEATURES="jemalloc"
 ```
 
 Example output:
@@ -165,7 +162,7 @@ samples: they are not an atomic per-query balance, and neither is a measure of R
 | jemalloc_allocated               | Native memory usage for the executor process (requires `jemalloc` feature)                                                                                                                                                                                     |
 | jvm_arrow_allocated              | Bytes charged to Comet's Arrow allocator tree on the JVM, including buffers imported from native over the Arrow C Data Interface                                                                                                                               |
 | jvm_arrow_imported               | Bytes charged to the Arrow C Data Interface import allocator, a subset of `jvm_arrow_allocated`. An allocator charge, not a measure of where the bytes were allocated; see above.                                                                              |
-| native_allocated                 | Bytes handed out by the Rust global allocator, process-wide (requires `alloc-accounting` feature). Approximate to within 64 KiB of un-flushed delta per live thread.                                                                                           |
+| native_allocated                 | Bytes handed out by the Rust global allocator, process-wide. Approximate to within 64 KiB of un-flushed delta per live thread.                                                                                                                                 |
 | comet_memory_reserved_total      | Total memory reserved across every live Comet memory pool, process-wide, whatever the configured pool type. Counts a pool shared between execution contexts once, so unlike the per-thread counters it can be compared directly against an allocation counter. |
 | thread_NNN_comet_memory_reserved | Memory reserved by Comet's DataFusion memory pool (summed across all contexts on the thread). NNN is the Rust thread ID. Do not sum these across threads: a shared pool reports its full reservation on every thread that references it.                       |
 | thread_NNN_comet_jvm_shuffle     | Off-heap memory allocated by Comet for columnar shuffle. NNN is the Rust thread ID.                                                                                                                                                                            |

docs/source/user-guide/latest/tuning.md

@@ -108,6 +108,7 @@ there is. That includes:
 
 Reserved memory is therefore a lower bound on what Comet really uses, and how far below it sits depends on the
 workload. This is why Comet can stay within the pool's limit and still push the executor past its container limit.
+Each executor logs how far apart the two are while Comet runs; see [Sizing the Overhead from the Memory Usage Log].
 To leave room for the part that is not counted, set `spark.comet.exec.memoryPool.fraction` to a value less than
 `1.0`, which restricts the amount of memory Comet is allowed to reserve.
 
@@ -180,10 +181,65 @@ spark.executor.memoryOverheadFactor=0.2
 
 Raise the value further if executors are killed by the cluster manager (on Kubernetes,
 `ExecutorLostFailure` with exit code 137) rather than failing with a task-level out-of-memory error.
+To measure how much Comet needs rather than guessing, see [Sizing the Overhead from the Memory Usage Log].
 
 Note that on Kubernetes and YARN the overhead is added to the container size, so raising it reduces
 how many executors fit on a node.
 
+[Sizing the Overhead from the Memory Usage Log]: #sizing-the-overhead-from-the-memory-usage-log
+
+### Sizing the Overhead from the Memory Usage Log
+
+While Comet native plans are running, each executor logs its native memory usage at INFO level,
+one line every 10 seconds for the whole executor:
+
+```
+Comet native memory usage: allocated 5412.3 MiB, reserved 3890.0 MiB (16 native plans, 8 memory pools)
+```
+
+- `allocated` is the memory that Comet's native code has allocated and not yet freed, whether or not
+  a pool tracks it.
+- `reserved` is the part that Comet's memory pools track. It is charged against
+  `spark.memory.offHeap.size`, so the container already has room for it.
+
+The difference between the two, `allocated - reserved`, is Comet's untracked native memory. It is
+the part of Comet's footprint that has to fit in `spark.executor.memoryOverhead`, alongside the
+JVM's own non-heap memory. To size the overhead from it:
+
+1. Run a representative workload and find the line with the largest difference in each executor's
+   log. Take both figures from the same line: they are sampled together, and figures from different
+   lines describe different moments. Setting `spark.comet.memory.logInterval=1s` for this run makes a
+   short-lived peak less likely to fall between samples.
+2. Start from the overhead the executors had before Comet was enabled, which covers the JVM's own
+   non-heap memory, and add the largest difference seen on any executor.
+3. Add a margin on top. The log can miss the true peak between samples, and neither figure includes
+   the allocator's fragmentation and retained pages, memory allocated by native C libraries such as
+   zstd, or Comet's Arrow buffers on the JVM side.
+
+For example, a 16 GiB executor derives an overhead of 1638 MiB. If the largest difference in its
+log is the 1522.3 MiB in the line above, the overhead needs to be at least 1638 + 1523 = 3161 MiB
+before any margin, so `spark.executor.memoryOverhead=4g` would be a reasonable setting.
+
+The executor also logs a warning when its native memory looks larger than its container allows:
+when the difference, plus everything in use in Spark's off-heap memory pool (which includes Comet's
+reservations), exceeds `spark.memory.offHeap.size` plus the memory overhead. This counts the part of
+the off-heap pool that nothing has acquired at that moment, which untracked memory can occupy until
+Spark hands it out, so a quiet log is not a sign that the overhead is large enough: size it from the
+largest difference as described above. The overhead also has to hold the JVM's own non-heap memory,
+so by the time the warning appears the executor has likely outgrown its container. It warns the first time this
+happens, and again each time it happens after dropping back below. The overhead it uses is
+`spark.executor.memoryOverhead` if set, otherwise `spark.executor.memoryOverheadFactor` of
+`spark.executor.memory` with a minimum of `spark.executor.minMemoryOverhead`, as Spark sizes the
+default container. There is no warning in local mode.
+
+Look more closely before raising the overhead if the difference keeps growing through a run rather
+than levelling off: native memory that is not being released will exhaust any overhead eventually.
+The executor logs one more line after its last native plan finishes, and an `allocated` figure there
+that grows from one query to the next points the same way.
+
+`spark.comet.memory.logInterval` is read when an executor starts its first Comet native plan, so set
+it when the application is submitted. Set it to `0` to turn the log off.
+
 ### Determining How Much Memory to Allocate
 
 Generally, increasing the amount of memory allocated to Comet will improve query performance by reducing the

native/common/src/bin/analyze_trace.rs

@@ -31,8 +31,8 @@ use std::{env, fs::File};
 
 /// The process-wide allocation counters the tool understands, most preferred first.
 ///
-/// `native_allocated` (the `alloc-accounting` feature) counts only the bytes Rust code holds from
-/// the global allocator, so it is the tighter comparison against pool reservations.
+/// `native_allocated` (emitted by every build) counts only the bytes Rust code holds from the
+/// global allocator, so it is the tighter comparison against pool reservations.
 /// `jemalloc_allocated` (the `jemalloc` feature) also includes jemalloc's own metadata. A trace
 /// that carries both is analyzed against `native_allocated` alone; a trace with neither cannot be
 /// analyzed.
@@ -250,7 +250,8 @@ fn main() {
     let Some(source) = source.map(|rank| ALLOCATED_COUNTERS[rank]) else {
         eprintln!(
             "No process-wide allocation counter found in the trace: expected one of {}. \
-             Build the native library with the `alloc-accounting` or `jemalloc` feature.",
+             Was the trace produced by a native library older than the `native_allocated` \
+             counter?",
             ALLOCATED_COUNTERS.join(", ")
         );
         std::process::exit(1);

native/core/Cargo.toml

@@ -114,12 +114,6 @@ jemalloc = ["tikv-jemallocator", "tikv-jemalloc-ctl"]
 # Default builds carry zero Delta surface.
 contrib-delta = ["dep:comet-contrib-delta"]
 
-# Observability for real native memory usage. Wraps the global allocator to track the bytes it
-# hands out, and reports the total as the `native_allocated` tracing metric so it can be compared
-# against the memory pool's reservations. Never rejects an allocation. Off by default; a build
-# without it has no wrapper and no per-allocation work.
-alloc-accounting = []
-
 # exclude optional packages from cargo machete verifications
 [package.metadata.cargo-machete]
 ignored = ["hdfs-sys", "paste"]