andygrove opened a new issue, #6936: URL: https://github.com/apache/datafusion-comet/issues/6936
### Describe the bug #6927 runs `min` and `max` over `STRING` and `BINARY` natively through DataFusion's `min_udaf` and `max_udaf`. Grouped aggregates use DataFusion's `MinMaxBytesAccumulator`, which under-reports its memory. In DataFusion 55.2, `MinMaxBytesState::set_value` (`datafusion-functions-aggregate/src/min_max/min_max_bytes.rs`) replaces a group's value with `clear()` and `extend_from_slice` on the existing `Vec<u8>`. The group keeps the largest allocation it has held. `size()` returns the sum of the value lengths, not the capacities. When a long value is followed by a shorter extremum, the difference is never reported, so the aggregate's reservation is lower than the memory it holds and it spills later than it should. DataFusion main has the same code. sunchao measured this on #6927 with the unmodified 55.2 accumulator. Four cohorts of 64 groups, each receiving a 65,536-byte value and then a one-byte winner, retained 16,783,360 bytes while reporting 6,400 bytes. The same holds for `max`, for binary input, and for `merge_batch`. ### Steps to reproduce The exposed plan shape is a hash aggregate. String or binary `min`/`max` next to a `TypedImperativeAggregate` makes Spark plan `ObjectHashAggregateExec`, for example: ```sql SELECT k, min(s), collect_set(tag) FROM t GROUP BY k ``` It shows up when a group's extremum gets shorter over time. A `min` over URLs or paths, where a prefix arrives after a longer value, is the realistic case. Plain string `min`/`max` is planned as `SortAggregateExec`. Comet runs that as DataFusion's ordered aggregate streams, which emit each group as soon as it is complete, so the retained capacity is freed group by group. Global `min`/`max` uses `MinAccumulator`/`MaxAccumulator`, whose `ScalarValue` size counts the `String` capacity. Neither is affected. ### Expected behavior The accumulator's `size()` covers the capacity it retains, so Comet's memory pool sees what the aggregate holds and spills on time. ### Additional context With a 4 MiB pool the hash shape still spilled and matched Spark, so the visible effect is native memory above the reservation rather than a failed query. The fix belongs in DataFusion. `MinMaxBytesState` can track capacity separately from the length that `evaluate` uses to size its builders, or reallocate when the new value is much shorter than the buffer. We can pick it up with the next DataFusion upgrade or a branch-55 backport. Until then, `spark.sql.execution.useObjectHashAggregateExec=false` makes Spark plan the sort aggregate instead, which is not affected. Found in sunchao's review of #6927: https://github.com/apache/datafusion-comet/pull/6927#discussion_r4253196841 -- 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]
