Comet Native Executors: let a fragment's partial aggregate stop aggregating, including AVG #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):

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 Adaptive partial aggregation can shuffle 10x more rows than 1.0.0, with no supported way to turn it off #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 (#7409). Part of #7288.