andygrove opened a new issue, #7406: URL: https://github.com/apache/datafusion-comet/issues/7406
### What is the problem the feature request solves? On Comet Native Executors (#7289), a partial aggregate in a stage never stops aggregating, even when its keys are almost all distinct. TPC-H q17's subquery is `avg(l_quantity) GROUP BY l_partkey` over all of `lineitem`: at SF1000 the partial aggregate takes in about 6 billion rows and puts out about as many. Comet builds a hash table of every key in every partition, while DataFusion (and so Ballista) checks the ratio after the first rows and passes the rest through as partial states. Two things keep Comet from skipping: 1. `configure_skip_partial_aggregation` (`native/core/src/execution/jni_api.rs`) only admits a plan whose root is a `ShuffleWriter`. A Native Executors fragment has no `ShuffleWriter` root (Ballista writes the shuffle), so a fragment is never eligible, even with `spark.comet.exec.aggregate.skipPartial.enabled=true`. 2. The check admits only single-argument `COUNT`. Spark plans a decimal average as `avg(UnscaledValue(x))`, which Comet runs with its `Float64` `AvgGroupsAccumulator` (`native/spark-expr/src/agg_funcs/avg.rs`), and that accumulator's `convert_to_state` returns `NotImplemented`. Measured on TPC-H q17 at SF1000 in pipelined mode, 2 executors of 8 cores and 56 GiB, with an experiment that fixes both (branch `native-executors-skip-partial-exp` on andygrove/datafusion-comet): | | q17 | | --- | ---: | | Default | 293.3 s, 276.7 s, 276.1 s; one executor OOM-killed at 58.6 GB resident on a repeated run | | Skipping, with `spark.comet.exec.aggregate.skipPartial.enabled=true` | 241.8 s, 242.4 s, correct result, no OOM | ### Describe the potential solution - Treat a Native Executors fragment as feeding a shuffle in the eligibility check. The experiment does it by checking the block as if it had a `ShuffleWriter` root when the fragment's session sets a flag. - Implement `convert_to_state` for the `Float64` `AvgGroupsAccumulator`: each row's state is its value and a count of one, or a zero sum and count for a null or filtered-out row, which is what `state()` emits for such a group. The experiment adds this with a unit test that checks it against `update_batch` followed by `state`. - Admit `AVG` in the check, at least for this accumulator. Skipping stays opt-in, since #6466 showed it can shuffle many times more rows when keys repeat after a distinct start. The accumulator and the check also apply to Comet on Spark. ### Additional context Found while comparing q17 with Ballista (#7392). Part of #7288. -- 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]
