Skip to content

refactor(hash-aggr): Support spilling for ordered aggregation#23657

Open
2010YOUY01 wants to merge 4 commits into
apache:mainfrom
2010YOUY01:split-aggr-spill-order-aggr
Open

refactor(hash-aggr): Support spilling for ordered aggregation#23657
2010YOUY01 wants to merge 4 commits into
apache:mainfrom
2010YOUY01:split-aggr-spill-order-aggr

Conversation

@2010YOUY01

@2010YOUY01 2010YOUY01 commented Jul 17, 2026

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Part of #22710

Rationale for this change

This PR implements larger-than-memory execution for ordered partial and ordered final aggregation. These two modes are implemented first because:

  • [EPIC] Split Aggregation Logic into Dedicated Streams #22710 splits ordered and unordered aggregation into separate implementations.
  • Unordered aggregation internally reuses ordered aggregation for its larger-than-memory execution logic. So larger-than-memory support for ordered aggregation become the groundwork for other aggregation modes.

The high-level plan for two-stage ordered hash aggregation is:

AggregateExec(mode=final, ordered)
--RepartitionExec(hash)
----AggregateExec(mode=partial, ordered)

Group keys are fully ordered (input ordered by a, b; query uses GROUP BY a, b)

Execution must use bounded memory because only one group is active at any given time. Therefore, the implementation simply returns an execution error if a memory reservation fails.

Group keys are partially ordered (input ordered by a; query uses GROUP BY a, b)

At any given time, we must hold all groups for a particular value of a in memory, so it is possible to run out of memory. This is handled as follows:

If the partial aggregate runs out of memory, it directly materializes the accumulated groups as aggregate states and emits them early to the final stage.

The final aggregate continues aggregating until its memory budget is reached, then:

  1. Sorts the aggregated batches by the group keys and writes them to disk.
  2. Repeats this process until the input is exhausted.
  3. Constructs a sort-preserving merge stream over all spill files.
  4. Constructs an OrderedFinalAggregateStream using the sort-preserving merge stream as its input. The new stream's group keys are fully ordered, whereas those of the original stream were only partially ordered. It then evaluates this ordered stream to produce the final result.

Design Tradeoffs

Note that the same implementation is currently shared by two execution paths, controlled by flags: aggregation with fully ordered group keys and aggregation with partially ordered group keys. This is not an ideal pattern. However:

  • Fully ordered aggregation can later be split into a separate implementation with potentially more aggressive optimizations—for example, it does not require a hash table. The the current implementation only have to handle partially-ordered cases.
  • For now, the behavior remains consistent with the original implementation, allowing us to complete the migration sooner.

What changes are included in this PR?

The core changes are in ordered_partial_stream.rs and ordered_final_stream.rs, for the above algorithm. See code comments for implementation details.

Are these changes tested?

After implementing this change, I checked the coverage using cargo llvm-cov and found that this feature was supported by the original implementation but was not covered by either regular tests or fuzz tests. This is a bit of a horror story.

I think the old implementation is heavily multiplexed, so its overall line coverage is probably good. However, spilling for ordered aggregation follows a specific execution path that was never exercised.

Therefore, this PR adds slt test coverage. I also think we need more fuzzing cases (filed #23658)

Are there any user-facing changes?

@github-actions github-actions Bot added sqllogictest SQL Logic Tests (.slt) physical-plan Changes to the physical-plan crate labels Jul 17, 2026
@codecov-commenter

codecov-commenter commented Jul 17, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 79.06977% with 99 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.66%. Comparing base (f151c10) to head (a780de8).
⚠️ Report is 18 commits behind head on main.

Files with missing lines Patch % Lines
...ysical-plan/src/aggregates/ordered_final_stream.rs 79.38% 68 Missing and 6 partials ⚠️
...ical-plan/src/aggregates/ordered_partial_stream.rs 73.68% 10 Missing and 5 partials ⚠️
.../aggregates/aggregate_hash_table/common_ordered.rs 90.32% 0 Missing and 3 partials ⚠️
...afusion/physical-plan/src/aggregates/order/full.rs 0.00% 3 Missing ⚠️
datafusion/physical-plan/src/aggregates/mod.rs 71.42% 0 Missing and 2 partials ⚠️
...tafusion/physical-plan/src/aggregates/order/mod.rs 66.66% 2 Missing ⚠️
Additional details and impacted files
@@            Coverage Diff             @@
##             main   #23657      +/-   ##
==========================================
- Coverage   80.66%   80.66%   -0.01%     
==========================================
  Files        1086     1086              
  Lines      366673   367100     +427     
  Branches   366673   367100     +427     
==========================================
+ Hits       295786   296119     +333     
- Misses      53263    53333      +70     
- Partials    17624    17648      +24     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

physical-plan Changes to the physical-plan crate sqllogictest SQL Logic Tests (.slt)

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants