feat: report JVM Arrow memory in the executor memory usage log

docs/source/contributor-guide/memory_management.md

@@ -113,7 +113,17 @@ for both at once, and raises the pod's memory request by the same amount.
 (`CometNativeArrowSource`), broadcast coalescing, 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.
+ceiling and no backpressure. 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.
+
+Charging these buffers to Spark's off-heap pool instead was tried in
+[#5998](https://github.com/apache/datafusion-comet/pull/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](https://github.com/apache/datafusion-comet/issues/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
@@ -430,7 +440,8 @@ gap is workload-dependent. The margin that covers it has to come from
 one; see [Where Comet's budget comes from](#where-comets-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; see
+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][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).
@@ -519,7 +530,8 @@ much they matter:
   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.
+- **`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.
@@ -531,14 +543,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                               |
-| [Memory usage log](../user-guide/latest/tuning/memory.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                         |
+| 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/memory.md#sizing-the-overhead-from-the-memory-usage-log) | Executor-wide native allocation vs pool reservations, and JVM Arrow memory, 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:
 
@@ -550,7 +562,7 @@ A checklist for triaging an executor OOM kill:
 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.
+   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

docs/source/contributor-guide/tracing.md

@@ -134,6 +134,7 @@ Arrow memory on the JVM side is reported separately, because it is off-heap and
 `jvm_heap_used`. Comet imports batches from native over the Arrow C Data Interface, and Arrow
 charges a buffer to whichever allocator owns it, so those imports are taken against a dedicated
 child allocator and reported as `jvm_arrow_imported`, within the `jvm_arrow_allocated` total.
+The executor's periodic memory usage log reports the same two figures.
 
 Both are allocator charges. They report what each allocator is accountable for, not where the bytes
 were allocated, and their difference is not a bound on the Arrow memory the JVM allocated itself.

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

@@ -149,48 +149,58 @@ While Comet native plans are running, each executor logs its native memory usage
 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)
+Comet native memory usage: allocated 5412.3 MiB, reserved 3890.0 MiB (16 native plans, 8 memory pools); JVM Arrow allocated 310.4 MiB, 96.2 MiB of it imported from native
 ```
 
 - `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.
+- `JVM Arrow allocated` is the Arrow memory Comet holds on the JVM side, such as batches read from
+  Comet's in-memory cache, broadcast data, and batches exchanged with native code or Python workers.
+  The part imported from native was allocated by Comet's native code, so `allocated` already counts
+  it, and the rest the JVM allocated itself. A pool reserves the JVM's part only while a native
+  operator holds on to one of its batches, such as a sort buffering its input. The split follows
+  which allocator a buffer is charged to rather than where it was allocated, so treat the JVM's
+  part as an estimate.
+
+Comet's untracked memory is what Comet holds outside the JVM heap that no pool reserves: `allocated`,
+plus the JVM Arrow figure less the part imported from native, minus `reserved`. 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 most untracked memory in each
+   executor's log. Take all the 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.
+   non-heap memory, and add the most untracked memory seen on any executor.
+3. Add a margin on top. The log can miss the true peak between samples, and none of the figures
+   includes the allocator's fragmentation and retained pages, or memory allocated by native C
+   libraries such as zstd.
 
-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.
+For example, a 16 GiB executor derives an overhead of 1638 MiB. If the line above has the most
+untracked memory in its log, that is 5412.3 + (310.4 - 96.2) - 3890.0 = 1736.5 MiB. The overhead
+then needs to be at least 1638 + 1737 = 3375 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.
+when the untracked memory, 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 most untracked memory 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 untracked memory keeps growing through a run
+rather than levelling off: 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` or
+`JVM Arrow 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.

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -362,9 +362,10 @@ object CometConf extends ShimCometConf {
     .category(CATEGORY_TUNING)
     .doc(
       "How often each executor logs its native memory usage at INFO level while Comet native " +
-        "plans are running: the bytes the native allocator has handed out, and the bytes " +
-        "reserved in Comet's memory pools. The difference is native memory that the pools are " +
-        "not accounting for. The executor logs one line per interval however many tasks are " +
+        "plans are running: the bytes the native allocator has handed out, the bytes reserved " +
+        "in Comet's memory pools, and the Arrow memory Comet holds on the JVM side. The " +
+        "difference between the first two is native memory that the pools are not accounting " +
+        "for. The executor logs one line per interval however many tasks are " +
         "running, and one more after the last plan finishes. It logs a warning when the " +
         "native memory looks larger than the executor's container allows. This is an executor " +
         "setting, read when an executor starts its first Comet native plan, so it must be set " +