You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Run SortAggregateExec as a native hash aggregate without its input sort #7146
Spark plans SortAggregateExec when an aggregate buffer is not UnsafeRow-mutable (for example string min/max), and EnsureRequirements puts a SortExec on the grouping keys under it. Comet keeps that sort, so DataFusion runs its ordered aggregate streams on sorted input. The sort exists only because of Spark's hash map limitation, and in Comet it is the expensive part:
On input sorted on every grouping key, the partial aggregate cannot spill and fails when a reservation is refused. The hash partial emits early instead.
On a job grouping by 87 columns with string min/max, 39 of 40 map tasks failed with Additional allocation failed for OrderedPartialAggregateStream[0], and retries failed in ExternalSorterMerge[0].
Describe the potential solution
When a SortAggregateExec sits on the SortExec for its grouping keys, run the native aggregate over the sort's child:
Partial aggregate under a shuffle exchange: drop the sort and report no output ordering.
Final aggregate: drop the sort, and keep ordered_by_grouping_keys so the native planner sorts the aggregated output, unless the parent is an exchange.
Leave first, last and collect_list on the current path, and gate the change behind a config.
Additional context
Apache Auron does this in AuronConverters.convertSortAggregateExec, unless a parent needs the order. #6936 (hash-mode string min/max under-reports memory) should land first or together. Related: #1994, #4565, #6880.
Activity
comphead commentedon Oct 21, 2026
Potential workaround
comphead commentedon Oct 28, 2026
depends on DF56