parthchandra opened a new issue, #7323:
URL: https://github.com/apache/datafusion-comet/issues/7323
### Describe the bug
### What is the problem the feature request solves?
Spark plans an aggregate as `SortAggregateExec` when its buffer can't live
in an `UnsafeRow` and it isn't a `TypedImperativeAggregate`. One example is
`min`/`max` over a struct, which now runs natively (#7191).
`EnsureRequirements` then puts a `SortExec` on the grouping keys below both the
partial and the final aggregate. Comet converts both into native sorts.
Comet's native aggregate doesn't need sorted input. It is DataFusion's
`AggregateExec`, which hashes groups when the input is unordered. So for the
partial stage, the sort is pure overhead: every input row, with all its
columns, is sorted before being reduced to one row per group.
With wide rows this cost dominates. In `CometAggregateExpressionBenchmark`
("Struct min/max benchmarks": 200k rows, 1,000 groups, 4 KiB string payloads),
`SELECT MAX(struct(k, wide)) FROM t GROUP BY grp` takes:
| | Best time |
|---|---|
| Spark | 258 ms |
| Comet | 398 ms |
The global version without grouping, which has no sort, is at parity.
Measured natively on the same data:
| | Time |
|---|---|
| DataFusion sort by the grouping key | ~200 ms |
| Struct max accumulator | ~20 ms |
With 64-byte payloads the same grouped query is 2.6x faster in Comet, so the
extra cost grows with row width.
### Describe the potential solution
When a `CometSortAggregateExec` runs in `Partial` mode, remove the
`SortExec`/`CometSort` below it that exists only to satisfy
`SortAggregateExec.requiredChildOrdering`, and lthe native aggregate hash its
input.
**Check the partial's parent before dropping the sort.** It's
only safe whel relies on its output ordering. Usually the parent is a shuffle,
whichdoesn't preserve order, so the sort can go. But when the input is already
partitioned by the grouping keys and not ordered by them, Spark plans no
exchange between partial andfinal. One example is the output of a shuffled hash
join on the grouping kew. There, the final `SortAggregateExec` relies on the
partial's outputordering for its own required ordering, so `EnsureRequirements`
adds no sort before the final. Dropping the sort under the partial would leave
the final with unsorted input. Either keep the sort in that case, or move it
up between partial and final,r group instead of every input row. When the child
is already ordered by the grouping keys, for example
a sort-merge join on those keys, Spark adds no sort at all, so there is
nothing to remove.
Other points:
- **Plan-time decision:** the partial aggregate reports
`orderedByGroupingKeys = true` to the native planner. That has to become false
when the sort is removed, so the native side doesn't assume sorted input.
- **Memory behavior:** a sorted partial emits finished groups after each
batch. A hash partial holds every group until it emits early under memory
pressure. Comet's partial aggregates already behave this way for
`HashAggregateExec`, so this should
- **PartialMerge stages** from distinct-aggregate rewrites can likely be
treated the same way when they feed a shuffle, with the same parent check.
The final stage is a separate question. There, the sort after the shuffle
lets DataFusion use its ordered final stream, which keeps memory bounded.
Dropping it would trade sort
time for a hash final that must hold or spill every group. It could be cons
behind a config.
### Additional context
- Benchmark cases: `CometAggregateExpressionBenchmark`, `max_struct_wide`
/d) and `*_global`.
- The no-exchange case, confirmed on Spark 4.1 with broadcast joins and AQE
turned off. The partial and final sort aggregates both run natively with no
exchange between them,
and only the partial has a sort below it:
```sql
SELECT /*+ SHUFFLE_HASH(b) */ a.k, max(struct(a.v, b.s))
FROM a JOIN b ON a.k = b.k
GROUP BY a.k
```
```
CometSortAggregate [Final]
+- CometSortAggregate [Partial]
+- CometSort [k ASC NULLS FIRST]
+- CometProject
+- CometHashJoin [k], [k], Inner, BuildRight
:- CometExchange hashpartitioning(k, 200)
+- CometExchange hashpartitioning(k, 200)
```
- Related: #7191 (native struct min/max), #4565 (`SortAggregateExec` suppor
### Steps to reproduce
_No response_
### Expected behavior
_No response_
### Additional context
_No response_
### Willingness to contribute
None
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
