pmoust opened a new issue, #26165: URL: https://github.com/apache/datafusion/issues/26165
### Describe the bug `GroupedHashAggregateStream` shrinks its `MemoryReservation` to the emptied group state as soon as it emits, while the emitted `RecordBatch`, which holds those same bytes, is still alive and is handed out in `batch_size` slices. From that point until the last slice is dropped downstream, the memory pool believes the bytes are free and can grant them to other consumers or queries. The relevant code in DataFusion 54.1.0 (`datafusion-physical-plan/src/aggregates/row_hash.rs`): 1. `emit` (around line 1103) moves the group state out with `group_values.emit(...)` (around line 1114). For string keys, `GroupValuesBytesView::emit` takes the whole `ArrowBytesViewMap` and converts it zero-copy with `into_state`. 2. `emit` builds the aggregate output arrays with `evaluate` / `state` (around lines 1120-1127). These are new allocations. 3. `update_memory_reservation` (around line 1133) then resizes the reservation to the now-empty state. 4. `set_input_done_and_produce_output` parks the batch in `ExecutionState::ProducingOutput`, and `poll_next` hands it out in `batch_size` slices that share the batch's buffers. So all of it stays live, uncharged, until the last slice is dropped. ### To Reproduce We measured this with a `MemoryPool` that records per-consumer reservations, sampling jemalloc's `stats.allocated` every 20 ms. Each run was a single statement over 20M rows of ClickBench `hits`, with nothing else running. | Statement | Reservation at emit | Allocated at emit | Uncharged | Duration | |---|---|---|---|---| | ClickBench Q33, `GROUP BY WatchID, ClientIP` | fell from 2.22 GB to 0.47 GB | stayed at 1.68 GB | about 1.2 GB | 60-80 ms | | ClickBench Q19 | | | about 0.77 GB | | | ClickBench Q17 | | | about 0.39 GB | | One statement alone survives this. Under concurrency, the pool hands those bytes to other queries while they are still live. In our case, 10 concurrent connections drove the process 2-3 GB past its pool limit and into the kernel OOM killer. ### Expected behavior While the stream is in `ProducingOutput`, its reservation should cover the emitted batch, for example `get_array_memory_size` of what is still to be sliced out. The reservation should shrink as slices are handed out, or when the batch is dropped, rather than immediately at emit. ### Additional context We work around it in our own `MemoryPool`: shrinks from a `GroupedHashAggregateStream[..]` consumer that cannot spill (`!consumer.can_spill()`) stay charged until the consumer unregisters. That over-charges until the stream ends, and it keys on the consumer's name, which is fragile. A fix in the stream itself would be exact. Related: #23393 (`size()` under-reporting), which covers a different part of the same gap. -- 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]
