You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "2010YOUY01 (via GitHub)" <gi...@apache.org> on 2026/07/17 08:10:03 UTC
[PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
2010YOUY01 opened a new pull request, #23657:
URL: https://github.com/apache/datafusion/pull/23657
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Part of https://github.com/apache/datafusion/issues/22710
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue, then this section is not needed.
Clearly explaining why changes are proposed helps reviewers understand your changes and offer better suggestions.
-->
This PR implements larger-than-memory execution for ordered partial and ordered final aggregation. These two modes are implemented first because:
- #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?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes 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?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
6. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
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 (TODO: open an issue).
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3638086781
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -64,6 +119,135 @@ type OrderedFinalAggregateStateTransition = ControlFlow<
OrderedFinalAggregateState,
>;
+impl OrderedFinalSpillContext {
+ fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ batch_size: usize,
+ input_order_mode: &InputOrderMode,
+ spill_schema: &SchemaRef,
+ spill_metrics: SpillMetrics,
+ ) -> Result<Self> {
+ let group_schema = agg.group_by.group_schema(spill_schema)?;
+ let output_ordering = agg.cache.output_ordering();
+ let InputOrderMode::PartiallySorted(order_indices) = input_order_mode else {
+ return internal_err!("Ordered final spill requires partially ordered input");
+ };
+ let spill_indices = order_indices.iter().copied().chain(
+ (0..group_schema.fields().len()).filter(|idx| !order_indices.contains(idx)),
+ );
+ let spill_sort_exprs = spill_indices.map(|idx| {
+ let field = group_schema.field(idx);
+ let output_expr = Column::new(field.name(), idx);
+ let sort_options = output_ordering
+ .and_then(|ordering| ordering.get_sort_options(&output_expr))
+ .unwrap_or_default();
+ PhysicalSortExpr::new(Arc::new(output_expr), sort_options)
+ });
+ let Some(spill_expr) = LexOrdering::new(spill_sort_exprs) else {
+ return internal_err!("Ordered final spill expression is empty");
+ };
+
+ let spill_manager = SpillManager::new(
+ context.runtime_env(),
+ spill_metrics,
+ Arc::clone(spill_schema),
+ )
+ .with_compression_type(context.session_config().spill_compression());
+
+ Ok(Self {
+ agg: agg.clone(),
+ context: Arc::clone(context),
+ partition,
+ batch_size,
+ spill_expr,
+ spill_manager,
+ spills: vec![],
+ })
+ }
+
+ fn has_spills(&self) -> bool {
+ !self.spills.is_empty()
+ }
+
+ /// Sort and spill the aggregated groups. Memory reservation should be cleared
Review Comment:
added
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3638079850
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -64,6 +119,135 @@ type OrderedFinalAggregateStateTransition = ControlFlow<
OrderedFinalAggregateState,
>;
+impl OrderedFinalSpillContext {
+ fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ batch_size: usize,
+ input_order_mode: &InputOrderMode,
+ spill_schema: &SchemaRef,
+ spill_metrics: SpillMetrics,
+ ) -> Result<Self> {
+ let group_schema = agg.group_by.group_schema(spill_schema)?;
+ let output_ordering = agg.cache.output_ordering();
+ let InputOrderMode::PartiallySorted(order_indices) = input_order_mode else {
+ return internal_err!("Ordered final spill requires partially ordered input");
+ };
+ let spill_indices = order_indices.iter().copied().chain(
+ (0..group_schema.fields().len()).filter(|idx| !order_indices.contains(idx)),
+ );
+ let spill_sort_exprs = spill_indices.map(|idx| {
+ let field = group_schema.field(idx);
+ let output_expr = Column::new(field.name(), idx);
+ let sort_options = output_ordering
+ .and_then(|ordering| ordering.get_sort_options(&output_expr))
+ .unwrap_or_default();
+ PhysicalSortExpr::new(Arc::new(output_expr), sort_options)
+ });
+ let Some(spill_expr) = LexOrdering::new(spill_sort_exprs) else {
+ return internal_err!("Ordered final spill expression is empty");
+ };
+
+ let spill_manager = SpillManager::new(
+ context.runtime_env(),
+ spill_metrics,
+ Arc::clone(spill_schema),
+ )
+ .with_compression_type(context.session_config().spill_compression());
+
+ Ok(Self {
+ agg: agg.clone(),
+ context: Arc::clone(context),
+ partition,
+ batch_size,
+ spill_expr,
+ spill_manager,
+ spills: vec![],
+ })
+ }
+
+ fn has_spills(&self) -> bool {
+ !self.spills.is_empty()
+ }
+
+ /// Sort and spill the aggregated groups. Memory reservation should be cleared
+ /// by the caller of this function.
+ ///
+ /// See [`OrderedFinalAggregateStream`] for spilling details.
+ ///
+ /// It must not be called without rows to spill in the aggregate table, otherwise
Review Comment:
This boolean is not needed. Simplified it in [e1d551a](https://github.com/apache/datafusion/pull/23657/commits/e1d551aeb5a165e3a22ef1a0141ff27c9fe487ba)
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5058448659
Thanks for the review, all comments are addressed.
> As an aside / editorial comment: I am really liking this refactor -- I can already see ahead to when we can do stuff like "we don't implement segmented storage for ordered aggregate table, as it is less likely to accumulate massive groups" -- and we can scope new features more carefully / easily. Love it
Absolutely!
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5041363329
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/23657#issuecomment-5041227943)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
Comparing HEAD and split-aggr-spill-order-aggr
--------------------
Benchmark clickbench_partitioned.json
--------------------
βββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β split-aggr-spill-order-aggr β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 0 β 1.43 / 4.50 Β±5.91 / 16.32 ms β 1.23 / 4.07 Β±5.56 / 15.18 ms β +1.11x faster β
β QQuery 1 β 14.22 / 14.36 Β±0.11 / 14.54 ms β 13.18 / 13.46 Β±0.15 / 13.60 ms β +1.07x faster β
β QQuery 2 β 37.41 / 37.83 Β±0.22 / 38.04 ms β 36.76 / 37.13 Β±0.27 / 37.56 ms β no change β
β QQuery 3 β 32.95 / 33.45 Β±0.64 / 34.68 ms β 30.82 / 31.13 Β±0.34 / 31.75 ms β +1.07x faster β
β QQuery 4 β 263.65 / 266.47 Β±1.76 / 269.07 ms β 230.58 / 254.16 Β±18.39 / 271.85 ms β no change β
β QQuery 5 β 301.89 / 307.38 Β±4.39 / 314.45 ms β 282.38 / 301.32 Β±12.86 / 314.94 ms β no change β
β QQuery 6 β 1.43 / 1.62 Β±0.22 / 2.04 ms β 1.26 / 1.42 Β±0.22 / 1.86 ms β +1.14x faster β
β QQuery 7 β 15.12 / 15.27 Β±0.14 / 15.51 ms β 14.08 / 14.22 Β±0.09 / 14.32 ms β +1.07x faster β
β QQuery 8 β 324.06 / 347.25 Β±18.38 / 375.08 ms β 330.76 / 342.97 Β±12.91 / 367.35 ms β no change β
β QQuery 9 β 456.79 / 460.43 Β±2.21 / 463.73 ms β 456.20 / 469.52 Β±8.63 / 480.24 ms β no change β
β QQuery 10 β 71.44 / 72.59 Β±0.74 / 73.49 ms β 71.37 / 72.29 Β±1.47 / 75.22 ms β no change β
β QQuery 11 β 82.51 / 83.98 Β±0.98 / 85.40 ms β 82.28 / 83.44 Β±0.88 / 84.95 ms β no change β
β QQuery 12 β 263.47 / 270.21 Β±4.66 / 276.19 ms β 269.20 / 274.42 Β±5.32 / 283.81 ms β no change β
β QQuery 13 β 361.63 / 386.33 Β±16.61 / 408.60 ms β 367.18 / 381.74 Β±11.02 / 397.92 ms β no change β
β QQuery 14 β 300.13 / 320.67 Β±11.76 / 335.92 ms β 281.23 / 288.56 Β±5.39 / 295.85 ms β +1.11x faster β
β QQuery 15 β 264.55 / 280.03 Β±12.19 / 294.41 ms β 268.41 / 278.78 Β±10.74 / 298.92 ms β no change β
β QQuery 16 β 602.72 / 617.16 Β±12.19 / 639.22 ms β 612.51 / 621.48 Β±7.65 / 633.20 ms β no change β
β QQuery 17 β 610.25 / 627.67 Β±11.17 / 641.47 ms β 626.52 / 701.23 Β±46.70 / 765.03 ms β 1.12x slower β
β QQuery 18 β 1261.75 / 1300.37 Β±26.45 / 1340.54 ms β 1284.15 / 1425.48 Β±81.92 / 1530.93 ms β 1.10x slower β
β QQuery 19 β 30.08 / 30.86 Β±0.71 / 32.18 ms β 29.63 / 31.80 Β±3.63 / 39.02 ms β no change β
β QQuery 20 β 515.71 / 534.93 Β±19.25 / 566.32 ms β 517.63 / 541.95 Β±32.26 / 604.60 ms β no change β
β QQuery 21 β 515.39 / 521.35 Β±5.67 / 530.70 ms β 517.12 / 532.60 Β±16.50 / 561.66 ms β no change β
β QQuery 22 β 981.14 / 995.98 Β±12.88 / 1019.67 ms β 987.05 / 1003.09 Β±15.87 / 1030.78 ms β no change β
β QQuery 23 β 3075.00 / 3137.92 Β±83.61 / 3299.60 ms β 3015.24 / 3106.99 Β±64.63 / 3205.19 ms β no change β
β QQuery 24 β 41.34 / 41.70 Β±0.34 / 42.33 ms β 41.17 / 42.04 Β±0.91 / 43.79 ms β no change β
β QQuery 25 β 110.94 / 113.35 Β±2.11 / 116.88 ms β 113.05 / 117.75 Β±5.34 / 125.37 ms β no change β
β QQuery 26 β 41.93 / 45.60 Β±4.86 / 55.07 ms β 41.98 / 44.24 Β±2.55 / 48.98 ms β no change β
β QQuery 27 β 514.80 / 530.48 Β±13.75 / 553.67 ms β 515.77 / 520.79 Β±3.05 / 524.50 ms β no change β
β QQuery 28 β 2875.49 / 2991.36 Β±76.41 / 3075.96 ms β 2884.52 / 2925.57 Β±35.07 / 2989.42 ms β no change β
β QQuery 29 β 42.94 / 53.75 Β±11.65 / 74.71 ms β 41.10 / 48.22 Β±13.91 / 76.04 ms β +1.11x faster β
β QQuery 30 β 298.55 / 309.45 Β±10.21 / 327.88 ms β 299.13 / 305.23 Β±5.68 / 313.00 ms β no change β
β QQuery 31 β 287.02 / 302.40 Β±12.46 / 315.44 ms β 289.24 / 296.02 Β±6.29 / 307.44 ms β no change β
β QQuery 32 β 913.44 / 970.17 Β±43.94 / 1042.34 ms β 937.85 / 1043.50 Β±55.27 / 1090.44 ms β 1.08x slower β
β QQuery 33 β 1429.48 / 1509.79 Β±76.74 / 1651.84 ms β 1465.94 / 1577.65 Β±89.27 / 1687.59 ms β no change β
β QQuery 34 β 1472.14 / 1504.33 Β±45.99 / 1594.24 ms β 1455.71 / 1538.87 Β±62.04 / 1626.48 ms β no change β
β QQuery 35 β 272.39 / 319.34 Β±52.24 / 397.50 ms β 282.59 / 309.30 Β±30.56 / 349.83 ms β no change β
β QQuery 36 β 65.80 / 67.22 Β±0.99 / 68.48 ms β 67.42 / 69.14 Β±1.14 / 70.96 ms β no change β
β QQuery 37 β 39.13 / 45.01 Β±4.86 / 52.25 ms β 36.21 / 41.62 Β±7.57 / 55.94 ms β +1.08x faster β
β QQuery 38 β 41.53 / 47.76 Β±4.80 / 53.75 ms β 42.57 / 45.35 Β±3.38 / 51.64 ms β +1.05x faster β
β QQuery 39 β 144.91 / 155.69 Β±10.76 / 173.83 ms β 146.68 / 154.77 Β±6.12 / 163.36 ms β no change β
β QQuery 40 β 14.99 / 18.66 Β±2.68 / 21.04 ms β 14.37 / 15.10 Β±0.99 / 17.05 ms β +1.24x faster β
β QQuery 41 β 14.24 / 15.18 Β±1.40 / 17.94 ms β 14.14 / 16.91 Β±4.88 / 26.66 ms β 1.11x slower β
β QQuery 42 β 12.97 / 13.34 Β±0.25 / 13.75 ms β 13.59 / 15.30 Β±2.74 / 20.75 ms β 1.15x slower β
βββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββ
ββββββββββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 19723.16ms β
β Total Time (split-aggr-spill-order-aggr) β 19940.60ms β
β Average Time (HEAD) β 458.68ms β
β Average Time (split-aggr-spill-order-aggr) β 463.73ms β
β Queries Faster β 10 β
β Queries Slower β 5 β
β Queries with No Change β 28 β
β Queries with Failure β 0 β
ββββββββββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**clickbench_partitioned β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 100.0s |
| Peak memory | 12.6 GiB |
| Avg memory | 4.7 GiB |
| CPU user | 1012.1s |
| CPU sys | 68.9s |
| Peak spill | 0 B |
**clickbench_partitioned β branch**
| Metric | Value |
|--------|-------|
| Wall time | 105.0s |
| Peak memory | 11.3 GiB |
| Avg memory | 4.2 GiB |
| CPU user | 1021.4s |
| CPU sys | 73.2s |
| Peak spill | 0 B |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3637354976
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -157,21 +409,86 @@ impl OrderedFinalAggregateStream {
if let Err(e) = result {
return ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
));
}
+ // Check memory reservation, and potentially spill.
let timer = elapsed_compute.timer();
- let result = table.next_output_batch();
+ let resize_result =
+ self.reservation
+ .try_resize(Self::reservation_size_for_table(
+ &table,
+ spill_context.as_deref(),
+ ));
timer.done();
+ match resize_result {
+ Ok(()) => {}
+ Err(e @ DataFusionError::ResourcesExhausted(_)) => {
+ let Some(spill_context) = spill_context else {
+ return ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ ));
+ };
+ if table.is_empty() {
+ return ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ ));
+ }
+ return ControlFlow::Continue(
+ OrderedFinalAggregateState::Spilling {
+ table,
+ spill_context,
+ },
+ );
+ }
+ Err(e) => {
+ return ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ ));
+ }
+ }
+
+ let result = if spill_context
+ .as_ref()
+ .is_some_and(|spill_context| spill_context.has_spills())
+ {
+ // Once one incomplete run is spilled, every remaining state
+ // must participate in replay so no group is finalized twice.
+ Ok(None)
+ } else {
+ let timer = elapsed_compute.timer();
+ let result = table.next_output_batch();
+ timer.done();
+ result
+ };
match result {
// Some finalized groups can be emitted. Yield them, then
// continue aggregating input in the current state.
Ok(Some(batch)) => {
- let next_state =
- OrderedFinalAggregateState::ReadingInput { table };
- self.resize_reservation_for_state(&next_state);
+ if let Err(e) =
Review Comment:
Tracked in the clean-up items in https://github.com/apache/datafusion/issues/22710
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5041242899
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/23657#issuecomment-5041227943)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c5041227943-1214-lmmw4 6.12.85+ #1 SMP Mon May 11 08:17:35 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing split-aggr-spill-order-aggr (a780de8773903645de7432bb6372c42ff8e549f4) to f151c10 (merge-base) [diff](https://github.com/apache/datafusion/compare/f151c10fdee89bd6f42c0f786988bd09557ed0fd..a780de8773903645de7432bb6372c42ff8e549f4) using: tpcds
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3638098844
##########
datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt:
##########
@@ -0,0 +1,201 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# End-to-end tests for ordered aggregation under finite memory.
+
+# Result set more than 100 lines will be hashed
+hash-threshold 100
+
+statement ok
+SET datafusion.execution.target_partitions = 2
+
+statement ok
+SET datafusion.execution.batch_size = 128
+
+statement ok
+SET datafusion.optimizer.repartition_aggregations = true
+
+statement ok
+SET datafusion.optimizer.prefer_existing_sort = true
+
+statement ok
+SET datafusion.execution.enable_migration_aggregate = true
Review Comment:
Yes, I think we could leave it here to be more explicit.
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3638101166
##########
datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt:
##########
@@ -0,0 +1,201 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# End-to-end tests for ordered aggregation under finite memory.
+
+# Result set more than 100 lines will be hashed
+hash-threshold 100
+
+statement ok
+SET datafusion.execution.target_partitions = 2
+
+statement ok
+SET datafusion.execution.batch_size = 128
+
+statement ok
+SET datafusion.optimizer.repartition_aggregations = true
+
+statement ok
+SET datafusion.optimizer.prefer_existing_sort = true
+
+statement ok
+SET datafusion.execution.enable_migration_aggregate = true
+
+statement ok
+SET datafusion.runtime.memory_limit = '1M'
+
+# ==================================================================================
+# Input is fully ordered by group keys (input order by (a,b), query is 'group by a,b')
+# ==================================================================================
+
+# Fully ordered input uses the ordered partial and final streams without spill.
+query TT
+EXPLAIN ANALYZE
+SELECT v1, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY v1
+----
+Plan with Metrics
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>ordering_mode=Sorted, metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+02)--RepartitionExec:<slt:ignore>preserve_order=true<slt:ignore>
+03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=Sorted, metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+<slt:ignore>
+
+query II rowsort
+SELECT v1, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY v1
+----
+40002 values hashing to 34c2b23730596cbd2489ed4a627c17d7
+
+# The same fully ordered query cannot spill and reports OOM under tighter memory.
+statement ok
+SET datafusion.runtime.memory_limit = '1K'
+
+query error Resources exhausted
+SELECT v1, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY v1
+
+# ==================================================================================
+# Input is partially ordered by group keys (input order by (a), query is 'group by a,b')
+#
+# Try different memory limits, ensure result is the same, but spill count differ
+
+# HACK: check `spilled_bytes=x KB` to ensure it has spilled. If it has not spilled,
+# the it shows `spilled_bytes = 0B`. Should better check spill_count, but it's not
+# stable due to ordered hash repartition, and `sqllogictest` don't support regex.
+# ==================================================================================
+
+statement ok
+SET datafusion.runtime.memory_limit = '2M'
+
+statement ok
+SET datafusion.optimizer.enable_round_robin_repartition = false
+
+# Round 1: The input is partially ordered and does not spill with a 2 MB limit.
+query TT
+EXPLAIN ANALYZE
+SELECT round(v1, -4), v1 % 5000, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY round(v1, -4), v1 % 5000
+----
+Plan with Metrics
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+02)--RepartitionExec:<slt:ignore>input_partitions=1, maintains_sort_order=true<slt:ignore>
+03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+<slt:ignore>
+
+# All rounds should have the same result hash
+query III rowsort
+SELECT round(v1, -4), v1 % 5000, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY round(v1, -4), v1 % 5000
+----
+45000 values hashing to e6ece4b4b86e6152a1c785e90ab5ba12
+
+# Round 2: The same query spills five times with a 600 KB limit.
+statement ok
+SET datafusion.runtime.memory_limit = '600K'
+
+query TT
+EXPLAIN ANALYZE
+SELECT round(v1, -4), v1 % 5000, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY round(v1, -4), v1 % 5000
+----
+Plan with Metrics
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spilled_bytes=<slt:ignore> KB,<slt:ignore>]
+02)--RepartitionExec:<slt:ignore>input_partitions=1, maintains_sort_order=true<slt:ignore>
+03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+<slt:ignore>
+
+# All rounds should have the same result hash
Review Comment:
I agree, but this might need some extension to sqllogictest runner π€
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5073702510
π
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3638095484
##########
datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs:
##########
@@ -69,14 +70,50 @@ use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream, metric
/// `k = 100`, it is safe to emit all groups with keys less than 100 because the
/// input is ordered.
///
+/// # Memory Pressure and Spilling
+///
+/// ## Fully ordered case
+///
+/// If the input is ordered by every group key, for example:
+///
+/// - Input order: `a, b`
+/// - `GROUP BY`: `a, b`
+///
+/// Completed groups can be emitted as soon as the next group is observed. Thus,
+/// only the current group remains active after completed groups are emitted, and
+/// memory usage does not grow with the total number of groups.
+///
+/// If a memory reservation nevertheless fails, the stream returns the error
+/// directly, indicating an unexpected behavior.
+///
+/// ## Partially ordered case
+///
+/// If the input is ordered by only a subset of the group keys, for example:
+///
+/// - Input order: `a`
+/// - `GROUP BY`: `a, b`
+///
+/// If one `a` value contains many distinct `b` values, the table may accumulate
+/// enough groups to exceed the memory limit.
+///
+/// - `OrderedPartialAggregateStream`: On reservation failure, it emits all current
+/// intermediate states downstream and resets the table. The final stage can
+/// merge repeated `(a, b)` state rows, so no disk spill is required.
+/// - `OrderedFinalAggregateStream`: It cannot emit incomplete final results. On
+/// reservation failure, it sorts the current intermediate states by the complete
+/// group key and spills them as one run. After the input ends, it spills any
+/// remaining states, performs a sort-preserving merge of all runs, and feeds the
+/// merged input into a fully ordered final aggregate stream.
+///
/// ## Implementation Note
///
/// This is intentionally kept simple and closely maps to
/// `GroupedHashAggregateStream` to finish the refactor sooner.
///
+/// More applicable optimizations are left to future work.
Review Comment:
Removed. That was just what I was thinking when implementing it, there is no existing issue for it, I'll try to explore it later.
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5041242630
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/23657#issuecomment-5041227943)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c5041227943-1213-64mhz 6.12.85+ #1 SMP Mon May 11 08:17:35 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing split-aggr-spill-order-aggr (a780de8773903645de7432bb6372c42ff8e549f4) to f151c10 (merge-base) [diff](https://github.com/apache/datafusion/compare/f151c10fdee89bd6f42c0f786988bd09557ed0fd..a780de8773903645de7432bb6372c42ff8e549f4) using: clickbench_partitioned
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5041227943
run benchmarks
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3637382793
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
Review Comment:
I agree this is a better pattern, but the existing implementation do explicitly `drop(state)` on `ControlFlow::Break`. We could clean it up later, tracked in the EPIC similarly.
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3637263902
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -23,22 +23,44 @@ use std::task::{Context, Poll};
use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
-use datafusion_common::Result;
+use datafusion_common::{DataFusionError, Result, internal_err};
use datafusion_execution::TaskContext;
use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use datafusion_physical_expr::PhysicalSortExpr;
+use datafusion_physical_expr::expressions::Column;
+use datafusion_physical_expr_common::sort_expr::LexOrdering;
use futures::stream::{Stream, StreamExt};
use super::AggregateExec;
use super::aggregate_hash_table::{FinalMarker, OrderedAggregateTable};
+use super::group_values::GroupByMetrics;
use crate::aggregates::AggregateMode;
use crate::metrics::{BaselineMetrics, RecordOutput, SpillMetrics};
+use crate::sorts::IncrementalSortIterator;
+use crate::sorts::streaming_merge::{SortedSpillFile, StreamingMergeBuilder};
+use crate::spill::spill_manager::SpillManager;
use crate::stream::EmptyRecordBatchStream;
use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream};
/// Final aggregate stream for `InputOrderMode::Sorted` and
/// `InputOrderMode::PartiallySorted`.
///
/// See comments at [`super::ordered_partial_stream`] for details.
+///
+/// # Spilling
+///
+/// This section is only for implementation notes, for background, see [`super::ordered_partial_stream`]
Review Comment:
[ae022e6](https://github.com/apache/datafusion/pull/23657/commits/ae022e6f613be0760f1426b36319749adf80cb7b)
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -47,13 +69,46 @@ pub(crate) struct OrderedFinalAggregateStream {
state: Option<OrderedFinalAggregateState>,
}
+/// Spill configuration and accumulated runs for partially ordered final
+/// aggregation. Each file is one fully group-key-sorted intermediate-state run;
Review Comment:
[d8fc050](https://github.com/apache/datafusion/pull/23657/commits/d8fc0509f9e215ffcc9d58e84fa4cd24ed0a979e)
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3626896203
##########
datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs:
##########
@@ -234,6 +250,43 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
+ self.buffer.group_indices.allocated_size()
}
+ pub(in crate::aggregates) fn group_by_metrics(&self) -> GroupByMetrics {
+ self.group_by_metrics.clone()
+ }
+
+ /// Takes every intermediate aggregate state and resets the table so it can
+ /// continue with a new ordered input segment.
+ ///
+ /// Unlike normal ordered emission, this operation is allowed to take the
+ /// active (incomplete) groups. Partial aggregation can pass those states to
+ /// its final stage, while final aggregation sorts and spills them before
+ /// replay.
+ pub(in crate::aggregates) fn take_state_batch(
+ &mut self,
+ ) -> Result<Option<RecordBatch>> {
+ if self.buffer.group_values.is_empty() {
+ return Ok(None);
+ }
+
+ let mut output = self.buffer.group_values.emit(EmitTo::All)?;
+ for acc in &mut self.buffer.accumulators {
+ output.extend(acc.state(EmitTo::All)?);
+ }
+
+ let batch = RecordBatch::try_new(Arc::clone(&self.state_schema), output)?;
+ debug_assert!(batch.num_rows() > 0);
+
+ // `emit(EmitTo::All)` resets accumulator state. Explicitly shrink the
Review Comment:
One thought I had was maybe it would be worth moving all this resetting of fields of self.buffer into a method on `OrderedAggregateTableBuffer`? likewise for instead of directly manipulating `self.buffer.group_values` we could add a method like `OrderedAggregateTableBuffer::emit` to ecapsulate the fields more
On the other hand, maybe that just obscures the intent more π€· I don't feel strongly -- just a thought
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -47,13 +69,46 @@ pub(crate) struct OrderedFinalAggregateStream {
state: Option<OrderedFinalAggregateState>,
}
+/// Spill configuration and accumulated runs for partially ordered final
+/// aggregation. Each file is one fully group-key-sorted intermediate-state run;
Review Comment:
I read this comment as suggesting there is one file per group that is spilled (e.g each group is spilled into its own file).
I think it is more like a file per spill event (aka the all the groups in memory at the time of a spill, hen sorted by group key)
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -23,22 +23,44 @@ use std::task::{Context, Poll};
use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
-use datafusion_common::Result;
+use datafusion_common::{DataFusionError, Result, internal_err};
use datafusion_execution::TaskContext;
use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use datafusion_physical_expr::PhysicalSortExpr;
+use datafusion_physical_expr::expressions::Column;
+use datafusion_physical_expr_common::sort_expr::LexOrdering;
use futures::stream::{Stream, StreamExt};
use super::AggregateExec;
use super::aggregate_hash_table::{FinalMarker, OrderedAggregateTable};
+use super::group_values::GroupByMetrics;
use crate::aggregates::AggregateMode;
use crate::metrics::{BaselineMetrics, RecordOutput, SpillMetrics};
+use crate::sorts::IncrementalSortIterator;
+use crate::sorts::streaming_merge::{SortedSpillFile, StreamingMergeBuilder};
+use crate::spill::spill_manager::SpillManager;
use crate::stream::EmptyRecordBatchStream;
use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream};
/// Final aggregate stream for `InputOrderMode::Sorted` and
/// `InputOrderMode::PartiallySorted`.
///
/// See comments at [`super::ordered_partial_stream`] for details.
+///
+/// # Spilling
+///
+/// This section is only for implementation notes, for background, see [`super::ordered_partial_stream`]
Review Comment:
I think the context is actually on `OrderedPartialAggregateStream`
```suggestion
/// This section is only for implementation notes, for background, see [`super::ordered_partial_stream::OrderedPartialAggregateStream`]
```
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -47,13 +69,46 @@ pub(crate) struct OrderedFinalAggregateStream {
state: Option<OrderedFinalAggregateState>,
}
+/// Spill configuration and accumulated runs for partially ordered final
+/// aggregation. Each file is one fully group-key-sorted intermediate-state run;
+/// all runs are replayed together after the original input is exhausted.
+struct OrderedFinalSpillContext {
+ /// Aggregate configuration
+ agg: AggregateExec,
+ /// Task context
+ context: Arc<TaskContext>,
+ /// Original partition index
+ partition: usize,
+ /// Target batch size from configuration
+ batch_size: usize,
+ /// Full group-key ordering, such ordering with be kept in: a) individual spill
+ /// files, b) order after final merging and streaming aggregate
+ spill_expr: LexOrdering,
+ /// Spill I/O and metrics manager.
+ spill_manager: SpillManager,
+ /// Fully sorted spill runs waiting to be merged.
+ spills: Vec<SortedSpillFile>,
+}
+
/// See comments at `poll_next()` for details.
enum OrderedFinalAggregateState {
ReadingInput {
table: OrderedAggregateTable<FinalMarker>,
+ spill_context: Option<Box<OrderedFinalSpillContext>>,
},
- DrainingFinal {
+ Spilling {
table: OrderedAggregateTable<FinalMarker>,
+ spill_context: Box<OrderedFinalSpillContext>,
+ },
+ ProducingOutput {
+ table: OrderedAggregateTable<FinalMarker>,
+ },
+ PreparingMergeInput {
Review Comment:
it might help readability to have a sentence or two comment explaining what the difference is between `PreparingMergeInput` and `MergingSpills`
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -64,6 +119,135 @@ type OrderedFinalAggregateStateTransition = ControlFlow<
OrderedFinalAggregateState,
>;
+impl OrderedFinalSpillContext {
+ fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ batch_size: usize,
+ input_order_mode: &InputOrderMode,
+ spill_schema: &SchemaRef,
+ spill_metrics: SpillMetrics,
+ ) -> Result<Self> {
+ let group_schema = agg.group_by.group_schema(spill_schema)?;
+ let output_ordering = agg.cache.output_ordering();
+ let InputOrderMode::PartiallySorted(order_indices) = input_order_mode else {
+ return internal_err!("Ordered final spill requires partially ordered input");
+ };
+ let spill_indices = order_indices.iter().copied().chain(
+ (0..group_schema.fields().len()).filter(|idx| !order_indices.contains(idx)),
+ );
+ let spill_sort_exprs = spill_indices.map(|idx| {
+ let field = group_schema.field(idx);
+ let output_expr = Column::new(field.name(), idx);
+ let sort_options = output_ordering
+ .and_then(|ordering| ordering.get_sort_options(&output_expr))
+ .unwrap_or_default();
+ PhysicalSortExpr::new(Arc::new(output_expr), sort_options)
+ });
+ let Some(spill_expr) = LexOrdering::new(spill_sort_exprs) else {
+ return internal_err!("Ordered final spill expression is empty");
+ };
+
+ let spill_manager = SpillManager::new(
+ context.runtime_env(),
+ spill_metrics,
+ Arc::clone(spill_schema),
+ )
+ .with_compression_type(context.session_config().spill_compression());
+
+ Ok(Self {
+ agg: agg.clone(),
+ context: Arc::clone(context),
+ partition,
+ batch_size,
+ spill_expr,
+ spill_manager,
+ spills: vec![],
+ })
+ }
+
+ fn has_spills(&self) -> bool {
+ !self.spills.is_empty()
+ }
+
+ /// Sort and spill the aggregated groups. Memory reservation should be cleared
+ /// by the caller of this function.
+ ///
+ /// See [`OrderedFinalAggregateStream`] for spilling details.
+ ///
+ /// It must not be called without rows to spill in the aggregate table, otherwise
+ /// it returns `Ok(false)`, and the caller propagates the error.
+ fn spill_table(
+ &mut self,
+ table: &mut OrderedAggregateTable<FinalMarker>,
+ ) -> Result<bool> {
+ let Some(batch) = table.take_state_batch()? else {
+ return Ok(false);
+ };
+
+ let sorted_iter =
+ IncrementalSortIterator::new(batch, self.spill_expr.clone(), self.batch_size);
Review Comment:
TIL `IncrementalSortIterator` (from @EmilyMatt in #20314 it seems β€οΈ )
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
}
}
Poll::Ready(Some(Err(e))) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
Poll::Ready(None) => {
self.close_input();
- table.input_done();
- ControlFlow::Continue(OrderedFinalAggregateState::DrainingFinal { table })
+ match spill_context {
+ Some(spill_context) if spill_context.has_spills() => {
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::PreparingMergeInput {
+ table,
+ spill_context,
+ },
+ )
+ }
+ _ => {
+ table.input_done();
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::ProducingOutput { table },
+ )
+ }
+ }
+ }
+ }
+ }
+
+ /// Sorts and spills one complete in-memory state run, then resumes input.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_spilling(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::Spilling {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected spilling state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let mut result = spill_context.spill_table(&mut table);
+
+ // Spilling shrinks the aggregate table and releases its accumulated
+ // memory. Update the reservation accordingly.
+ if self.reservation.try_resize(table.memory_size()).is_err() {
+ // Fold spill error and reservation error into one
+ result = internal_err!("Reservation after spilling should succeed")
+ }
+
+ timer.done();
+
+ match result {
+ // Finished spilling the aggregate table, continue aggregating from input
+ Ok(true) => ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context: Some(spill_context),
+ }),
+ Ok(false) => ControlFlow::Break((
+ Poll::Ready(Some(internal_err!(
+ "Ordered final aggregation entered Spilling with an empty table"
+ ))),
+ OrderedFinalAggregateState::Done,
+ )),
+ Err(e) => ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ )),
+ }
+ }
+
+ /// 1. Spills the last in-memory run.
+ /// 2. Constructs a globally ordered input stream by applying a sort-preserving
+ /// merge to all spills.
+ /// 3. Constructs a replay stream: an ordered aggregate stream over the fully
+ /// ordered input constructed from the spills.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_preparing_merge_input(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::PreparingMergeInput {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected preparing merge input state")
Review Comment:
it could be more defensive and return an Err here (an make the function return Result<..>) as a wa to avoid potential future refactors introducing panics
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -86,6 +270,35 @@ impl OrderedFinalAggregateStream {
partition: usize,
input: SendableRecordBatchStream,
input_order_mode: &InputOrderMode,
+ ) -> Result<Self> {
+ let baseline_metrics = BaselineMetrics::new(&agg.metrics, partition);
+ let group_by_metrics = GroupByMetrics::new(&agg.metrics, partition);
+ let spill_metrics = SpillMetrics::new(&agg.metrics, partition);
+ Self::new_with_input_and_metrics(
+ agg,
+ context,
+ partition,
+ input,
+ input_order_mode,
+ baseline_metrics,
+ group_by_metrics,
+ Some(spill_metrics),
+ )
+ }
+
+ #[expect(
Review Comment:
maybe eventually it would be good to make a bulder or something ? That is probably too much code for now though
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -64,6 +119,135 @@ type OrderedFinalAggregateStateTransition = ControlFlow<
OrderedFinalAggregateState,
>;
+impl OrderedFinalSpillContext {
+ fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ batch_size: usize,
+ input_order_mode: &InputOrderMode,
+ spill_schema: &SchemaRef,
+ spill_metrics: SpillMetrics,
+ ) -> Result<Self> {
+ let group_schema = agg.group_by.group_schema(spill_schema)?;
+ let output_ordering = agg.cache.output_ordering();
+ let InputOrderMode::PartiallySorted(order_indices) = input_order_mode else {
+ return internal_err!("Ordered final spill requires partially ordered input");
+ };
+ let spill_indices = order_indices.iter().copied().chain(
+ (0..group_schema.fields().len()).filter(|idx| !order_indices.contains(idx)),
+ );
+ let spill_sort_exprs = spill_indices.map(|idx| {
+ let field = group_schema.field(idx);
+ let output_expr = Column::new(field.name(), idx);
+ let sort_options = output_ordering
+ .and_then(|ordering| ordering.get_sort_options(&output_expr))
+ .unwrap_or_default();
+ PhysicalSortExpr::new(Arc::new(output_expr), sort_options)
+ });
+ let Some(spill_expr) = LexOrdering::new(spill_sort_exprs) else {
+ return internal_err!("Ordered final spill expression is empty");
+ };
+
+ let spill_manager = SpillManager::new(
+ context.runtime_env(),
+ spill_metrics,
+ Arc::clone(spill_schema),
+ )
+ .with_compression_type(context.session_config().spill_compression());
+
+ Ok(Self {
+ agg: agg.clone(),
+ context: Arc::clone(context),
+ partition,
+ batch_size,
+ spill_expr,
+ spill_manager,
+ spills: vec![],
+ })
+ }
+
+ fn has_spills(&self) -> bool {
+ !self.spills.is_empty()
+ }
+
+ /// Sort and spill the aggregated groups. Memory reservation should be cleared
+ /// by the caller of this function.
+ ///
+ /// See [`OrderedFinalAggregateStream`] for spilling details.
+ ///
+ /// It must not be called without rows to spill in the aggregate table, otherwise
Review Comment:
Could you also document what the return value means? Is it "table was empty"? If it is an error, why not just return the error rather than also a boolean?
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
}
}
Poll::Ready(Some(Err(e))) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
Poll::Ready(None) => {
self.close_input();
- table.input_done();
- ControlFlow::Continue(OrderedFinalAggregateState::DrainingFinal { table })
+ match spill_context {
+ Some(spill_context) if spill_context.has_spills() => {
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::PreparingMergeInput {
+ table,
+ spill_context,
+ },
+ )
+ }
+ _ => {
+ table.input_done();
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::ProducingOutput { table },
+ )
+ }
+ }
+ }
+ }
+ }
+
+ /// Sorts and spills one complete in-memory state run, then resumes input.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_spilling(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::Spilling {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected spilling state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let mut result = spill_context.spill_table(&mut table);
+
+ // Spilling shrinks the aggregate table and releases its accumulated
+ // memory. Update the reservation accordingly.
+ if self.reservation.try_resize(table.memory_size()).is_err() {
+ // Fold spill error and reservation error into one
+ result = internal_err!("Reservation after spilling should succeed")
Review Comment:
I think this can happen if other plans are allocating from the memory manager at the same time (e.g. the aggregate releases the memory but then something in another plan reserves it)
In other words, I think we should just propagte the "resources exhausted" error here
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
Review Comment:
Rather than putting the state back into ReadingInput, should we move it to the terminal state (Done) instead? That would could potentially release the intermediate state quickly on error rather than waiting for dropping the plan
I realize this PR follows the existing pattern, but it occured to me while reviewing
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -157,21 +409,86 @@ impl OrderedFinalAggregateStream {
if let Err(e) = result {
return ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
));
}
+ // Check memory reservation, and potentially spill.
let timer = elapsed_compute.timer();
- let result = table.next_output_batch();
+ let resize_result =
+ self.reservation
+ .try_resize(Self::reservation_size_for_table(
+ &table,
+ spill_context.as_deref(),
+ ));
timer.done();
+ match resize_result {
+ Ok(()) => {}
+ Err(e @ DataFusionError::ResourcesExhausted(_)) => {
+ let Some(spill_context) = spill_context else {
+ return ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ ));
+ };
+ if table.is_empty() {
+ return ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ ));
+ }
+ return ControlFlow::Continue(
+ OrderedFinalAggregateState::Spilling {
+ table,
+ spill_context,
+ },
+ );
+ }
+ Err(e) => {
+ return ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ ));
+ }
+ }
+
+ let result = if spill_context
+ .as_ref()
+ .is_some_and(|spill_context| spill_context.has_spills())
+ {
+ // Once one incomplete run is spilled, every remaining state
+ // must participate in replay so no group is finalized twice.
+ Ok(None)
+ } else {
+ let timer = elapsed_compute.timer();
+ let result = table.next_output_batch();
+ timer.done();
+ result
+ };
match result {
// Some finalized groups can be emitted. Yield them, then
// continue aggregating input in the current state.
Ok(Some(batch)) => {
- let next_state =
- OrderedFinalAggregateState::ReadingInput { table };
- self.resize_reservation_for_state(&next_state);
+ if let Err(e) =
Review Comment:
This pattern (of handing the "can't reserve what is needed") is repeated a few times and I think contributes to the length of the `poll_inner` here. I wonder if there is some way to factor it out into a new function to make it easier to read / keep it encapsulated π€
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3626983420
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
}
}
Poll::Ready(Some(Err(e))) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
Poll::Ready(None) => {
self.close_input();
- table.input_done();
- ControlFlow::Continue(OrderedFinalAggregateState::DrainingFinal { table })
+ match spill_context {
+ Some(spill_context) if spill_context.has_spills() => {
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::PreparingMergeInput {
+ table,
+ spill_context,
+ },
+ )
+ }
+ _ => {
+ table.input_done();
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::ProducingOutput { table },
+ )
+ }
+ }
+ }
+ }
+ }
+
+ /// Sorts and spills one complete in-memory state run, then resumes input.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_spilling(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::Spilling {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected spilling state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let mut result = spill_context.spill_table(&mut table);
+
+ // Spilling shrinks the aggregate table and releases its accumulated
+ // memory. Update the reservation accordingly.
+ if self.reservation.try_resize(table.memory_size()).is_err() {
+ // Fold spill error and reservation error into one
+ result = internal_err!("Reservation after spilling should succeed")
+ }
+
+ timer.done();
+
+ match result {
+ // Finished spilling the aggregate table, continue aggregating from input
+ Ok(true) => ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context: Some(spill_context),
+ }),
+ Ok(false) => ControlFlow::Break((
+ Poll::Ready(Some(internal_err!(
+ "Ordered final aggregation entered Spilling with an empty table"
+ ))),
+ OrderedFinalAggregateState::Done,
+ )),
+ Err(e) => ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ )),
+ }
+ }
+
+ /// 1. Spills the last in-memory run.
+ /// 2. Constructs a globally ordered input stream by applying a sort-preserving
+ /// merge to all spills.
+ /// 3. Constructs a replay stream: an ordered aggregate stream over the fully
+ /// ordered input constructed from the spills.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_preparing_merge_input(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::PreparingMergeInput {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected preparing merge input state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let replay = match spill_context.spill_table(&mut table) {
+ Ok(_) => {
+ let group_by_metrics = table.group_by_metrics();
+ drop(table);
+ match self.reservation.try_resize(0) {
+ Ok(()) => (*spill_context).into_replay_stream(
+ &self.baseline_metrics,
+ group_by_metrics,
+ self.reservation.new_empty(),
+ ),
+ Err(e) => Err(e),
+ }
+ }
+ Err(e) => Err(e),
+ };
+ timer.done();
+
+ match replay {
+ Ok(stream) => {
+ ControlFlow::Continue(OrderedFinalAggregateState::MergingSpills {
+ stream,
+ })
}
+ Err(e) => ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ )),
+ }
+ }
+
+ /// Forwards output from the fully ordered stream that consumes the merged
+ /// spill runs.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_merging_spills(
+ &mut self,
+ cx: &mut Context<'_>,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::MergingSpills { mut stream } = original_state
+ else {
+ unreachable!("expected merging spills state")
Review Comment:
ditto here about internal error vs panic
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -64,6 +119,135 @@ type OrderedFinalAggregateStateTransition = ControlFlow<
OrderedFinalAggregateState,
>;
+impl OrderedFinalSpillContext {
+ fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ batch_size: usize,
+ input_order_mode: &InputOrderMode,
+ spill_schema: &SchemaRef,
+ spill_metrics: SpillMetrics,
+ ) -> Result<Self> {
+ let group_schema = agg.group_by.group_schema(spill_schema)?;
+ let output_ordering = agg.cache.output_ordering();
+ let InputOrderMode::PartiallySorted(order_indices) = input_order_mode else {
+ return internal_err!("Ordered final spill requires partially ordered input");
+ };
+ let spill_indices = order_indices.iter().copied().chain(
+ (0..group_schema.fields().len()).filter(|idx| !order_indices.contains(idx)),
+ );
+ let spill_sort_exprs = spill_indices.map(|idx| {
+ let field = group_schema.field(idx);
+ let output_expr = Column::new(field.name(), idx);
+ let sort_options = output_ordering
+ .and_then(|ordering| ordering.get_sort_options(&output_expr))
+ .unwrap_or_default();
+ PhysicalSortExpr::new(Arc::new(output_expr), sort_options)
+ });
+ let Some(spill_expr) = LexOrdering::new(spill_sort_exprs) else {
+ return internal_err!("Ordered final spill expression is empty");
+ };
+
+ let spill_manager = SpillManager::new(
+ context.runtime_env(),
+ spill_metrics,
+ Arc::clone(spill_schema),
+ )
+ .with_compression_type(context.session_config().spill_compression());
+
+ Ok(Self {
+ agg: agg.clone(),
+ context: Arc::clone(context),
+ partition,
+ batch_size,
+ spill_expr,
+ spill_manager,
+ spills: vec![],
+ })
+ }
+
+ fn has_spills(&self) -> bool {
+ !self.spills.is_empty()
+ }
+
+ /// Sort and spill the aggregated groups. Memory reservation should be cleared
Review Comment:
it might also help to explain somewhere what sort key the spill files are sorted by (is it the partial sort prefix, or the full group key)? I think it is the full group key.
##########
datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs:
##########
@@ -69,14 +70,50 @@ use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream, metric
/// `k = 100`, it is safe to emit all groups with keys less than 100 because the
/// input is ordered.
///
+/// # Memory Pressure and Spilling
+///
+/// ## Fully ordered case
+///
+/// If the input is ordered by every group key, for example:
+///
+/// - Input order: `a, b`
+/// - `GROUP BY`: `a, b`
+///
+/// Completed groups can be emitted as soon as the next group is observed. Thus,
+/// only the current group remains active after completed groups are emitted, and
+/// memory usage does not grow with the total number of groups.
+///
+/// If a memory reservation nevertheless fails, the stream returns the error
+/// directly, indicating an unexpected behavior.
+///
+/// ## Partially ordered case
+///
+/// If the input is ordered by only a subset of the group keys, for example:
+///
+/// - Input order: `a`
+/// - `GROUP BY`: `a, b`
+///
+/// If one `a` value contains many distinct `b` values, the table may accumulate
+/// enough groups to exceed the memory limit.
+///
+/// - `OrderedPartialAggregateStream`: On reservation failure, it emits all current
+/// intermediate states downstream and resets the table. The final stage can
+/// merge repeated `(a, b)` state rows, so no disk spill is required.
+/// - `OrderedFinalAggregateStream`: It cannot emit incomplete final results. On
+/// reservation failure, it sorts the current intermediate states by the complete
+/// group key and spills them as one run. After the input ends, it spills any
+/// remaining states, performs a sort-preserving merge of all runs, and feeds the
+/// merged input into a fully ordered final aggregate stream.
+///
/// ## Implementation Note
///
/// This is intentionally kept simple and closely maps to
/// `GroupedHashAggregateStream` to finish the refactor sooner.
///
+/// More applicable optimizations are left to future work.
Review Comment:
this is somwhat mysterious π maybe we can link to a ticket to track such potential future optimizations (or just remove this comment)
##########
datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt:
##########
@@ -0,0 +1,201 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# End-to-end tests for ordered aggregation under finite memory.
+
+# Result set more than 100 lines will be hashed
+hash-threshold 100
+
+statement ok
+SET datafusion.execution.target_partitions = 2
+
+statement ok
+SET datafusion.execution.batch_size = 128
+
+statement ok
+SET datafusion.optimizer.repartition_aggregations = true
+
+statement ok
+SET datafusion.optimizer.prefer_existing_sort = true
+
+statement ok
+SET datafusion.execution.enable_migration_aggregate = true
Review Comment:
This is already the default, right?
##########
datafusion/physical-plan/src/aggregates/ordered_partial_stream.rs:
##########
@@ -69,14 +70,50 @@ use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream, metric
/// `k = 100`, it is safe to emit all groups with keys less than 100 because the
/// input is ordered.
///
+/// # Memory Pressure and Spilling
+///
+/// ## Fully ordered case
+///
+/// If the input is ordered by every group key, for example:
+///
+/// - Input order: `a, b`
+/// - `GROUP BY`: `a, b`
+///
+/// Completed groups can be emitted as soon as the next group is observed. Thus,
+/// only the current group remains active after completed groups are emitted, and
+/// memory usage does not grow with the total number of groups.
+///
+/// If a memory reservation nevertheless fails, the stream returns the error
+/// directly, indicating an unexpected behavior.
+///
+/// ## Partially ordered case
+///
+/// If the input is ordered by only a subset of the group keys, for example:
+///
+/// - Input order: `a`
+/// - `GROUP BY`: `a, b`
+///
+/// If one `a` value contains many distinct `b` values, the table may accumulate
+/// enough groups to exceed the memory limit.
+///
+/// - `OrderedPartialAggregateStream`: On reservation failure, it emits all current
Review Comment:
this is a great description
##########
datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt:
##########
@@ -0,0 +1,201 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
I actually ran just this test under coverage
```shell
cargo llvm-cov --html --test sqllogictests -- ordered_aggregate_spill.slt
```
And it has quite impressive coverage π
<img width="1536" height="271" alt="Image" src="https://github.com/user-attachments/assets/5b0dcee7-4995-4fef-bbd7-1b9be5b906a4" />
##########
datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt:
##########
@@ -0,0 +1,201 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# End-to-end tests for ordered aggregation under finite memory.
+
+# Result set more than 100 lines will be hashed
+hash-threshold 100
+
+statement ok
+SET datafusion.execution.target_partitions = 2
+
+statement ok
+SET datafusion.execution.batch_size = 128
+
+statement ok
+SET datafusion.optimizer.repartition_aggregations = true
+
+statement ok
+SET datafusion.optimizer.prefer_existing_sort = true
+
+statement ok
+SET datafusion.execution.enable_migration_aggregate = true
+
+statement ok
+SET datafusion.runtime.memory_limit = '1M'
+
+# ==================================================================================
+# Input is fully ordered by group keys (input order by (a,b), query is 'group by a,b')
+# ==================================================================================
+
+# Fully ordered input uses the ordered partial and final streams without spill.
+query TT
+EXPLAIN ANALYZE
+SELECT v1, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY v1
+----
+Plan with Metrics
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>ordering_mode=Sorted, metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+02)--RepartitionExec:<slt:ignore>preserve_order=true<slt:ignore>
+03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=Sorted, metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+<slt:ignore>
+
+query II rowsort
+SELECT v1, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY v1
+----
+40002 values hashing to 34c2b23730596cbd2489ed4a627c17d7
+
+# The same fully ordered query cannot spill and reports OOM under tighter memory.
+statement ok
+SET datafusion.runtime.memory_limit = '1K'
+
+query error Resources exhausted
+SELECT v1, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY v1
+
+# ==================================================================================
+# Input is partially ordered by group keys (input order by (a), query is 'group by a,b')
+#
+# Try different memory limits, ensure result is the same, but spill count differ
+
+# HACK: check `spilled_bytes=x KB` to ensure it has spilled. If it has not spilled,
+# the it shows `spilled_bytes = 0B`. Should better check spill_count, but it's not
+# stable due to ordered hash repartition, and `sqllogictest` don't support regex.
+# ==================================================================================
+
+statement ok
+SET datafusion.runtime.memory_limit = '2M'
+
+statement ok
+SET datafusion.optimizer.enable_round_robin_repartition = false
+
+# Round 1: The input is partially ordered and does not spill with a 2 MB limit.
+query TT
+EXPLAIN ANALYZE
+SELECT round(v1, -4), v1 % 5000, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY round(v1, -4), v1 % 5000
+----
+Plan with Metrics
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+02)--RepartitionExec:<slt:ignore>input_partitions=1, maintains_sort_order=true<slt:ignore>
+03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+<slt:ignore>
+
+# All rounds should have the same result hash
+query III rowsort
+SELECT round(v1, -4), v1 % 5000, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY round(v1, -4), v1 % 5000
+----
+45000 values hashing to e6ece4b4b86e6152a1c785e90ab5ba12
+
+# Round 2: The same query spills five times with a 600 KB limit.
+statement ok
+SET datafusion.runtime.memory_limit = '600K'
+
+query TT
+EXPLAIN ANALYZE
+SELECT round(v1, -4), v1 % 5000, sum(v1 * 2)
+FROM generate_series(20000) AS t1(v1)
+GROUP BY round(v1, -4), v1 % 5000
+----
+Plan with Metrics
+01)AggregateExec: mode=FinalPartitioned,<slt:ignore>aggr=[sum(t1.v1 * Int64(2))], ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spilled_bytes=<slt:ignore> KB,<slt:ignore>]
+02)--RepartitionExec:<slt:ignore>input_partitions=1, maintains_sort_order=true<slt:ignore>
+03)----AggregateExec: mode=Partial,<slt:ignore>ordering_mode=PartiallySorted([0]), metrics=[<slt:ignore>spill_count=0,<slt:ignore>]
+<slt:ignore>
+
+# All rounds should have the same result hash
Review Comment:
it would be nice if there was some way to programatically ensure the hashses are the same, but I think the comments you have inline here are good enough
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -47,13 +69,46 @@ pub(crate) struct OrderedFinalAggregateStream {
state: Option<OrderedFinalAggregateState>,
}
+/// Spill configuration and accumulated runs for partially ordered final
+/// aggregation. Each file is one fully group-key-sorted intermediate-state run;
+/// all runs are replayed together after the original input is exhausted.
+struct OrderedFinalSpillContext {
+ /// Aggregate configuration
+ agg: AggregateExec,
+ /// Task context
+ context: Arc<TaskContext>,
+ /// Original partition index
+ partition: usize,
+ /// Target batch size from configuration
+ batch_size: usize,
+ /// Full group-key ordering, such ordering with be kept in: a) individual spill
+ /// files, b) order after final merging and streaming aggregate
+ spill_expr: LexOrdering,
+ /// Spill I/O and metrics manager.
+ spill_manager: SpillManager,
+ /// Fully sorted spill runs waiting to be merged.
+ spills: Vec<SortedSpillFile>,
+}
+
/// See comments at `poll_next()` for details.
enum OrderedFinalAggregateState {
ReadingInput {
table: OrderedAggregateTable<FinalMarker>,
+ spill_context: Option<Box<OrderedFinalSpillContext>>,
},
- DrainingFinal {
+ Spilling {
table: OrderedAggregateTable<FinalMarker>,
+ spill_context: Box<OrderedFinalSpillContext>,
+ },
+ ProducingOutput {
+ table: OrderedAggregateTable<FinalMarker>,
+ },
+ PreparingMergeInput {
Review Comment:
(I see now it is documted below at `impl Stream for OrderedFinalAggregateStream {` -- maybe we could leave a link / hit in the comments to look there for the state transitions
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5041243672
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/23657#issuecomment-5041227943)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c5041227943-1215-t48hj 6.12.85+ #1 SMP Mon May 11 08:17:35 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing split-aggr-spill-order-aggr (a780de8773903645de7432bb6372c42ff8e549f4) to f151c10 (merge-base) [diff](https://github.com/apache/datafusion/compare/f151c10fdee89bd6f42c0f786988bd09557ed0fd..a780de8773903645de7432bb6372c42ff8e549f4) using: tpch
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 merged PR #23657:
URL: https://github.com/apache/datafusion/pull/23657
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3637198733
##########
datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs:
##########
@@ -234,6 +250,43 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
+ self.buffer.group_indices.allocated_size()
}
+ pub(in crate::aggregates) fn group_by_metrics(&self) -> GroupByMetrics {
+ self.group_by_metrics.clone()
+ }
+
+ /// Takes every intermediate aggregate state and resets the table so it can
+ /// continue with a new ordered input segment.
+ ///
+ /// Unlike normal ordered emission, this operation is allowed to take the
+ /// active (incomplete) groups. Partial aggregation can pass those states to
+ /// its final stage, while final aggregation sorts and spills them before
+ /// replay.
+ pub(in crate::aggregates) fn take_state_batch(
+ &mut self,
+ ) -> Result<Option<RecordBatch>> {
+ if self.buffer.group_values.is_empty() {
+ return Ok(None);
+ }
+
+ let mut output = self.buffer.group_values.emit(EmitTo::All)?;
+ for acc in &mut self.buffer.accumulators {
+ output.extend(acc.state(EmitTo::All)?);
+ }
+
+ let batch = RecordBatch::try_new(Arc::clone(&self.state_schema), output)?;
+ debug_assert!(batch.num_rows() > 0);
+
+ // `emit(EmitTo::All)` resets accumulator state. Explicitly shrink the
Review Comment:
I think since they're not reused elsewhere, probably its clearer to inline them here.
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3638083928
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
}
}
Poll::Ready(Some(Err(e))) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
Poll::Ready(None) => {
self.close_input();
- table.input_done();
- ControlFlow::Continue(OrderedFinalAggregateState::DrainingFinal { table })
+ match spill_context {
+ Some(spill_context) if spill_context.has_spills() => {
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::PreparingMergeInput {
+ table,
+ spill_context,
+ },
+ )
+ }
+ _ => {
+ table.input_done();
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::ProducingOutput { table },
+ )
+ }
+ }
+ }
+ }
+ }
+
+ /// Sorts and spills one complete in-memory state run, then resumes input.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_spilling(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::Spilling {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected spilling state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let mut result = spill_context.spill_table(&mut table);
+
+ // Spilling shrinks the aggregate table and releases its accumulated
+ // memory. Update the reservation accordingly.
+ if self.reservation.try_resize(table.memory_size()).is_err() {
+ // Fold spill error and reservation error into one
+ result = internal_err!("Reservation after spilling should succeed")
+ }
+
+ timer.done();
+
+ match result {
+ // Finished spilling the aggregate table, continue aggregating from input
+ Ok(true) => ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context: Some(spill_context),
+ }),
+ Ok(false) => ControlFlow::Break((
+ Poll::Ready(Some(internal_err!(
+ "Ordered final aggregation entered Spilling with an empty table"
+ ))),
+ OrderedFinalAggregateState::Done,
+ )),
+ Err(e) => ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ )),
+ }
+ }
+
+ /// 1. Spills the last in-memory run.
+ /// 2. Constructs a globally ordered input stream by applying a sort-preserving
+ /// merge to all spills.
+ /// 3. Constructs a replay stream: an ordered aggregate stream over the fully
+ /// ordered input constructed from the spills.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_preparing_merge_input(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::PreparingMergeInput {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected preparing merge input state")
Review Comment:
[7f8e4be](https://github.com/apache/datafusion/pull/23657/commits/7f8e4be4d0da987badb10237e51e974e5f9f4238)
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
}
}
Poll::Ready(Some(Err(e))) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
Poll::Ready(None) => {
self.close_input();
- table.input_done();
- ControlFlow::Continue(OrderedFinalAggregateState::DrainingFinal { table })
+ match spill_context {
+ Some(spill_context) if spill_context.has_spills() => {
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::PreparingMergeInput {
+ table,
+ spill_context,
+ },
+ )
+ }
+ _ => {
+ table.input_done();
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::ProducingOutput { table },
+ )
+ }
+ }
+ }
+ }
+ }
+
+ /// Sorts and spills one complete in-memory state run, then resumes input.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_spilling(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::Spilling {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected spilling state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let mut result = spill_context.spill_table(&mut table);
+
+ // Spilling shrinks the aggregate table and releases its accumulated
+ // memory. Update the reservation accordingly.
+ if self.reservation.try_resize(table.memory_size()).is_err() {
+ // Fold spill error and reservation error into one
+ result = internal_err!("Reservation after spilling should succeed")
+ }
+
+ timer.done();
+
+ match result {
+ // Finished spilling the aggregate table, continue aggregating from input
+ Ok(true) => ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context: Some(spill_context),
+ }),
+ Ok(false) => ControlFlow::Break((
+ Poll::Ready(Some(internal_err!(
+ "Ordered final aggregation entered Spilling with an empty table"
+ ))),
+ OrderedFinalAggregateState::Done,
+ )),
+ Err(e) => ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ )),
+ }
+ }
+
+ /// 1. Spills the last in-memory run.
+ /// 2. Constructs a globally ordered input stream by applying a sort-preserving
+ /// merge to all spills.
+ /// 3. Constructs a replay stream: an ordered aggregate stream over the fully
+ /// ordered input constructed from the spills.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_preparing_merge_input(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::PreparingMergeInput {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected preparing merge input state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let replay = match spill_context.spill_table(&mut table) {
+ Ok(_) => {
+ let group_by_metrics = table.group_by_metrics();
+ drop(table);
+ match self.reservation.try_resize(0) {
+ Ok(()) => (*spill_context).into_replay_stream(
+ &self.baseline_metrics,
+ group_by_metrics,
+ self.reservation.new_empty(),
+ ),
+ Err(e) => Err(e),
+ }
+ }
+ Err(e) => Err(e),
+ };
+ timer.done();
+
+ match replay {
+ Ok(stream) => {
+ ControlFlow::Continue(OrderedFinalAggregateState::MergingSpills {
+ stream,
+ })
}
+ Err(e) => ControlFlow::Break((
+ Poll::Ready(Some(Err(e))),
+ OrderedFinalAggregateState::Done,
+ )),
+ }
+ }
+
+ /// Forwards output from the fully ordered stream that consumes the merged
+ /// spill runs.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_merging_spills(
+ &mut self,
+ cx: &mut Context<'_>,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::MergingSpills { mut stream } = original_state
+ else {
+ unreachable!("expected merging spills state")
Review Comment:
[7f8e4be](https://github.com/apache/datafusion/pull/23657/commits/7f8e4be4d0da987badb10237e51e974e5f9f4238)
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5041337768
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/23657#issuecomment-5041227943)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
Comparing HEAD and split-aggr-spill-order-aggr
--------------------
Benchmark tpcds_sf1.json
--------------------
βββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β split-aggr-spill-order-aggr β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 5.50 / 6.03 Β±0.91 / 7.84 ms β 5.52 / 6.04 Β±0.92 / 7.88 ms β no change β
β QQuery 2 β 82.60 / 83.02 Β±0.28 / 83.43 ms β 81.66 / 82.34 Β±0.44 / 82.91 ms β no change β
β QQuery 3 β 29.68 / 30.07 Β±0.28 / 30.49 ms β 30.10 / 30.25 Β±0.12 / 30.45 ms β no change β
β QQuery 4 β 483.51 / 487.22 Β±2.96 / 491.49 ms β 483.90 / 489.15 Β±4.41 / 494.59 ms β no change β
β QQuery 5 β 52.57 / 52.93 Β±0.30 / 53.39 ms β 52.51 / 52.89 Β±0.27 / 53.12 ms β no change β
β QQuery 6 β 36.48 / 36.94 Β±0.42 / 37.59 ms β 36.86 / 37.57 Β±0.40 / 38.01 ms β no change β
β QQuery 7 β 95.64 / 96.43 Β±0.51 / 97.18 ms β 95.72 / 96.23 Β±0.36 / 96.83 ms β no change β
β QQuery 8 β 37.32 / 39.42 Β±3.53 / 46.47 ms β 37.90 / 39.57 Β±1.93 / 43.14 ms β no change β
β QQuery 9 β 54.80 / 56.56 Β±1.35 / 58.85 ms β 53.38 / 55.56 Β±1.26 / 56.99 ms β no change β
β QQuery 10 β 64.18 / 64.63 Β±0.44 / 65.35 ms β 64.09 / 64.68 Β±0.36 / 65.08 ms β no change β
β QQuery 11 β 305.70 / 308.44 Β±1.77 / 310.46 ms β 301.46 / 307.39 Β±5.06 / 316.03 ms β no change β
β QQuery 12 β 29.46 / 29.71 Β±0.22 / 30.09 ms β 29.07 / 29.55 Β±0.26 / 29.77 ms β no change β
β QQuery 13 β 120.92 / 121.68 Β±0.49 / 122.45 ms β 119.99 / 120.19 Β±0.15 / 120.41 ms β no change β
β QQuery 14 β 415.38 / 420.96 Β±5.57 / 428.05 ms β 418.22 / 423.29 Β±4.39 / 431.43 ms β no change β
β QQuery 15 β 57.64 / 58.27 Β±0.39 / 58.76 ms β 58.88 / 59.39 Β±0.26 / 59.65 ms β no change β
β QQuery 16 β 6.65 / 6.78 Β±0.19 / 7.15 ms β 6.70 / 6.83 Β±0.21 / 7.24 ms β no change β
β QQuery 17 β 79.62 / 80.94 Β±0.85 / 82.19 ms β 80.81 / 81.85 Β±1.38 / 84.58 ms β no change β
β QQuery 18 β 124.99 / 126.68 Β±1.61 / 129.18 ms β 125.57 / 126.99 Β±1.22 / 128.83 ms β no change β
β QQuery 19 β 42.07 / 42.40 Β±0.25 / 42.72 ms β 42.72 / 42.96 Β±0.17 / 43.20 ms β no change β
β QQuery 20 β 36.02 / 37.33 Β±1.13 / 39.30 ms β 36.73 / 37.94 Β±0.90 / 39.09 ms β no change β
β QQuery 21 β 17.89 / 18.16 Β±0.21 / 18.40 ms β 17.81 / 18.46 Β±0.64 / 19.68 ms β no change β
β QQuery 22 β 62.95 / 63.48 Β±0.74 / 64.91 ms β 63.34 / 64.16 Β±0.64 / 65.20 ms β no change β
β QQuery 23 β 345.88 / 349.18 Β±2.25 / 351.79 ms β 347.96 / 351.19 Β±2.57 / 354.83 ms β no change β
β QQuery 24 β 224.86 / 227.27 Β±2.36 / 230.73 ms β 224.96 / 228.36 Β±3.30 / 234.44 ms β no change β
β QQuery 25 β 110.62 / 111.42 Β±1.15 / 113.67 ms β 111.15 / 111.90 Β±0.48 / 112.37 ms β no change β
β QQuery 26 β 58.44 / 61.47 Β±3.24 / 67.46 ms β 58.23 / 59.41 Β±1.18 / 61.65 ms β no change β
β QQuery 27 β 6.34 / 6.47 Β±0.20 / 6.86 ms β 6.25 / 6.37 Β±0.12 / 6.59 ms β no change β
β QQuery 28 β 61.96 / 62.85 Β±0.62 / 63.46 ms β 62.16 / 63.02 Β±0.53 / 63.61 ms β no change β
β QQuery 29 β 97.28 / 99.05 Β±2.23 / 103.10 ms β 98.20 / 100.62 Β±1.96 / 103.45 ms β no change β
β QQuery 30 β 33.62 / 33.90 Β±0.38 / 34.63 ms β 33.13 / 33.92 Β±0.97 / 35.82 ms β no change β
β QQuery 31 β 112.92 / 113.26 Β±0.29 / 113.62 ms β 113.35 / 114.06 Β±0.55 / 114.62 ms β no change β
β QQuery 32 β 20.08 / 20.51 Β±0.35 / 21.12 ms β 20.49 / 21.07 Β±0.35 / 21.46 ms β no change β
β QQuery 33 β 38.26 / 38.72 Β±0.51 / 39.61 ms β 39.04 / 40.68 Β±2.82 / 46.30 ms β 1.05x slower β
β QQuery 34 β 10.01 / 10.23 Β±0.36 / 10.94 ms β 10.16 / 10.76 Β±0.84 / 12.42 ms β 1.05x slower β
β QQuery 35 β 73.42 / 73.86 Β±0.51 / 74.81 ms β 73.17 / 74.74 Β±0.79 / 75.31 ms β no change β
β QQuery 36 β 5.83 / 5.96 Β±0.21 / 6.37 ms β 5.82 / 5.95 Β±0.18 / 6.30 ms β no change β
β QQuery 37 β 6.89 / 6.97 Β±0.07 / 7.10 ms β 6.96 / 7.05 Β±0.07 / 7.13 ms β no change β
β QQuery 38 β 62.17 / 62.95 Β±0.61 / 63.94 ms β 62.83 / 63.08 Β±0.20 / 63.26 ms β no change β
β QQuery 39 β 90.22 / 91.32 Β±0.86 / 92.85 ms β 91.44 / 93.87 Β±2.95 / 99.38 ms β no change β
β QQuery 40 β 23.40 / 23.78 Β±0.41 / 24.52 ms β 23.87 / 24.43 Β±0.30 / 24.71 ms β no change β
β QQuery 41 β 11.69 / 11.82 Β±0.15 / 12.11 ms β 11.68 / 11.90 Β±0.20 / 12.22 ms β no change β
β QQuery 42 β 23.86 / 24.33 Β±0.42 / 24.84 ms β 24.44 / 25.11 Β±0.54 / 26.00 ms β no change β
β QQuery 43 β 4.97 / 5.10 Β±0.14 / 5.35 ms β 5.08 / 5.17 Β±0.13 / 5.42 ms β no change β
β QQuery 44 β 9.45 / 9.58 Β±0.08 / 9.69 ms β 9.53 / 9.70 Β±0.14 / 9.87 ms β no change β
β QQuery 45 β 38.66 / 38.85 Β±0.16 / 39.12 ms β 38.47 / 39.26 Β±0.48 / 39.76 ms β no change β
β QQuery 46 β 11.88 / 12.02 Β±0.11 / 12.21 ms β 11.96 / 13.38 Β±2.02 / 17.37 ms β 1.11x slower β
β QQuery 47 β 227.75 / 229.21 Β±2.10 / 233.36 ms β 226.28 / 233.34 Β±5.60 / 241.91 ms β no change β
β QQuery 48 β 97.31 / 97.86 Β±0.33 / 98.19 ms β 97.50 / 97.82 Β±0.23 / 98.17 ms β no change β
β QQuery 49 β 77.88 / 79.76 Β±3.37 / 86.49 ms β 76.50 / 78.50 Β±2.48 / 83.26 ms β no change β
β QQuery 50 β 59.84 / 60.63 Β±0.64 / 61.37 ms β 59.31 / 60.35 Β±0.94 / 61.91 ms β no change β
β QQuery 51 β 93.16 / 94.97 Β±2.11 / 99.04 ms β 92.06 / 93.92 Β±1.52 / 95.96 ms β no change β
β QQuery 52 β 24.38 / 24.70 Β±0.35 / 25.34 ms β 24.45 / 24.67 Β±0.18 / 24.94 ms β no change β
β QQuery 53 β 29.85 / 32.14 Β±3.96 / 40.01 ms β 30.26 / 30.81 Β±0.48 / 31.68 ms β no change β
β QQuery 54 β 56.27 / 57.07 Β±0.49 / 57.73 ms β 56.28 / 57.09 Β±0.92 / 58.56 ms β no change β
β QQuery 55 β 23.38 / 23.92 Β±0.37 / 24.53 ms β 23.91 / 24.22 Β±0.35 / 24.91 ms β no change β
β QQuery 56 β 39.13 / 39.72 Β±0.44 / 40.38 ms β 39.89 / 40.08 Β±0.16 / 40.29 ms β no change β
β QQuery 57 β 178.60 / 179.81 Β±1.27 / 182.17 ms β 177.21 / 180.59 Β±2.61 / 185.19 ms β no change β
β QQuery 58 β 116.13 / 117.16 Β±0.79 / 118.39 ms β 117.50 / 119.58 Β±2.17 / 123.45 ms β no change β
β QQuery 59 β 119.88 / 120.55 Β±0.68 / 121.78 ms β 119.93 / 120.62 Β±0.57 / 121.48 ms β no change β
β QQuery 60 β 39.81 / 40.33 Β±0.38 / 40.78 ms β 39.95 / 40.71 Β±0.60 / 41.52 ms β no change β
β QQuery 61 β 12.26 / 12.48 Β±0.18 / 12.80 ms β 12.17 / 12.35 Β±0.15 / 12.54 ms β no change β
β QQuery 62 β 46.92 / 47.54 Β±0.64 / 48.77 ms β 46.72 / 47.30 Β±0.86 / 49.02 ms β no change β
β QQuery 63 β 30.17 / 30.55 Β±0.38 / 31.21 ms β 30.22 / 30.40 Β±0.16 / 30.67 ms β no change β
β QQuery 64 β 409.72 / 415.72 Β±3.73 / 421.35 ms β 411.54 / 413.31 Β±1.96 / 416.26 ms β no change β
β QQuery 65 β 128.04 / 129.93 Β±1.81 / 132.34 ms β 126.18 / 128.17 Β±1.37 / 130.14 ms β no change β
β QQuery 66 β 80.61 / 81.30 Β±0.53 / 82.18 ms β 80.47 / 80.90 Β±0.31 / 81.41 ms β no change β
β QQuery 67 β 241.93 / 244.12 Β±1.88 / 246.94 ms β 237.25 / 242.46 Β±3.79 / 246.30 ms β no change β
β QQuery 68 β 12.17 / 12.36 Β±0.21 / 12.77 ms β 12.00 / 12.20 Β±0.23 / 12.64 ms β no change β
β QQuery 69 β 58.70 / 60.53 Β±2.45 / 65.39 ms β 58.45 / 59.94 Β±1.56 / 62.87 ms β no change β
β QQuery 70 β 108.14 / 109.16 Β±0.66 / 109.97 ms β 107.45 / 109.29 Β±2.17 / 113.26 ms β no change β
β QQuery 71 β 35.67 / 36.55 Β±1.11 / 38.75 ms β 36.51 / 36.80 Β±0.29 / 37.33 ms β no change β
β QQuery 72 β 2079.68 / 2180.06 Β±95.45 / 2317.64 ms β 2105.27 / 2213.98 Β±67.26 / 2301.21 ms β no change β
β QQuery 73 β 9.57 / 9.82 Β±0.26 / 10.31 ms β 9.68 / 9.87 Β±0.23 / 10.32 ms β no change β
β QQuery 74 β 171.95 / 175.19 Β±3.59 / 181.50 ms β 171.86 / 176.50 Β±6.91 / 190.21 ms β no change β
β QQuery 75 β 150.55 / 151.72 Β±0.80 / 152.72 ms β 150.09 / 153.99 Β±5.09 / 164.01 ms β no change β
β QQuery 76 β 36.23 / 36.68 Β±0.49 / 37.64 ms β 36.11 / 36.26 Β±0.13 / 36.44 ms β no change β
β QQuery 77 β 62.24 / 62.69 Β±0.54 / 63.73 ms β 62.66 / 63.94 Β±1.51 / 66.78 ms β no change β
β QQuery 78 β 197.48 / 200.74 Β±1.81 / 202.91 ms β 199.08 / 203.44 Β±4.90 / 211.24 ms β no change β
β QQuery 79 β 68.75 / 71.77 Β±3.59 / 78.62 ms β 68.81 / 69.18 Β±0.23 / 69.53 ms β no change β
β QQuery 80 β 99.55 / 101.70 Β±2.10 / 105.41 ms β 100.33 / 101.54 Β±0.92 / 103.10 ms β no change β
β QQuery 81 β 26.10 / 26.47 Β±0.26 / 26.89 ms β 26.51 / 28.84 Β±2.45 / 32.02 ms β 1.09x slower β
β QQuery 82 β 16.82 / 17.26 Β±0.67 / 18.59 ms β 17.32 / 17.60 Β±0.33 / 18.06 ms β no change β
β QQuery 83 β 40.81 / 43.65 Β±4.23 / 52.00 ms β 40.26 / 40.61 Β±0.24 / 40.94 ms β +1.07x faster β
β QQuery 84 β 30.94 / 31.57 Β±0.41 / 32.12 ms β 30.16 / 30.38 Β±0.16 / 30.63 ms β no change β
β QQuery 85 β 108.50 / 109.23 Β±0.55 / 109.87 ms β 106.81 / 109.55 Β±5.21 / 119.97 ms β no change β
β QQuery 86 β 25.74 / 26.06 Β±0.34 / 26.68 ms β 25.96 / 26.26 Β±0.20 / 26.51 ms β no change β
β QQuery 87 β 62.83 / 66.23 Β±2.76 / 71.13 ms β 63.45 / 64.17 Β±0.58 / 65.19 ms β no change β
β QQuery 88 β 63.52 / 63.97 Β±0.44 / 64.80 ms β 64.86 / 65.08 Β±0.25 / 65.56 ms β no change β
β QQuery 89 β 36.27 / 36.66 Β±0.27 / 36.92 ms β 36.75 / 38.53 Β±3.11 / 44.73 ms β 1.05x slower β
β QQuery 90 β 17.28 / 17.42 Β±0.17 / 17.75 ms β 17.48 / 18.15 Β±0.92 / 19.97 ms β no change β
β QQuery 91 β 46.26 / 48.12 Β±2.33 / 52.60 ms β 46.63 / 46.92 Β±0.22 / 47.17 ms β no change β
β QQuery 92 β 30.72 / 31.87 Β±1.45 / 34.72 ms β 30.18 / 30.55 Β±0.36 / 31.24 ms β no change β
β QQuery 93 β 49.50 / 51.46 Β±1.25 / 52.88 ms β 50.92 / 51.85 Β±0.52 / 52.40 ms β no change β
β QQuery 94 β 38.85 / 39.28 Β±0.38 / 39.94 ms β 38.77 / 39.20 Β±0.43 / 39.93 ms β no change β
β QQuery 95 β 81.15 / 82.48 Β±2.16 / 86.79 ms β 82.05 / 84.20 Β±1.72 / 86.50 ms β no change β
β QQuery 96 β 24.73 / 24.90 Β±0.18 / 25.22 ms β 24.65 / 24.83 Β±0.23 / 25.27 ms β no change β
β QQuery 97 β 47.01 / 47.41 Β±0.26 / 47.81 ms β 48.02 / 48.27 Β±0.24 / 48.67 ms β no change β
β QQuery 98 β 43.05 / 43.56 Β±0.47 / 44.35 ms β 43.73 / 44.28 Β±0.43 / 44.80 ms β no change β
β QQuery 99 β 71.37 / 73.41 Β±3.83 / 81.06 ms β 71.75 / 72.37 Β±0.76 / 73.84 ms β no change β
βββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββ
ββββββββββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 10018.43ms β
β Total Time (split-aggr-spill-order-aggr) β 10074.01ms β
β Average Time (HEAD) β 101.20ms β
β Average Time (split-aggr-spill-order-aggr) β 101.76ms β
β Queries Faster β 1 β
β Queries Slower β 5 β
β Queries with No Change β 93 β
β Queries with Failure β 0 β
ββββββββββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**tpcds β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 55.0s |
| Peak memory | 2.1 GiB |
| Avg memory | 1.4 GiB |
| CPU user | 228.4s |
| CPU sys | 5.8s |
| Peak spill | 0 B |
**tpcds β branch**
| Metric | Value |
|--------|-------|
| Wall time | 55.0s |
| Peak memory | 1.9 GiB |
| Avg memory | 1.3 GiB |
| CPU user | 228.9s |
| CPU sys | 6.1s |
| Peak spill | 0 B |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5041328033
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/23657#issuecomment-5041227943)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
Comparing HEAD and split-aggr-spill-order-aggr
--------------------
Benchmark tpch_sf1.json
--------------------
βββββββββββββ³βββββββββββββββββββββββββββββββββ³βββββββββββββββββββββββββββββββββ³ββββββββββββ
β Query β HEAD β split-aggr-spill-order-aggr β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 38.17 / 38.93 Β±1.12 / 41.13 ms β 38.10 / 39.43 Β±1.48 / 41.56 ms β no change β
β QQuery 2 β 19.10 / 19.68 Β±0.56 / 20.68 ms β 19.25 / 19.43 Β±0.17 / 19.74 ms β no change β
β QQuery 3 β 30.37 / 31.92 Β±1.50 / 34.09 ms β 30.71 / 32.04 Β±1.35 / 34.12 ms β no change β
β QQuery 4 β 17.41 / 17.91 Β±0.42 / 18.66 ms β 17.03 / 17.37 Β±0.20 / 17.66 ms β no change β
β QQuery 5 β 39.31 / 39.83 Β±0.33 / 40.28 ms β 39.14 / 40.29 Β±0.62 / 40.87 ms β no change β
β QQuery 6 β 16.18 / 16.59 Β±0.33 / 17.19 ms β 16.20 / 16.39 Β±0.23 / 16.83 ms β no change β
β QQuery 7 β 44.98 / 47.11 Β±2.25 / 51.39 ms β 44.97 / 46.36 Β±1.13 / 47.61 ms β no change β
β QQuery 8 β 42.34 / 43.08 Β±0.53 / 43.97 ms β 42.38 / 43.17 Β±0.57 / 44.08 ms β no change β
β QQuery 9 β 49.92 / 50.68 Β±0.90 / 52.33 ms β 49.14 / 50.90 Β±1.41 / 53.08 ms β no change β
β QQuery 10 β 42.02 / 42.62 Β±0.96 / 44.54 ms β 41.97 / 42.69 Β±1.24 / 45.17 ms β no change β
β QQuery 11 β 13.19 / 13.28 Β±0.07 / 13.37 ms β 13.03 / 13.52 Β±0.60 / 14.67 ms β no change β
β QQuery 12 β 24.19 / 24.94 Β±1.31 / 27.56 ms β 23.73 / 24.06 Β±0.22 / 24.41 ms β no change β
β QQuery 13 β 32.65 / 34.13 Β±1.58 / 37.10 ms β 32.78 / 34.01 Β±1.69 / 37.34 ms β no change β
β QQuery 14 β 23.31 / 23.63 Β±0.18 / 23.85 ms β 23.50 / 23.65 Β±0.18 / 23.98 ms β no change β
β QQuery 15 β 31.17 / 32.17 Β±0.94 / 33.50 ms β 31.09 / 31.89 Β±0.67 / 32.76 ms β no change β
β QQuery 16 β 14.01 / 14.16 Β±0.12 / 14.33 ms β 13.97 / 14.16 Β±0.17 / 14.47 ms β no change β
β QQuery 17 β 70.19 / 71.30 Β±1.35 / 73.91 ms β 69.74 / 71.31 Β±1.19 / 73.44 ms β no change β
β QQuery 18 β 58.14 / 59.51 Β±1.13 / 61.18 ms β 58.02 / 59.91 Β±2.65 / 65.08 ms β no change β
β QQuery 19 β 33.13 / 34.75 Β±1.66 / 37.03 ms β 33.07 / 33.98 Β±0.80 / 35.11 ms β no change β
β QQuery 20 β 31.50 / 32.27 Β±0.79 / 33.73 ms β 31.75 / 31.89 Β±0.10 / 32.04 ms β no change β
β QQuery 21 β 54.59 / 56.68 Β±2.11 / 60.06 ms β 54.06 / 55.56 Β±1.38 / 57.57 ms β no change β
β QQuery 22 β 13.57 / 13.80 Β±0.15 / 14.01 ms β 13.74 / 13.81 Β±0.07 / 13.92 ms β no change β
βββββββββββββ΄βββββββββββββββββββββββββββββββββ΄βββββββββββββββββββββββββββββββββ΄ββββββββββββ
ββββββββββββββββββββββββββββββββββββββββββββββ³βββββββββββ
β Benchmark Summary β β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 758.97ms β
β Total Time (split-aggr-spill-order-aggr) β 755.79ms β
β Average Time (HEAD) β 34.50ms β
β Average Time (split-aggr-spill-order-aggr) β 34.35ms β
β Queries Faster β 0 β
β Queries Slower β 0 β
β Queries with No Change β 22 β
β Queries with Failure β 0 β
ββββββββββββββββββββββββββββββββββββββββββββββ΄βββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**tpch β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 1.2 GiB |
| Avg memory | 520.1 MiB |
| CPU user | 21.6s |
| CPU sys | 1.7s |
| Peak spill | 0 B |
**tpch β branch**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 1.3 GiB |
| Avg memory | 533.3 MiB |
| CPU user | 21.6s |
| CPU sys | 1.7s |
| Peak spill | 0 B |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#issuecomment-5001398035
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/23657?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `79.06977%` with `99 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 80.66%. Comparing base ([`f151c10`](https://app.codecov.io/gh/apache/datafusion/commit/f151c10fdee89bd6f42c0f786988bd09557ed0fd?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a780de8`](https://app.codecov.io/gh/apache/datafusion/commit/a780de8773903645de7432bb6372c42ff8e549f4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/23657?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ysical-plan/src/aggregates/ordered\_final\_stream.rs](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Faggregates%2Fordered_final_stream.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9hZ2dyZWdhdGVzL29yZGVyZWRfZmluYWxfc3RyZWFtLnJz) | 79.38% | [68 Missing and 6 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ical-plan/src/aggregates/ordered\_partial\_stream.rs](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Faggregates%2Fordered_partial_stream.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9hZ2dyZWdhdGVzL29yZGVyZWRfcGFydGlhbF9zdHJlYW0ucnM=) | 73.68% | [10 Missing and 5 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../aggregates/aggregate\_hash\_table/common\_ordered.rs](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Faggregates%2Faggregate_hash_table%2Fcommon_ordered.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9hZ2dyZWdhdGVzL2FnZ3JlZ2F0ZV9oYXNoX3RhYmxlL2NvbW1vbl9vcmRlcmVkLnJz) | 90.32% | [0 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...afusion/physical-plan/src/aggregates/order/full.rs](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Faggregates%2Forder%2Ffull.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9hZ2dyZWdhdGVzL29yZGVyL2Z1bGwucnM=) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [datafusion/physical-plan/src/aggregates/mod.rs](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Faggregates%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9hZ2dyZWdhdGVzL21vZC5ycw==) | 71.42% | [0 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...tafusion/physical-plan/src/aggregates/order/mod.rs](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Faggregates%2Forder%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy9hZ2dyZWdhdGVzL29yZGVyL21vZC5ycw==) | 66.66% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/23657?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ 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
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/23657?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3638082174
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
}
}
Poll::Ready(Some(Err(e))) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
Poll::Ready(None) => {
self.close_input();
- table.input_done();
- ControlFlow::Continue(OrderedFinalAggregateState::DrainingFinal { table })
+ match spill_context {
+ Some(spill_context) if spill_context.has_spills() => {
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::PreparingMergeInput {
+ table,
+ spill_context,
+ },
+ )
+ }
+ _ => {
+ table.input_done();
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::ProducingOutput { table },
+ )
+ }
+ }
+ }
+ }
+ }
+
+ /// Sorts and spills one complete in-memory state run, then resumes input.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_spilling(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::Spilling {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected spilling state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let mut result = spill_context.spill_table(&mut table);
+
+ // Spilling shrinks the aggregate table and releases its accumulated
+ // memory. Update the reservation accordingly.
+ if self.reservation.try_resize(table.memory_size()).is_err() {
+ // Fold spill error and reservation error into one
+ result = internal_err!("Reservation after spilling should succeed")
Review Comment:
[1e75980](https://github.com/apache/datafusion/pull/23657/commits/1e75980db23c01bcd3516c9624babb5162bb9036)
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3637293062
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -86,6 +270,35 @@ impl OrderedFinalAggregateStream {
partition: usize,
input: SendableRecordBatchStream,
input_order_mode: &InputOrderMode,
+ ) -> Result<Self> {
+ let baseline_metrics = BaselineMetrics::new(&agg.metrics, partition);
+ let group_by_metrics = GroupByMetrics::new(&agg.metrics, partition);
+ let spill_metrics = SpillMetrics::new(&agg.metrics, partition);
+ Self::new_with_input_and_metrics(
+ agg,
+ context,
+ partition,
+ input,
+ input_order_mode,
+ baseline_metrics,
+ group_by_metrics,
+ Some(spill_metrics),
+ )
+ }
+
+ #[expect(
Review Comment:
I agree. I will keep track of those smaller cleanup ideas in the EPIC issue: https://github.com/apache/datafusion/issues/22710
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3637266690
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -47,13 +69,46 @@ pub(crate) struct OrderedFinalAggregateStream {
state: Option<OrderedFinalAggregateState>,
}
+/// Spill configuration and accumulated runs for partially ordered final
+/// aggregation. Each file is one fully group-key-sorted intermediate-state run;
+/// all runs are replayed together after the original input is exhausted.
+struct OrderedFinalSpillContext {
+ /// Aggregate configuration
+ agg: AggregateExec,
+ /// Task context
+ context: Arc<TaskContext>,
+ /// Original partition index
+ partition: usize,
+ /// Target batch size from configuration
+ batch_size: usize,
+ /// Full group-key ordering, such ordering with be kept in: a) individual spill
+ /// files, b) order after final merging and streaming aggregate
+ spill_expr: LexOrdering,
+ /// Spill I/O and metrics manager.
+ spill_manager: SpillManager,
+ /// Fully sorted spill runs waiting to be merged.
+ spills: Vec<SortedSpillFile>,
+}
+
/// See comments at `poll_next()` for details.
enum OrderedFinalAggregateState {
ReadingInput {
table: OrderedAggregateTable<FinalMarker>,
+ spill_context: Option<Box<OrderedFinalSpillContext>>,
},
- DrainingFinal {
+ Spilling {
table: OrderedAggregateTable<FinalMarker>,
+ spill_context: Box<OrderedFinalSpillContext>,
+ },
+ ProducingOutput {
+ table: OrderedAggregateTable<FinalMarker>,
+ },
+ PreparingMergeInput {
Review Comment:
[1aeb794](https://github.com/apache/datafusion/pull/23657/commits/1aeb79459568098f7845452dbdbcf0107b63d9ef)
--
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
Re: [PR] refactor(hash-aggr): Support spilling for ordered aggregation [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #23657:
URL: https://github.com/apache/datafusion/pull/23657#discussion_r3626976863
##########
datafusion/physical-plan/src/aggregates/ordered_final_stream.rs:
##########
@@ -180,36 +497,183 @@ impl OrderedFinalAggregateStream {
next_state,
))
}
+ // Can't do early emit, continue aggregating.
Ok(None) => {
- // Ordered variant doesn't support memory-limited
- // execution, so it errors when memory reservation fails.
- if let Err(e) = self.reservation.try_resize(table.memory_size()) {
- return ControlFlow::Break((
- Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
- ));
- }
-
- // Can't do early emit, continue aggregating.
ControlFlow::Continue(OrderedFinalAggregateState::ReadingInput {
table,
+ spill_context,
})
}
Err(e) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
}
}
Poll::Ready(Some(Err(e))) => ControlFlow::Break((
Poll::Ready(Some(Err(e))),
- OrderedFinalAggregateState::ReadingInput { table },
+ OrderedFinalAggregateState::ReadingInput {
+ table,
+ spill_context,
+ },
)),
Poll::Ready(None) => {
self.close_input();
- table.input_done();
- ControlFlow::Continue(OrderedFinalAggregateState::DrainingFinal { table })
+ match spill_context {
+ Some(spill_context) if spill_context.has_spills() => {
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::PreparingMergeInput {
+ table,
+ spill_context,
+ },
+ )
+ }
+ _ => {
+ table.input_done();
+ ControlFlow::Continue(
+ OrderedFinalAggregateState::ProducingOutput { table },
+ )
+ }
+ }
+ }
+ }
+ }
+
+ /// Sorts and spills one complete in-memory state run, then resumes input.
+ ///
+ /// See comments at `poll_next()` for details.
+ ///
+ /// Returns the next operator state with control flow decision.
+ fn handle_spilling(
+ &mut self,
+ original_state: OrderedFinalAggregateState,
+ ) -> OrderedFinalAggregateStateTransition {
+ let OrderedFinalAggregateState::Spilling {
+ mut table,
+ mut spill_context,
+ } = original_state
+ else {
+ unreachable!("expected spilling state")
+ };
+
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+ let timer = elapsed_compute.timer();
+ let mut result = spill_context.spill_table(&mut table);
+
+ // Spilling shrinks the aggregate table and releases its accumulated
+ // memory. Update the reservation accordingly.
+ if self.reservation.try_resize(table.memory_size()).is_err() {
+ // Fold spill error and reservation error into one
+ result = internal_err!("Reservation after spilling should succeed")
Review Comment:
I was confused for a while as I thought this could be a normal error (e,g, happen if other plans are allocating from the memory manager at the same time). In other words, I think we should just propagte the "resources exhausted" error here
However, i see now this is trying to *shrink* the size of the reservation, which will only fail on underflow (bad accounting)
Maybe we could clarify the message here from `"Reservation after spilling should succeed"` to `"Decreasing allocation after spilling should succeed"`" and also icnlude the errir itself (for debugging) rather than just swalliowing the error
--
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