comphead commented on code in PR #7016: URL: https://github.com/apache/datafusion-comet/pull/7016#discussion_r4338820294
########## docs/source/contributor-guide/memory_management.md: ########## @@ -110,12 +110,14 @@ 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, 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. 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. +(`CometNativeArrowSource`), broadcast coalescing, the decoded broadcasts that an executor's tasks +share (`DecodedBroadcastCache`), 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 Review Comment: Each task's native join still reserves its whole build side. DataFusion 55.2's `collect_left_input` counts batches with a `RecordBatchMemoryCounter`, which skips only buffers already counted by the same plan. So with one shared copy, N tasks reserve N times the bytes that are resident. Spilling behaves as before, but two places assume a JVM buffer that a native operator holds is reserved once: - This guide's line 528 says `reserved()` is a lower bound on Comet's real footprint. That no longer holds while tasks share a broadcast. - `CometExecIterator.untrackedMemory` computes `allocated + jvmArrow.allocatedByJvm - reserved`, and the `nativeMemoryLimitWarning` scaladoc says such bytes "are counted once". With 8 tasks sharing a 100 MB build side, the JVM figure holds 100 MB and `reserved` holds 800 MB, so the untracked figure drops by 700 MB. That can hide other untracked memory in the memory usage log and in the limit warning, or clamp to 0. Could this paragraph say that sharing lowers resident memory but not reservations, and that the untracked figure undercounts while tasks share a broadcast? This is from reading the code. I have not run it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
