You are viewing a plain text version of this content. The canonical link for it is here.

Posted to github@datafusion.apache.org by "comphead (via GitHub)" <gi...@apache.org> on 2026/10/21 16:31:48 UTC

[I] Run SortAggregateExec as a native hash aggregate without its input sort [datafusion-comet]

comphead opened a new issue, #7146:
URL: https://github.com/apache/datafusion-comet/issues/7146

### What is the problem the feature request solves?

Spark plans `SortAggregateExec` when an aggregate buffer cannot be updated in place in an `UnsafeRow`, for example `min`/`max` over strings (`UnsafeRow.isMutable` excludes strings, checked in Spark 3.4.3 and 4.1.3). `SortAggregateExec.requiredChildOrdering` is the grouping keys, so `EnsureRequirements` adds a `SortExec` below every sort aggregate, partial and final.

Comet converts both. The aggregate serializes to the same native `HashAggregate` that `HashAggregateExec` uses, but the `CometSort` runs in the same native plan, so DataFusion sees input sorted on every grouping key and picks its ordered aggregate streams. Spark needs that sort only because of its hash map limitation. In Comet it costs memory and CPU (DataFusion 55.2):

1. `ExternalSorter` reserves each buffered batch at its memory size plus its sliced size, about 2x. When it spills, `sort_and_spill_in_mem_batches` frees the `sort_spill_reservation_bytes` headroom, then asks again for the row-format keys and the merge, and another task can take that memory first (apache/datafusion#24739, fixed on DataFusion main by apache/datafusion#24740, not in 55.x).
2. On input sorted on every grouping key, the partial aggregate (`OrderedPartialAggregateStream`) registers as non-spillable and returns `ResourcesExhausted` when a reservation is refused. The hash partial (`PartialHashAggregateStream`) emits its table early instead.
3. The sort compares every row on every grouping key.

Example: a job grouping by 87 columns, most of them `bigint`, with `sum` and string `min`/`max`. About 3M rows per task, 12 tasks per executor, 32 GiB off-heap. The plan is

```
CometScan -> CometSort -> CometSortAggregate (Partial) -> CometExchange -> CometSort -> CometSortAggregate (Final)
```

39 of the 40 map tasks failed with

```
Additional allocation failed for OrderedPartialAggregateStream[0] with top memory consumers (across reservations) as:
ShuffleRepartitioner[0]#12(can spill: true) consumed 2.9 GB, peak 2.9 GB,
ExternalSorterMerge[0]#1(can spill: false) consumed 256.0 MB, peak 3.0 GB,
OrderedPartialAggregateStream[0]#6(can spill: false) consumed 473.9 KB, peak 11.9 MB.
```

and 18 retries failed in `ExternalSorterMerge[0]` during the sort's first spill. Both failures come from the sort.

### Describe the potential solution

When a `SortAggregateExec`'s child is the `SortExec` on its grouping keys, plan the native aggregate over the sort's child, so DataFusion runs its hash streams.

1. Partial aggregates whose parent is a shuffle exchange: drop the sort, set `ordered_by_grouping_keys` to false, and report no output ordering. The exchange hides the order.
2. Final and Complete aggregates: drop the input sort, and keep the order wherever something above may rely on it.
- `planner/aggregate.rs` already sorts the aggregate's output when `ordered_by_grouping_keys` is set and the input is not sorted. That sorts aggregated rows instead of input rows, which is cheaper whenever the aggregate reduces rows.
- `RewriteJoin.restoreRequiredOrdering` re-adds a local sort under any operator whose child no longer satisfies its `requiredChildOrdering`. It cannot see a sort that Spark already removed because of the aggregate's reported ordering (the comment on `CometSortAggregateExec.orderedByGroupingKeys` describes this case). So emit unordered output only when the parent is an exchange, and sort the output otherwise.
3. Keep `first`, `last` and `collect_list` on today's path at first, since their results follow input order.
4. Put it behind a config, off by default until benchmarks show where hash aggregation wins.

Tests: plan checks that no `CometSort` remains under a partial aggregate. Ordering checks where a parent relies on the aggregate's order, for example `sortWithinPartitions` on the keys or a sort-merge join on them. Spill runs with a small pool. Benchmarks over high and low key cardinality, with string `min`/`max`.

### Additional context

- Apache Auron already does this. `AuronConverters.convertSortAggregateExec` runs a `SortAggregateExec` as `NativeAggBase.HashAgg` over `exec.child.children.head` when the child is a sort and the aggregate's `childOrderingRequiredTag` is not set. `AuronConvertStrategy` sets that tag from each parent's `requiredChildOrdering`, and from a write's `requiredOrdering`, and passes it down until a `SortExec`.
- #6936: string and binary `min`/`max` under-report memory in hash aggregates. This change makes that path more common, so it should land first or together.
- Distinct aggregates add `PartialMerge` sort aggregates between exchanges. Without the sort they would run on unsorted input, which may also sidestep #6880, but their parents need the same ordering check.
- Related: #1994, #4565, #6909, apache/datafusion#25804.

--
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: github-unsubscribe@datafusion.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] Run SortAggregateExec as a native hash aggregate without its input sort [datafusion-comet]

