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.

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.