[andygrove]

Which issue does this PR close?

N/A. This was split out of #7289.

Rationale for this change

spark.comet.exec.aggregate.skipPartial.enabled lets a native partial aggregate stop aggregating once its input looks mostly distinct and send the rest of the task's rows to the shuffle as partial states. It admits only partial aggregates whose functions are all single-argument COUNT, so a high-cardinality GROUP BY that also computes an AVG never benefits. DataFusion builds each row's partial state with GroupsAccumulator::convert_to_state, which Comet's AVG accumulators did not implement.

What changes are included in this PR?

• AvgGroupsAccumulator (Comet's Float64 AVG, which also serves integral inputs after a cast to double, and small decimals that Spark rewrites to avg(UnscaledValue(x))) now implements convert_to_state. Each row becomes its value and a count of one, or a zero sum and a zero count for a null or filtered-out row, which is what state() emits for such a group.
• configure_skip_partial_aggregation admits AVG next to COUNT, only when the aggregate's result type is DOUBLE. A decimal result is built as AvgDecimal, whose convert_to_state returns a "not implemented" error, so admitting it would fail a query once skipping started. A decimal AVG keeps the plan from skipping, as before. Implementing the decimal conversion is left out: its partial state carries overflow in the shared null buffer of the sum and count, which a per-row conversion would have to reproduce exactly.
• The .doc of spark.comet.exec.aggregate.skipPartial.enabled and the Adaptive Partial Aggregation section of the tuning guide describe the admitted aggregates.

This remains opt-in behind the existing experimental spark.comet.exec.aggregate.skipPartial.enabled setting, which defaults to false. Nothing changes for a query that does not enable it.

How are these changes tested?

• Rust: convert_to_state_matches_one_row_groups checks that converting rows gives the same states as aggregating each row alone into its own group, including a null row and a filtered-out row. skip_partial_eligibility_is_fail_closed covers the gate: AVG with a DOUBLE result is admitted, alone and next to COUNT; AVG without an input, without a result type, with a decimal result (alone and next to COUNT), or with a FLOAT result is not.
• Scala: skip partial aggregation converts AVG rows to partial states in CometAggregateSuite runs high-cardinality GROUP BY queries over avg of an int, a double, and combinations with count, and over avg of a DECIMAL(38, 10), with nulls in the values and keys that repeat across tasks. It lowers the probe row threshold to 100 so skipping starts, checks each answer against Spark, and checks the skipped_aggregation_rows metric: positive for the int and double cases, zero for the decimal case.

Commands run: cargo test -p datafusion-comet-spark-expr avg, cargo test -p datafusion-comet --lib skip_partial, cargo clippy --all-targets --workspace -- -D warnings, cargo fmt --check, and ./mvnw test -Dtest=none -Dsuites="org.apache.comet.exec.CometAggregateSuite skip partial aggregation" on JDK 17.

[comphead]

I checked convert_to_state against state() for a one-row group and against both merges that can consume it (Comet's merge_batch, which reads .values(), and Spark's Average, since supportsNativePartialToSparkFinal is true for a non-decimal AVG), and the (0.0, 0) rows are right for both. DataFusion 55 removed supports_convert_to_state(), so the allow-list is the only option here, and the DOUBLE result check picks out the planner's Float64 Avg branch. The comments are about the small-decimal case, a batch-level conversion, and FILTER coverage.

[comphead]

CAST(... AS DECIMAL(10, 2)) / 7 is written as DECIMAL(14, 6) (Divide.resultDecimalType with the literal as DECIMAL(1, 0)). DecimalAggregates only rewrites avg to avg(UnscaledValue(x)) when precision + 4 <= 15, so avg(small) plans as AvgDecimal like avg(big) and never skips. The small-decimal path the description mentions is not exercised. Would dropping the / 7 (keeping DECIMAL(10, 2)) and asserting skipped(avgQuery("avg(small)")) > 0L cover it? I have not run it.

[comphead]

The gate checks for a DOUBLE result, not a non-decimal input. Spark rewrites avg over a decimal with precision 11 or lower into avg(UnscaledValue(x)), which has a DOUBLE result and is admitted, so avg over DECIMAL(10, 2) does skip. Could this line, the matching one in the Eligibility paragraph below, and the .doc in CometConf say AVG except over decimals with precision above 11? Both lines here also run past the wrap the rest of the section uses.

[comphead]

Once skipping starts, every remaining row of the task goes through this loop, one is_valid and push per row. With no nulls and no filter, the common case, the input buffer can be reused as the sums with no copy and the counts filled with ones, as DataFusion's AvgGroupsAccumulator::convert_to_state does. Otherwise one NullBuffer::union of the filter (via filter_to_nulls in collect.rs) and the input nulls gives the mask, and the sums and counts come from one zip over it. Since Comet's merge_batch reads .values(), masked slots still need a zero sum, which this keeps. I have not compiled it.

[comphead]

The filter here has no null slot, so the filter.is_valid(idx) branch is not covered, and none of the Scala queries has a FILTER clause, so opt_filter is only reached from this test. Could one entry here be None, and could the Scala test add something like avg(i) FILTER (WHERE d > 1), which yields a null predicate wherever d is null? A sliced input such as values.slice(1, 3) would also cover the offset handling.