Posted by "comphead (via GitHub)" <gi...@apache.org>.

comphead commented on issue #7146:
URL: https://github.com/apache/datafusion-comet/issues/7146#issuecomment-6322451705

### Workaround until this lands: run grouped aggregates on DataFusion's legacy stream

Set both of these, at launch with `--conf` or per session with `SET`:

```
spark.comet.exec.respectDataFusionConfigs=true
spark.comet.datafusion.execution.enable_migration_aggregate=false
```

**Why it avoids the first failure.** For a `Partial` aggregate over input sorted on every grouping key, DataFusion 55.2 uses `OrderedPartialAggregateStream`. It reserves memory for the whole batch it just aggregated before it emits the completed groups. For fully ordered input it returns the error when the pool refuses (`resize_or_take_state_batch` in `ordered_partial_stream.rs`). That is the `Additional allocation failed for OrderedPartialAggregateStream[0]` in the description, and the upstream report is apache/datafusion#26575. The legacy `GroupedHashAggregateStream` emits the completed groups first and only then resizes its reservation, ignoring a refusal (`emit` in `grouped_hash_stream.rs`). Comet forwards `spark.comet.datafusion.*` to the native session when `respectDataFusionConfigs` is on, so the flag reaches DataFusion. With it set, memory errors name `GroupedHashAggregateStream[...]` instead of `OrderedPartialAggregateStream[...]`.

**What was checked.** A standalone DataFusion 55.2.0 reproduction with a `GreedyMemoryPool`, so Comet's pools were not involved: a `Partial` `min` over a string column, 65,536 distinct groups with 100-character keys, and a spillable consumer holding all but 256 KiB of a 64 MiB pool.

| Input and pool | default (`true`) | `false` |
|---|---|---|
| sorted on all keys, 256 KiB left | fails: `Failed to allocate additional 2.7 MB for OrderedPartialAggregateStream[0]` | succeeds |
| sorted on all keys, pool completely full | fails | succeeds |
| sorted on all keys, pool has room | succeeds | succeeds |
| unsorted (hash partial), 256 KiB left | succeeds | succeeds |

Under pressure the legacy run returned the same 65,536 rows as the run with room (same checksum), still sorted on the keys.

**Limits.**
- It only removes this failure. The retry failures in `ExternalSorterMerge[0]` come from the sort (apache/datafusion#24739) and the OOM-killed executors are separate. It has not been run on the 87-key job.
- It is session-wide. Every grouped aggregate Comet runs natively moves to the legacy stream, including hash partial and final aggregates. Performance and final-aggregate spilling were not measured. #6880 shows the legacy spill path failing under a tight shared pool for a shape that already runs on it.
- The legacy stream does not fail when the pool is full, it stops being accounted. The reproduction also completed with the pool completely full. For fully ordered input the unreserved state is about one batch of groups (the log showed a 12 MB request), and it releases its reservation at emit while the emitted batch is still alive (apache/datafusion#26165).
- It is temporary. `enable_migration_aggregate` is documented as going away, and apache/datafusion#25902 proposes removing it after 56.0.0. `respectDataFusionConfigs` is documented as a development and testing option.

--
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] Run SortAggregateExec as a native hash aggregate without its input sort [datafusion-comet]

Posted by "comphead (via GitHub)" <gi...@apache.org>.

comphead commented on issue #7146:
URL: https://github.com/apache/datafusion-comet/issues/7146#issuecomment-6482198486

   depends on DF56


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org