Repository navigation
Conversation
A partial aggregate that stops aggregating sends each row on as a partial state, which DataFusion builds with GroupsAccumulator::convert_to_state. Comet's Float64 AvgGroupsAccumulator, which runs Spark's average of a decimal as avg(UnscaledValue(x)), returned NotImplemented. Each row's state is now its value and a count of one, or a zero sum and count for a null or filtered-out row, what state() emits for such a group. (cherry picked from commit 5000f3a)
configure_skip_partial_aggregation only admitted single-argument COUNT. With the double AvgGroupsAccumulator now converting rows to partial states, a single-argument AVG whose result is a double is admitted too. A decimal AVG runs as AvgDecimal, whose convert_to_state is not implemented, so it stays excluded.
andygrove
marked this pull request as draft
November 2, 2026 19:54
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Part of #7406.
Rationale for this change
With
spark.comet.exec.aggregate.skipPartial.enabled, a native partial aggregate can stop aggregating once its input looks mostly distinct and send the rest of the task's rows on as partial states. Comet only allows this when every aggregate function is a single-argument COUNT, so a query such as TPC-H q17, which takes an AVG grouped by a nearly unique key, keeps aggregating. In #7289 this change cut q17 at SF1000 from about 277 s to 242 s and stopped an executor from being OOM-killed. That PR runs the query on Comet Native Executors, but the gate is shared, so the change helps Comet on Spark too. This splits it out of #7289.What changes are included in this PR?
AvgGroupsAccumulator, the Float64 average that Comet uses for every non-decimal AVG (and for Spark's rewrite of a low-precision decimal AVG toavg(UnscaledValue(x))), implementsconvert_to_state. It used to returnNotImplemented. Each row's state is its value and a count of one. A null or filtered-out row gets a zero sum and a zero count, which is the statestate()emits for such a group.configure_skip_partial_aggregationalso admits a single-argument AVG whose result type is a double. A decimal AVG runs asAvgDecimal, whoseconvert_to_stateis still not implemented, so it stays excluded.How are these changes tested?
avg.rschecks thatconvert_to_stategives the same states as aggregating each row into its own group, with a null row and a filtered-out row.skip_partial_eligibility_is_fail_closednow covers a double AVG (admitted), an AVG with no child, and a decimal AVG (both rejected).skip partial aggregation admits only supported native shuffle plansinCometAggregateSuitechecks that a double AVG over nullable columns skips and still matches Spark, and that aDECIMAL(30, 2)AVG does not skip.