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/06/03 09:19:07 UTC
[PR] refactor: Split hash aggregation logic into separated streams [datafusion]
2010YOUY01 opened a new pull request, #22729:
URL: https://github.com/apache/datafusion/pull/22729
## 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
- 1/N of https://github.com/apache/datafusion/pull/22712
## 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.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
See issues.
This PR split out partial and final aggregate strem from `GroupsHashAggregateStream`
To fully migrate hash aggregation, we have to
- Port this optimization back https://github.com/apache/datafusion/pull/11627
- Support spilling
I think they should be leave to follow up PRs
Todo in this PR:
- [ ] Add a temporary configuration `enable_migration_aggregate` to turn off this path
Since it should be a regression if the above features are not added, it also helps if to prevent potential regressions from the migration of other aggregate streams.
## 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.
-->
Split out the streams from `GroupsHashAggregateStream`
1. Partial stage of hash aggregation
2. Final stage of hash aggregation
## 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
3. 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)?
-->
## 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 merged PR #22729:
URL: https://github.com/apache/datafusion/pull/22729
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4644825105
Sorry for delay, make sense to improve readability in 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#discussion_r3363088976
##########
datafusion/physical-plan/src/aggregates/hash_table.rs:
##########
@@ -0,0 +1,615 @@
+// Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Good point, though I'd like to postpone this until later. I think we can organize files when the migration is nearly finished.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622519035
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#discussion_r3363075182
##########
datafusion/physical-plan/src/aggregates/hash_table.rs:
##########
@@ -0,0 +1,615 @@
+// 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.
+
+use std::collections::HashMap;
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BooleanArray, new_null_array};
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::RecordBatch;
+use datafusion_common::{Result, internal_err};
+use datafusion_execution::memory_pool::proxy::VecAllocExt;
+use datafusion_expr::{EmitTo, GroupsAccumulator};
+
+use super::group_values::{GroupByMetrics, GroupValues, new_group_values};
+use super::order::GroupOrdering;
+use super::row_hash::create_group_accumulator;
+use super::{
+ AggregateExec, PhysicalGroupBy, aggregate_expressions, evaluate_group_by,
+ group_id_array, max_duplicate_ordinal,
+};
+use crate::PhysicalExpr;
+use crate::metrics::{MetricBuilder, MetricCategory};
+
+/// Marker for raw rows -> partial state aggregation.
+pub(super) struct Partial;
+/// Marker for partial state -> final value aggregation.
+pub(super) struct Final;
+
+/// Grouped hash table shared by the initial-partial and partial-final paths.
+///
+/// While building, it consumes input batches and updates group / accumulator
+/// state. While outputting, it incrementally output the materialized batches.
+///
+/// # Marker Type
+/// `AggrMode` is a zero-sized marker type used with `PhantomData` to keep the
+/// initial-partial and partial-final update logic in separate impl blocks. Different
+/// stages has different semantics and applicable optimizations, this makes the
+/// implementation easier to follow.
+pub(super) struct AggregateHashTable<AggrMode> {
Review Comment:
It is intentionally unused, this is like compile time boolean flag to choose the implementation
```rs
// semantically equivalent to
fn aggregate_batch() {
if self.is_partial_aggregate {
self.aggregate_batch_partial()
} else {
self.aggregate_batch_final()
}
}
```
The benefits are
- Comparing with shared struct, less verbose
- Comparing with using runtime flag. Since two variants have different semantics (i.e. input schema), this approach is more explicit.
I am aware this is a not common pattern, I'll add more comments.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622481723
Benchmark for [this request](https://github.com/apache/datafusion/pull/22729#issuecomment-4622443332) failed.
Last 20 lines of output:
<details><summary>Click to expand</summary>
```
Compiling datafusion-optimizer v53.1.0 (/workspace/datafusion-branch/datafusion/optimizer)
Compiling datafusion-physical-plan v53.1.0 (/workspace/datafusion-branch/datafusion/physical-plan)
Compiling datafusion-physical-expr-adapter v53.1.0 (/workspace/datafusion-branch/datafusion/physical-expr-adapter)
warning: ignoring -C extra-filename flag due to -o flag
error[E0428]: the name `Partial` is defined multiple times
--> datafusion/physical-plan/src/aggregates/hash_table.rs:42:1
|
40 | pub(super) struct Partial;
| -------------------------- previous definition of the type `Partial` here
41 | /// Marker for partial state -> final value aggregation.
42 | pub(super) struct Partial;
| ^^^^^^^^^^^^^^^^^^^^^^^^^^ `Partial` redefined here
|
= note: `Partial` must be defined only once in the type namespace of this module
For more information about this error, try `rustc --explain E0428`.
warning: `datafusion-physical-plan` (lib) generated 1 warning
error: could not compile `datafusion-physical-plan` (lib) due to 1 previous error; 1 warning emitted
warning: build failed, waiting for other jobs to finish...
```
</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: Split hash aggregation logic into separated streams [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622453281
<!-- semver-check-comment -->
Thank you for opening this pull request!
Reviewer note: [cargo-semver-checks](https://github.com/obi1kenobi/cargo-semver-checks) reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).
<details>
<summary>Details</summary>
```
Cloning apache/main
Building datafusion-common v53.1.0 (current)
Built [ 33.462s] (current)
Parsing datafusion-common v53.1.0 (current)
Parsed [ 0.057s] (current)
Building datafusion-common v53.1.0 (baseline)
Built [ 33.133s] (baseline)
Parsing datafusion-common v53.1.0 (baseline)
Parsed [ 0.058s] (baseline)
Checking datafusion-common v53.1.0 -> v53.1.0 (no change; assume patch)
Checked [ 0.625s] 223 checks: 222 pass, 1 fail, 0 warn, 30 skip
--- failure constructible_struct_adds_field: externally-constructible struct adds field ---
Description:
A pub struct constructible with a struct literal has a new pub field. Existing struct literals must be updated to include the new field.
ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.48.0/src/lints/constructible_struct_adds_field.ron
Failed in:
field ExecutionOptions.enable_migration_aggregate in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:516
Summary semver requires new major version: 1 major and 0 minor checks failed
Finished [ 68.546s] datafusion-common
Building datafusion-physical-plan v53.1.0 (current)
Built [ 35.681s] (current)
Parsing datafusion-physical-plan v53.1.0 (current)
Parsed [ 0.127s] (current)
Building datafusion-physical-plan v53.1.0 (baseline)
Built [ 35.647s] (baseline)
Parsing datafusion-physical-plan v53.1.0 (baseline)
Parsed [ 0.125s] (baseline)
Checking datafusion-physical-plan v53.1.0 -> v53.1.0 (no change; assume patch)
Checked [ 0.574s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 73.262s] datafusion-physical-plan
Building datafusion-sqllogictest v53.1.0 (current)
Built [ 162.958s] (current)
Parsing datafusion-sqllogictest v53.1.0 (current)
Parsed [ 0.024s] (current)
Building datafusion-sqllogictest v53.1.0 (baseline)
Built [ 163.848s] (baseline)
Parsing datafusion-sqllogictest v53.1.0 (baseline)
Parsed [ 0.025s] (baseline)
Checking datafusion-sqllogictest v53.1.0 -> v53.1.0 (no change; assume patch)
Checked [ 0.095s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 329.322s] datafusion-sqllogictest
```
</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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622668423
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622519035)
**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-refactor-only
--------------------
Benchmark tpcds_sf1.json
--------------------
βββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β split-aggr-refactor-only β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 5.91 / 6.34 Β±0.79 / 7.91 ms β 5.79 / 6.27 Β±0.75 / 7.76 ms β no change β
β QQuery 2 β 80.06 / 80.59 Β±0.29 / 80.88 ms β 80.81 / 80.98 Β±0.09 / 81.07 ms β no change β
β QQuery 3 β 28.46 / 29.01 Β±0.44 / 29.62 ms β 28.65 / 28.89 Β±0.16 / 29.15 ms β no change β
β QQuery 4 β 487.89 / 492.39 Β±3.63 / 498.80 ms β 483.77 / 491.47 Β±3.95 / 494.87 ms β no change β
β QQuery 5 β 51.44 / 52.35 Β±1.28 / 54.83 ms β 50.96 / 51.63 Β±0.35 / 51.90 ms β no change β
β QQuery 6 β 36.14 / 36.73 Β±0.46 / 37.24 ms β 36.61 / 37.00 Β±0.39 / 37.70 ms β no change β
β QQuery 7 β 95.49 / 96.10 Β±0.45 / 96.72 ms β 95.78 / 96.18 Β±0.38 / 96.84 ms β no change β
β QQuery 8 β 36.63 / 38.66 Β±3.11 / 44.84 ms β 36.88 / 38.81 Β±2.89 / 44.52 ms β no change β
β QQuery 9 β 52.84 / 54.75 Β±1.40 / 56.73 ms β 51.48 / 53.21 Β±1.00 / 54.23 ms β no change β
β QQuery 10 β 68.67 / 68.96 Β±0.23 / 69.33 ms β 68.74 / 68.91 Β±0.13 / 69.05 ms β no change β
β QQuery 11 β 298.61 / 304.39 Β±3.96 / 311.02 ms β 299.76 / 303.28 Β±1.84 / 304.79 ms β no change β
β QQuery 12 β 28.56 / 29.02 Β±0.43 / 29.66 ms β 28.58 / 29.13 Β±0.53 / 30.09 ms β no change β
β QQuery 13 β 119.14 / 120.60 Β±2.27 / 125.10 ms β 119.96 / 121.08 Β±0.92 / 122.65 ms β no change β
β QQuery 14 β 507.55 / 509.52 Β±2.83 / 515.12 ms β 504.95 / 507.66 Β±1.78 / 510.07 ms β no change β
β QQuery 15 β 58.97 / 59.68 Β±0.70 / 61.02 ms β 57.95 / 60.65 Β±2.53 / 65.27 ms β no change β
β QQuery 16 β 6.66 / 6.82 Β±0.21 / 7.23 ms β 6.67 / 6.83 Β±0.16 / 7.10 ms β no change β
β QQuery 17 β 81.05 / 82.00 Β±1.36 / 84.69 ms β 81.05 / 81.76 Β±0.61 / 82.63 ms β no change β
β QQuery 18 β 125.36 / 126.05 Β±0.58 / 126.77 ms β 126.13 / 126.84 Β±0.60 / 127.88 ms β no change β
β QQuery 19 β 42.31 / 43.29 Β±1.26 / 45.72 ms β 42.42 / 43.48 Β±1.24 / 45.87 ms β no change β
β QQuery 20 β 35.66 / 36.22 Β±0.49 / 36.79 ms β 35.73 / 36.56 Β±0.56 / 37.47 ms β no change β
β QQuery 21 β 17.10 / 17.34 Β±0.25 / 17.75 ms β 16.83 / 17.20 Β±0.43 / 17.95 ms β no change β
β QQuery 22 β 62.04 / 64.67 Β±2.39 / 68.84 ms β 62.32 / 63.16 Β±0.78 / 64.56 ms β no change β
β QQuery 23 β 342.96 / 345.36 Β±2.55 / 348.86 ms β 394.51 / 398.01 Β±2.79 / 402.13 ms β 1.15x slower β
β QQuery 24 β 225.47 / 227.00 Β±1.62 / 230.02 ms β 225.47 / 229.96 Β±4.68 / 238.81 ms β no change β
β QQuery 25 β 112.78 / 114.74 Β±1.44 / 116.38 ms β 114.98 / 117.37 Β±2.20 / 121.25 ms β no change β
β QQuery 26 β 58.18 / 58.57 Β±0.31 / 58.96 ms β 58.30 / 58.97 Β±0.52 / 59.65 ms β no change β
β QQuery 27 β 6.65 / 6.78 Β±0.19 / 7.15 ms β 6.65 / 6.77 Β±0.17 / 7.09 ms β no change β
β QQuery 28 β 60.97 / 61.33 Β±0.40 / 62.05 ms β 57.59 / 63.94 Β±4.55 / 70.74 ms β no change β
β QQuery 29 β 98.46 / 102.64 Β±2.99 / 105.77 ms β 98.75 / 100.13 Β±1.67 / 102.47 ms β no change β
β QQuery 30 β 32.02 / 32.69 Β±0.70 / 33.94 ms β 32.49 / 32.72 Β±0.19 / 33.05 ms β no change β
β QQuery 31 β 111.43 / 112.44 Β±1.21 / 114.69 ms β 111.70 / 113.86 Β±1.61 / 116.50 ms β no change β
β QQuery 32 β 20.00 / 20.38 Β±0.34 / 20.98 ms β 19.93 / 20.17 Β±0.38 / 20.93 ms β no change β
β QQuery 33 β 38.97 / 39.51 Β±0.35 / 39.98 ms β 38.65 / 39.03 Β±0.37 / 39.69 ms β no change β
β QQuery 34 β 9.57 / 9.82 Β±0.24 / 10.18 ms β 9.37 / 10.32 Β±0.85 / 11.91 ms β 1.05x slower β
β QQuery 35 β 76.88 / 77.40 Β±0.38 / 77.99 ms β 77.24 / 80.43 Β±3.26 / 85.90 ms β no change β
β QQuery 36 β 5.84 / 5.95 Β±0.21 / 6.36 ms β 5.84 / 5.98 Β±0.20 / 6.38 ms β no change β
β QQuery 37 β 6.85 / 6.89 Β±0.04 / 6.94 ms β 6.90 / 7.11 Β±0.17 / 7.41 ms β no change β
β QQuery 38 β 63.21 / 64.13 Β±0.89 / 65.57 ms β 63.05 / 64.27 Β±0.81 / 65.06 ms β no change β
β QQuery 39 β 448.98 / 452.82 Β±3.68 / 459.53 ms β 454.03 / 460.59 Β±5.66 / 470.56 ms β no change β
β QQuery 40 β 22.94 / 23.15 Β±0.17 / 23.44 ms β 23.59 / 25.38 Β±3.21 / 31.79 ms β 1.10x slower β
β QQuery 41 β 11.33 / 11.46 Β±0.15 / 11.76 ms β 11.16 / 11.32 Β±0.16 / 11.64 ms β no change β
β QQuery 42 β 23.63 / 23.98 Β±0.52 / 25.00 ms β 24.26 / 24.76 Β±0.64 / 26.00 ms β no change β
β QQuery 43 β 4.94 / 4.99 Β±0.08 / 5.15 ms β 4.60 / 4.71 Β±0.16 / 5.04 ms β +1.06x faster β
β QQuery 44 β 10.78 / 10.84 Β±0.06 / 10.94 ms β 10.47 / 10.74 Β±0.17 / 10.88 ms β no change β
β QQuery 45 β 38.08 / 39.64 Β±2.10 / 43.81 ms β 37.97 / 38.76 Β±0.77 / 39.76 ms β no change β
β QQuery 46 β 11.95 / 13.23 Β±2.21 / 17.63 ms β 11.81 / 12.20 Β±0.28 / 12.65 ms β +1.08x faster β
β QQuery 47 β 230.20 / 235.44 Β±3.58 / 239.49 ms β 232.00 / 237.65 Β±5.51 / 246.01 ms β no change β
β QQuery 48 β 95.99 / 96.58 Β±0.93 / 98.43 ms β 97.04 / 100.39 Β±3.97 / 108.09 ms β no change β
β QQuery 49 β 76.73 / 79.47 Β±2.26 / 83.33 ms β 77.70 / 79.13 Β±1.19 / 80.32 ms β no change β
β QQuery 50 β 59.62 / 60.17 Β±0.29 / 60.44 ms β 59.37 / 59.76 Β±0.28 / 60.19 ms β no change β
β QQuery 51 β 90.30 / 92.85 Β±2.15 / 96.33 ms β 98.48 / 101.86 Β±3.07 / 107.00 ms β 1.10x slower β
β QQuery 52 β 23.94 / 24.19 Β±0.17 / 24.34 ms β 24.13 / 24.40 Β±0.22 / 24.77 ms β no change β
β QQuery 53 β 29.89 / 31.65 Β±3.17 / 37.99 ms β 29.75 / 29.97 Β±0.18 / 30.24 ms β +1.06x faster β
β QQuery 54 β 55.06 / 55.70 Β±0.46 / 56.27 ms β 55.22 / 55.71 Β±0.50 / 56.66 ms β no change β
β QQuery 55 β 23.45 / 23.66 Β±0.16 / 23.86 ms β 23.62 / 24.19 Β±0.43 / 24.81 ms β no change β
β QQuery 56 β 38.74 / 39.11 Β±0.24 / 39.40 ms β 38.86 / 39.43 Β±0.53 / 40.28 ms β no change β
β QQuery 57 β 176.50 / 178.71 Β±3.49 / 185.62 ms β 176.73 / 178.18 Β±1.25 / 180.03 ms β no change β
β QQuery 58 β 116.72 / 118.60 Β±1.90 / 122.22 ms β 117.01 / 117.96 Β±0.80 / 119.05 ms β no change β
β QQuery 59 β 117.48 / 117.71 Β±0.18 / 118.01 ms β 118.18 / 120.05 Β±1.76 / 122.61 ms β no change β
β QQuery 60 β 39.22 / 39.77 Β±0.42 / 40.42 ms β 39.69 / 40.24 Β±0.43 / 40.80 ms β no change β
β QQuery 61 β 13.02 / 14.59 Β±2.81 / 20.20 ms β 13.05 / 13.24 Β±0.23 / 13.68 ms β +1.10x faster β
β QQuery 62 β 46.54 / 46.76 Β±0.13 / 46.90 ms β 46.13 / 46.40 Β±0.23 / 46.79 ms β no change β
β QQuery 63 β 29.75 / 30.10 Β±0.27 / 30.44 ms β 29.51 / 31.50 Β±3.11 / 37.70 ms β no change β
β QQuery 64 β 398.83 / 404.82 Β±5.61 / 413.43 ms β 399.25 / 403.82 Β±5.13 / 413.71 ms β no change β
β QQuery 65 β 141.20 / 148.10 Β±4.14 / 153.00 ms β 142.74 / 147.69 Β±2.60 / 150.09 ms β no change β
β QQuery 66 β 79.16 / 80.34 Β±0.86 / 81.78 ms β 79.50 / 82.74 Β±3.39 / 87.19 ms β no change β
β QQuery 67 β 248.82 / 253.71 Β±5.54 / 263.37 ms β 245.91 / 256.59 Β±6.18 / 264.79 ms β no change β
β QQuery 68 β 11.65 / 11.83 Β±0.17 / 12.13 ms β 11.69 / 11.87 Β±0.18 / 12.22 ms β no change β
β QQuery 69 β 63.01 / 68.09 Β±9.60 / 87.28 ms β 62.35 / 62.93 Β±0.50 / 63.62 ms β +1.08x faster β
β QQuery 70 β 107.21 / 110.95 Β±6.47 / 123.87 ms β 107.26 / 112.54 Β±3.18 / 116.32 ms β no change β
β QQuery 71 β 35.59 / 36.00 Β±0.24 / 36.26 ms β 35.94 / 36.75 Β±0.46 / 37.33 ms β no change β
β QQuery 72 β 2090.99 / 2137.06 Β±28.65 / 2167.54 ms β 2109.95 / 2178.18 Β±46.13 / 2232.07 ms β no change β
β QQuery 73 β 9.23 / 9.51 Β±0.26 / 9.89 ms β 9.30 / 9.56 Β±0.26 / 10.02 ms β no change β
β QQuery 74 β 170.29 / 174.18 Β±5.33 / 184.53 ms β 173.18 / 176.81 Β±3.57 / 183.37 ms β no change β
β QQuery 75 β 146.65 / 154.77 Β±6.39 / 163.74 ms β 146.84 / 150.73 Β±4.46 / 159.24 ms β no change β
β QQuery 76 β 35.03 / 35.69 Β±0.40 / 36.14 ms β 35.77 / 36.29 Β±0.38 / 36.95 ms β no change β
β QQuery 77 β 61.04 / 61.67 Β±0.41 / 62.14 ms β 60.87 / 61.41 Β±0.48 / 62.19 ms β no change β
β QQuery 78 β 189.56 / 191.74 Β±2.14 / 194.45 ms β 189.43 / 195.15 Β±8.32 / 211.41 ms β no change β
β QQuery 79 β 67.34 / 72.71 Β±8.61 / 89.88 ms β 67.09 / 68.17 Β±1.01 / 69.92 ms β +1.07x faster β
β QQuery 80 β 102.11 / 104.00 Β±1.43 / 106.23 ms β 103.80 / 107.12 Β±3.01 / 111.61 ms β no change β
β QQuery 81 β 25.33 / 25.80 Β±0.25 / 26.02 ms β 25.77 / 26.20 Β±0.26 / 26.54 ms β no change β
β QQuery 82 β 16.32 / 16.46 Β±0.11 / 16.64 ms β 16.31 / 16.46 Β±0.11 / 16.64 ms β no change β
β QQuery 83 β 39.99 / 44.31 Β±5.20 / 51.17 ms β 39.77 / 40.10 Β±0.20 / 40.38 ms β +1.11x faster β
β QQuery 84 β 34.95 / 35.75 Β±0.59 / 36.66 ms β 34.13 / 34.58 Β±0.38 / 35.27 ms β no change β
β QQuery 85 β 106.67 / 108.54 Β±1.04 / 109.88 ms β 108.83 / 110.17 Β±1.86 / 113.81 ms β no change β
β QQuery 86 β 24.89 / 25.25 Β±0.31 / 25.68 ms β 24.64 / 25.17 Β±0.29 / 25.46 ms β no change β
β QQuery 87 β 63.81 / 66.69 Β±4.34 / 75.05 ms β 63.09 / 65.04 Β±2.23 / 69.24 ms β no change β
β QQuery 88 β 62.13 / 62.49 Β±0.36 / 63.19 ms β 62.43 / 63.21 Β±1.17 / 65.55 ms β no change β
β QQuery 89 β 35.55 / 35.82 Β±0.36 / 36.52 ms β 36.12 / 36.59 Β±0.25 / 36.81 ms β no change β
β QQuery 90 β 16.66 / 16.90 Β±0.15 / 17.14 ms β 16.66 / 16.90 Β±0.21 / 17.23 ms β no change β
β QQuery 91 β 44.54 / 46.81 Β±2.61 / 50.05 ms β 44.88 / 45.47 Β±0.47 / 46.16 ms β no change β
β QQuery 92 β 29.86 / 30.01 Β±0.17 / 30.32 ms β 29.16 / 29.69 Β±0.61 / 30.68 ms β no change β
β QQuery 93 β 51.90 / 52.62 Β±0.52 / 53.33 ms β 52.38 / 54.74 Β±2.45 / 58.30 ms β no change β
β QQuery 94 β 38.16 / 38.84 Β±0.40 / 39.29 ms β 38.58 / 38.90 Β±0.20 / 39.11 ms β no change β
β QQuery 95 β 83.61 / 85.62 Β±1.56 / 88.04 ms β 83.99 / 84.49 Β±0.34 / 85.07 ms β no change β
β QQuery 96 β 23.90 / 24.28 Β±0.28 / 24.72 ms β 23.84 / 24.15 Β±0.28 / 24.62 ms β no change β
β QQuery 97 β 45.70 / 46.57 Β±0.64 / 47.62 ms β 53.49 / 54.96 Β±1.72 / 58.27 ms β 1.18x slower β
β QQuery 98 β 42.72 / 43.95 Β±1.41 / 46.46 ms β 43.32 / 43.89 Β±0.48 / 44.69 ms β no change β
β QQuery 99 β 70.51 / 71.92 Β±2.55 / 77.02 ms β 70.35 / 70.77 Β±0.34 / 71.35 ms β no change β
βββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 10440.05ms β
β Total Time (split-aggr-refactor-only) β 10568.39ms β
β Average Time (HEAD) β 105.46ms β
β Average Time (split-aggr-refactor-only) β 106.75ms β
β Queries Faster β 7 β
β Queries Slower β 5 β
β Queries with No Change β 87 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**tpcds β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 55.0s |
| Peak memory | 6.9 GiB |
| Avg memory | 6.2 GiB |
| CPU user | 237.2s |
| CPU sys | 6.6s |
| Peak spill | 0 B |
**tpcds β branch**
| Metric | Value |
|--------|-------|
| Wall time | 55.0s |
| Peak memory | 6.9 GiB |
| Avg memory | 6.4 GiB |
| CPU user | 240.0s |
| CPU sys | 6.9s |
| 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622479717
Benchmark for [this request](https://github.com/apache/datafusion/pull/22729#issuecomment-4622443332) failed.
Last 20 lines of output:
<details><summary>Click to expand</summary>
```
Compiling datafusion-optimizer v53.1.0 (/workspace/datafusion-branch/datafusion/optimizer)
Compiling datafusion-physical-plan v53.1.0 (/workspace/datafusion-branch/datafusion/physical-plan)
Compiling datafusion-physical-expr-adapter v53.1.0 (/workspace/datafusion-branch/datafusion/physical-expr-adapter)
warning: ignoring -C extra-filename flag due to -o flag
error[E0428]: the name `Partial` is defined multiple times
--> datafusion/physical-plan/src/aggregates/hash_table.rs:42:1
|
40 | pub(super) struct Partial;
| -------------------------- previous definition of the type `Partial` here
41 | /// Marker for partial state -> final value aggregation.
42 | pub(super) struct Partial;
| ^^^^^^^^^^^^^^^^^^^^^^^^^^ `Partial` redefined here
|
= note: `Partial` must be defined only once in the type namespace of this module
For more information about this error, try `rustc --explain E0428`.
warning: `datafusion-physical-plan` (lib) generated 1 warning
error: could not compile `datafusion-physical-plan` (lib) due to 1 previous error; 1 warning emitted
warning: build failed, waiting for other jobs to finish...
```
</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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#discussion_r3363091286
##########
datafusion/sqllogictest/test_files/information_schema.slt:
##########
@@ -370,6 +371,7 @@ datafusion.execution.batch_size 8192 Default batch size while creating new batch
datafusion.execution.coalesce_batches true When set to true, record batches will be examined between each operator and small batches will be coalesced into larger batches. This is helpful when there are highly selective filters or joins that could produce tiny output batches. The target batch size is determined by the configuration setting
datafusion.execution.collect_statistics true Should DataFusion collect statistics when first creating a table. Has no effect after the table is created. Applies to the default `ListingTableProvider` in DataFusion. Defaults to true.
datafusion.execution.enable_ansi_mode false Whether to enable ANSI SQL mode. The flag is experimental and relevant only for DataFusion Spark built-in functions When `enable_ansi_mode` is set to `true`, the query engine follows ANSI SQL semantics for expressions, casting, and error handling. This means: - **Strict type coercion rules:** implicit casts between incompatible types are disallowed. - **Standard SQL arithmetic behavior:** operations such as division by zero, numeric overflow, or invalid casts raise runtime errors rather than returning `NULL` or adjusted values. - **Consistent ANSI behavior** for string concatenation, comparisons, and `NULL` handling. When `enable_ansi_mode` is `false` (the default), the engine uses a more permissive, non-ANSI mode designed for user convenience and backward compatibility. In this mode: - Implicit casts between types are allowed (e.g., string to integer when possible). - Arithmetic operations are more lenient β for example, `abs()` on
the minimum representable integer value returns the input value instead of raising overflow. - Division by zero or invalid casts may return `NULL` instead of failing. # Default `false` β ANSI SQL mode is disabled by default.
+datafusion.execution.enable_migration_aggregate true Temporary switch for aggregate stream implementations that are being migrated from `GroupedHashAggregateStream`. When set to true, DataFusion tries the migrated implementations when their preconditions are satisfied. When set to false, grouped aggregation falls back to `GroupedHashAggregateStream`. This option will be removed after the migration is finished. See <https://github.com/apache/datafusion/issues/22710> for details.
Review Comment:
Yes this is a good idea, I'll address 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4611653471
> I'm curious about the high-level vision: is the plan to close #15591 in favor of this new approach?
Yes, the goal is to support blocked state management.
The existing challenge is that the current implementation is hard to extend and review. I want to clean things up through this refactor first, and then apply the actual change.
> I would like the redesign of hash aggregation to take into account the memory constraints imposed by the finite memory pool, i.e. how does the implementation perform under OOM conditions.
>
> * how do we improve memory accounting (see [Hash aggregation produces batches reporting huge memory sizeΒ #22526](https://github.com/apache/datafusion/issues/22526)).
> * how do we avoid excessive memory allocations during OOM condition (see [fix: reduce memory allocation overhead during partial aggregation earβ¦Β #22165](https://github.com/apache/datafusion/pull/22165))
> * other issues such as [[EPIC] Eliminate Long Polls in HashAggregate via Chunked Storage and Incremental EmissionΒ #19906](https://github.com/apache/datafusion/issues/19906)
>
> Otherwise we'll end up with the same issues that exist now. E.g. EmitTo::First(n) wasn't designed for emitting a large portion of the existing groups, so it over-allocated when used for emitting early in partial aggregation OOM case.
All of these issues are symptoms of managing state in a large contiguous `Vec`. Blocked memory allocation should address them naturally.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#discussion_r3355621857
##########
datafusion/physical-plan/src/aggregates/hash_aggregate.rs:
##########
@@ -0,0 +1,349 @@
+// 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.
+
+//! Grouped hash aggregation for simple multi-stage aggregation paths.
+//!
+//! This module handles the basic grouped two-stage paths:
+//!
+//! ```text
+//! input rows -> GROUP BY hash table -> accumulator state rows
+//! state rows -> GROUP BY hash table -> final aggregate rows
+//! ```
+//!
+//! `AggregateExec` keeps finite-memory, ordered, limit, grouping-set,
+//! `partial state -> partial state`, and single-stage aggregation on
+//! `GroupedHashAggregateStream` for now.
+
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::RecordBatch;
+use datafusion_common::Result;
+use datafusion_execution::TaskContext;
+use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use futures::stream::{Stream, StreamExt};
+
+use super::AggregateExec;
+use super::hash_table::{AggregateHashTable, InitialPartial, PartialFinal};
+use crate::metrics::{BaselineMetrics, MetricBuilder, RecordOutput, SpillMetrics};
+use crate::stream::EmptyRecordBatchStream;
+use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream, metrics};
+
+/// Hash aggregation uses a 2-stage (partial and final) hash aggregation, this stream
+/// is for the partial stage.
+///
+/// # Example
+///
+/// select k, avg(v) from t group by k;
+///
+/// ## Plan
+/// AggregateExec(stage=final)
+/// -- RepartitionExec(hash(k))
+/// ---- AggregateExec(stage=partial)
+///
+/// ## Partial Stage Behavior
+/// Input: raw rows
+/// Output: partial states for all groups (e.g. for avg(x), it's sum(x), count(x))
+///
+/// ## Final Stage Behavior
+/// Input: partial states
+/// Output: results for all groups (e.g. for avg(x), it's avg(x) calculated from the state)
+pub(crate) struct InitialPartialHashAggregateStream {
+ /// Output schema: group columns followed by partial aggregate state columns.
+ schema: SchemaRef,
+
+ /// Input batches containing raw rows, not partial aggregate state.
+ input: SendableRecordBatchStream,
+
+ /// Hash table state for this aggregate stream.
+ hash_table: AggregateHashTable<InitialPartial>,
+
+ /// Memory reservation for group keys and accumulators.
+ reservation: MemoryReservation,
+
+ /// Execution metrics shared with the aggregate plan node.
+ baseline_metrics: BaselineMetrics,
+
+ /// Tracks partial aggregation row reduction, matching `GroupedHashAggregateStream`.
+ reduction_factor: metrics::RatioMetrics,
+}
+
+/// Hash aggregation uses a 2-stage (partial and final) hash aggregation, this stream
+/// is for the final stage.
+///
+/// See [`InitialPartialHashAggregateStream`] for details.
+pub(crate) struct PartialFinalHashAggregateStream {
Review Comment:
A minor nit is I found this naming confusing
* PartialFinalHashAggregateStream -- the `Partial` is before `Final`
* `InitialPartialHashAggregateStream` -- the `Partial` is after `Initial`
Would it be more consistent to call it `PartialInitialHashAggregateStream`
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -499,6 +506,8 @@ impl PartialEq for PhysicalGroupBy {
#[expect(clippy::large_enum_variant)]
enum StreamType {
AggregateStream(AggregateStream),
+ InitialPartialHash(InitialPartialHashAggregateStream),
+ PartialFinalHash(PartialFinalHashAggregateStream),
GroupedHash(GroupedHashAggregateStream),
Review Comment:
I am a little unclear what the difference betweeen `InitialPartialHash` `PartialFinalHash` and `GroupedHash`
Is it possible to leave some comments?
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -499,6 +506,8 @@ impl PartialEq for PhysicalGroupBy {
#[expect(clippy::large_enum_variant)]
enum StreamType {
AggregateStream(AggregateStream),
+ InitialPartialHash(InitialPartialHashAggregateStream),
+ PartialFinalHash(PartialFinalHashAggregateStream),
GroupedHash(GroupedHashAggregateStream),
Review Comment:
Also, I am not clear that the relationship between these different strema types are
For example is `GroupedHash` the same as InitialPartialHash?
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -499,6 +506,8 @@ impl PartialEq for PhysicalGroupBy {
#[expect(clippy::large_enum_variant)]
enum StreamType {
AggregateStream(AggregateStream),
+ InitialPartialHash(InitialPartialHashAggregateStream),
+ PartialFinalHash(PartialFinalHashAggregateStream),
GroupedHash(GroupedHashAggregateStream),
Review Comment:
Also, why os `GroupedHashAggregateStream`not changed π€
##########
datafusion/physical-plan/src/aggregates/hash_aggregate.rs:
##########
@@ -0,0 +1,349 @@
+// 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.
+
+//! Grouped hash aggregation for simple multi-stage aggregation paths.
+//!
+//! This module handles the basic grouped two-stage paths:
+//!
+//! ```text
+//! input rows -> GROUP BY hash table -> accumulator state rows
+//! state rows -> GROUP BY hash table -> final aggregate rows
+//! ```
+//!
+//! `AggregateExec` keeps finite-memory, ordered, limit, grouping-set,
+//! `partial state -> partial state`, and single-stage aggregation on
+//! `GroupedHashAggregateStream` for now.
Review Comment:
What does it mean "for now" ? That seems like a comment about the PR rather than about the state of the code (and thus may be confusing once this PR is merged)
##########
datafusion/physical-plan/src/aggregates/hash_aggregate.rs:
##########
@@ -0,0 +1,349 @@
+// 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.
+
+//! Grouped hash aggregation for simple multi-stage aggregation paths.
+//!
+//! This module handles the basic grouped two-stage paths:
+//!
+//! ```text
+//! input rows -> GROUP BY hash table -> accumulator state rows
+//! state rows -> GROUP BY hash table -> final aggregate rows
+//! ```
+//!
+//! `AggregateExec` keeps finite-memory, ordered, limit, grouping-set,
+//! `partial state -> partial state`, and single-stage aggregation on
+//! `GroupedHashAggregateStream` for now.
+
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::RecordBatch;
+use datafusion_common::Result;
+use datafusion_execution::TaskContext;
+use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use futures::stream::{Stream, StreamExt};
+
+use super::AggregateExec;
+use super::hash_table::{AggregateHashTable, InitialPartial, PartialFinal};
+use crate::metrics::{BaselineMetrics, MetricBuilder, RecordOutput, SpillMetrics};
+use crate::stream::EmptyRecordBatchStream;
+use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream, metrics};
+
+/// Hash aggregation uses a 2-stage (partial and final) hash aggregation, this stream
+/// is for the partial stage.
+///
+/// # Example
+///
+/// select k, avg(v) from t group by k;
+///
+/// ## Plan
+/// AggregateExec(stage=final)
+/// -- RepartitionExec(hash(k))
+/// ---- AggregateExec(stage=partial)
+///
+/// ## Partial Stage Behavior
+/// Input: raw rows
+/// Output: partial states for all groups (e.g. for avg(x), it's sum(x), count(x))
+///
+/// ## Final Stage Behavior
+/// Input: partial states
+/// Output: results for all groups (e.g. for avg(x), it's avg(x) calculated from the state)
+pub(crate) struct InitialPartialHashAggregateStream {
+ /// Output schema: group columns followed by partial aggregate state columns.
+ schema: SchemaRef,
+
+ /// Input batches containing raw rows, not partial aggregate state.
+ input: SendableRecordBatchStream,
+
+ /// Hash table state for this aggregate stream.
+ hash_table: AggregateHashTable<InitialPartial>,
+
+ /// Memory reservation for group keys and accumulators.
+ reservation: MemoryReservation,
+
+ /// Execution metrics shared with the aggregate plan node.
+ baseline_metrics: BaselineMetrics,
+
+ /// Tracks partial aggregation row reduction, matching `GroupedHashAggregateStream`.
+ reduction_factor: metrics::RatioMetrics,
+}
+
+/// Hash aggregation uses a 2-stage (partial and final) hash aggregation, this stream
+/// is for the final stage.
+///
+/// See [`InitialPartialHashAggregateStream`] for details.
+pub(crate) struct PartialFinalHashAggregateStream {
+ /// Output schema: group columns followed by final aggregate value columns.
+ schema: SchemaRef,
+
+ /// Input batches containing partial aggregate state rows.
+ input: SendableRecordBatchStream,
+
+ /// Hash table state for this aggregate stream.
+ hash_table: AggregateHashTable<PartialFinal>,
+
+ /// Execution metrics shared with the aggregate plan node.
+ baseline_metrics: BaselineMetrics,
+
+ /// Memory reservation for group keys and accumulators.
+ reservation: MemoryReservation,
+}
+
+impl InitialPartialHashAggregateStream {
+ pub fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ ) -> Result<Self> {
+ debug_assert_eq!(agg.mode, super::AggregateMode::Partial);
+ debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+
+ let schema = Arc::clone(&agg.schema);
+ let input = agg.input.execute(partition, Arc::clone(context))?;
+ let batch_size = context.session_config().batch_size();
+ let baseline_metrics = BaselineMetrics::new(&agg.metrics, partition);
+
+ // Preserve the existing aggregate metric surface for this plan node.
+ let _spill_metrics = SpillMetrics::new(&agg.metrics, partition);
+ let reduction_factor = MetricBuilder::new(&agg.metrics)
+ .with_type(metrics::MetricType::Summary)
+ .ratio_metrics("reduction_factor", partition);
+
+ let hash_table = AggregateHashTable::<InitialPartial>::new(
+ agg,
+ partition,
+ Arc::clone(&schema),
+ batch_size,
+ )?;
+
+ let reservation = MemoryConsumer::new(format!(
+ "InitialPartialHashAggregateStream[{partition}]"
+ ))
+ .register(context.memory_pool());
+
+ Ok(Self {
+ schema,
+ input,
+ hash_table,
+ baseline_metrics,
+ reservation,
+ reduction_factor,
+ })
+ }
+}
+
+impl Stream for InitialPartialHashAggregateStream {
+ type Item = Result<RecordBatch>;
+
+ fn poll_next(
+ mut self: std::pin::Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll<Option<Self::Item>> {
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+
+ loop {
+ if self.hash_table.is_done() {
+ let _ = self.reservation.try_resize(0);
+ return Poll::Ready(None);
+ } else if self.hash_table.is_building() {
+ match self.input.poll_next_unpin(cx) {
+ Poll::Pending => return Poll::Pending,
+ Poll::Ready(Some(Ok(batch))) => {
+ let timer = elapsed_compute.timer();
+ self.reduction_factor.add_total(batch.num_rows());
+ let result = self.hash_table.aggregate_batch(&batch);
+ timer.done();
+
+ if let Err(e) = result {
+ return Poll::Ready(Some(Err(e)));
+ }
+
+ // TODO: impl memory-limited aggr, when OOM directly send
+ // partial state to final aggregate stage
+ if let Err(e) =
+ self.reservation.try_resize(self.hash_table.memory_size())
+ {
+ return Poll::Ready(Some(Err(e)));
+ }
+ }
+ Poll::Ready(Some(Err(e))) => {
+ return Poll::Ready(Some(Err(e)));
+ }
+ Poll::Ready(None) => {
+ let input_schema = self.input.schema();
+ self.input = Box::pin(EmptyRecordBatchStream::new(input_schema));
+
+ let timer = elapsed_compute.timer();
+ let result = self.hash_table.start_output();
+ timer.done();
+
+ if let Err(e) = result {
+ return Poll::Ready(Some(Err(e)));
+ }
+ }
+ }
+ } else {
+ let timer = elapsed_compute.timer();
+ let result = self.hash_table.next_output_batch();
+ timer.done();
+
+ match result {
+ Ok(Some(batch)) => {
+ let _ =
+ self.reservation.try_resize(self.hash_table.memory_size());
+ self.reduction_factor.add_part(batch.num_rows());
+ debug_assert!(batch.num_rows() > 0);
+ return Poll::Ready(Some(Ok(
+ batch.record_output(&self.baseline_metrics)
+ )));
+ }
+ Ok(None) => {
+ let _ = self.reservation.try_resize(0);
+ return Poll::Ready(None);
+ }
+ Err(e) => return Poll::Ready(Some(Err(e))),
+ }
+ }
+ }
+ }
+}
+
+impl RecordBatchStream for InitialPartialHashAggregateStream {
+ fn schema(&self) -> SchemaRef {
+ Arc::clone(&self.schema)
+ }
+}
+
+impl PartialFinalHashAggregateStream {
+ pub fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ ) -> Result<Self> {
+ debug_assert!(matches!(
+ agg.mode,
+ super::AggregateMode::Final | super::AggregateMode::FinalPartitioned
+ ));
+ debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+
+ let schema = Arc::clone(&agg.schema);
+ let input = agg.input.execute(partition, Arc::clone(context))?;
+ let batch_size = context.session_config().batch_size();
+ let baseline_metrics = BaselineMetrics::new(&agg.metrics, partition);
+
+ // Preserve the existing aggregate metric surface for this plan node.
+ let _spill_metrics = SpillMetrics::new(&agg.metrics, partition);
+
+ let hash_table = AggregateHashTable::<PartialFinal>::new(
+ agg,
+ partition,
+ Arc::clone(&schema),
+ batch_size,
+ )?;
+
+ let reservation =
+ MemoryConsumer::new(format!("PartialFinalHashAggregateStream[{partition}]"))
+ .register(context.memory_pool());
+
+ Ok(Self {
+ schema,
+ input,
+ hash_table,
+ baseline_metrics,
+ reservation,
+ })
+ }
+}
+
+impl Stream for PartialFinalHashAggregateStream {
+ type Item = Result<RecordBatch>;
+
+ fn poll_next(
Review Comment:
Given these are two separate structs, what is the rationale for keepign them in the same module?
datafusion/physical-plan/src/aggregates/hash_aggregate.rs
Perhaps it would be clearer if we put them in their own modules (could be a follow on PR)
like
datafusion/physical-plan/src/aggregates/streams/partial_final.rs
datafusion/physical-plan/src/aggregates/streams/initial_partial.rs
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622592357
I expect very high-cardinality aggregations to be slower for now, because I plan to address that in a follow-up PR:
- https://github.com/apache/datafusion/pull/11627
I would change the default for the configuration added in https://github.com/apache/datafusion/commit/4a2b907285e3be77eab342353b8c2c6fd947035d to `false`, and enable it again once that feature is restored.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4638780436
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4638635412)
**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-refactor-only
--------------------
Benchmark clickbench_partitioned.json
--------------------
βββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β split-aggr-refactor-only β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 0 β 1.15 / 3.29 Β±4.19 / 11.66 ms β 1.17 / 3.30 Β±4.18 / 11.66 ms β no change β
β QQuery 1 β 12.35 / 12.62 Β±0.18 / 12.90 ms β 12.28 / 12.58 Β±0.17 / 12.80 ms β no change β
β QQuery 2 β 35.85 / 36.13 Β±0.16 / 36.30 ms β 35.26 / 35.67 Β±0.28 / 36.04 ms β no change β
β QQuery 3 β 30.55 / 31.30 Β±0.69 / 32.31 ms β 30.37 / 30.82 Β±0.43 / 31.34 ms β no change β
β QQuery 4 β 224.80 / 229.93 Β±2.62 / 232.00 ms β 223.67 / 229.17 Β±3.32 / 234.13 ms β no change β
β QQuery 5 β 270.71 / 273.88 Β±2.16 / 276.77 ms β 270.24 / 274.28 Β±2.23 / 276.24 ms β no change β
β QQuery 6 β 1.19 / 1.34 Β±0.23 / 1.78 ms β 1.20 / 1.37 Β±0.22 / 1.79 ms β no change β
β QQuery 7 β 13.53 / 13.66 Β±0.09 / 13.79 ms β 13.70 / 13.77 Β±0.08 / 13.92 ms β no change β
β QQuery 8 β 322.60 / 324.37 Β±2.03 / 327.91 ms β 319.60 / 326.35 Β±5.77 / 335.73 ms β no change β
β QQuery 9 β 448.67 / 459.16 Β±10.76 / 476.69 ms β 456.01 / 467.42 Β±12.31 / 490.26 ms β no change β
β QQuery 10 β 70.16 / 70.81 Β±0.52 / 71.63 ms β 69.03 / 69.85 Β±1.10 / 71.99 ms β no change β
β QQuery 11 β 81.99 / 84.56 Β±3.27 / 90.92 ms β 82.40 / 87.48 Β±8.51 / 104.43 ms β no change β
β QQuery 12 β 266.14 / 271.32 Β±4.79 / 278.11 ms β 268.89 / 276.71 Β±5.20 / 284.81 ms β no change β
β QQuery 13 β 378.99 / 391.78 Β±7.25 / 399.54 ms β 366.21 / 386.55 Β±16.99 / 406.91 ms β no change β
β QQuery 14 β 280.23 / 284.62 Β±4.09 / 291.97 ms β 284.95 / 287.74 Β±2.34 / 291.55 ms β no change β
β QQuery 15 β 269.41 / 274.68 Β±4.38 / 280.56 ms β 268.68 / 279.78 Β±9.66 / 294.28 ms β no change β
β QQuery 16 β 615.97 / 625.48 Β±7.13 / 634.23 ms β 616.68 / 627.90 Β±8.21 / 641.71 ms β no change β
β QQuery 17 β 622.72 / 629.55 Β±5.66 / 637.71 ms β 620.84 / 635.15 Β±10.70 / 653.69 ms β no change β
β QQuery 18 β 1274.02 / 1300.46 Β±23.79 / 1341.58 ms β 1282.85 / 1291.65 Β±10.80 / 1310.49 ms β no change β
β QQuery 19 β 27.55 / 33.00 Β±9.86 / 52.71 ms β 27.68 / 30.86 Β±6.24 / 43.34 ms β +1.07x faster β
β QQuery 20 β 517.58 / 538.23 Β±23.49 / 582.56 ms β 522.46 / 525.50 Β±1.94 / 527.97 ms β no change β
β QQuery 21 β 591.33 / 601.12 Β±6.03 / 609.20 ms β 591.50 / 599.71 Β±7.11 / 612.48 ms β no change β
β QQuery 22 β 1071.52 / 1082.50 Β±10.44 / 1100.28 ms β 1067.36 / 1081.15 Β±7.72 / 1090.70 ms β no change β
β QQuery 23 β 3200.18 / 3224.67 Β±22.12 / 3256.89 ms β 3201.30 / 3239.12 Β±23.40 / 3274.69 ms β no change β
β QQuery 24 β 41.82 / 43.05 Β±1.33 / 45.26 ms β 41.92 / 45.13 Β±5.72 / 56.56 ms β no change β
β QQuery 25 β 112.16 / 113.49 Β±1.49 / 116.32 ms β 111.82 / 115.63 Β±4.65 / 124.46 ms β no change β
β QQuery 26 β 42.37 / 44.43 Β±2.42 / 49.16 ms β 42.30 / 44.36 Β±2.85 / 49.95 ms β no change β
β QQuery 27 β 673.82 / 679.16 Β±4.69 / 686.99 ms β 670.43 / 675.46 Β±4.22 / 681.11 ms β no change β
β QQuery 28 β 3051.07 / 3081.90 Β±21.64 / 3117.92 ms β 3041.49 / 3078.23 Β±27.33 / 3125.45 ms β no change β
β QQuery 29 β 40.20 / 44.44 Β±7.78 / 59.99 ms β 39.92 / 51.72 Β±14.77 / 75.89 ms β 1.16x slower β
β QQuery 30 β 300.16 / 308.18 Β±6.73 / 318.91 ms β 300.20 / 311.45 Β±9.24 / 328.04 ms β no change β
β QQuery 31 β 292.58 / 299.88 Β±5.17 / 307.59 ms β 285.13 / 299.72 Β±16.19 / 330.22 ms β no change β
β QQuery 32 β 969.81 / 1015.61 Β±38.01 / 1061.22 ms β 923.66 / 951.12 Β±22.32 / 986.02 ms β +1.07x faster β
β QQuery 33 β 1468.90 / 1482.04 Β±8.28 / 1489.50 ms β 1455.28 / 1480.79 Β±17.54 / 1499.63 ms β no change β
β QQuery 34 β 1480.01 / 1492.98 Β±11.80 / 1511.18 ms β 1459.14 / 1491.11 Β±40.55 / 1561.45 ms β no change β
β QQuery 35 β 274.04 / 299.87 Β±26.10 / 348.39 ms β 276.70 / 305.36 Β±46.29 / 397.46 ms β no change β
β QQuery 36 β 66.15 / 80.04 Β±16.46 / 102.96 ms β 68.58 / 78.26 Β±13.46 / 104.84 ms β no change β
β QQuery 37 β 35.27 / 36.24 Β±0.97 / 37.89 ms β 34.89 / 38.26 Β±4.77 / 47.66 ms β 1.06x slower β
β QQuery 38 β 39.04 / 42.79 Β±2.37 / 46.47 ms β 40.20 / 46.53 Β±4.53 / 50.78 ms β 1.09x slower β
β QQuery 39 β 141.52 / 147.44 Β±4.34 / 151.53 ms β 148.34 / 154.70 Β±5.63 / 163.18 ms β no change β
β QQuery 40 β 13.54 / 14.33 Β±0.86 / 15.91 ms β 13.76 / 17.95 Β±8.00 / 33.94 ms β 1.25x slower β
β QQuery 41 β 13.32 / 15.96 Β±5.09 / 26.14 ms β 13.50 / 14.02 Β±0.39 / 14.38 ms β +1.14x faster β
β QQuery 42 β 12.82 / 12.98 Β±0.11 / 13.09 ms β 12.92 / 13.99 Β±1.89 / 17.76 ms β 1.08x slower β
βββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 20053.28ms β
β Total Time (split-aggr-refactor-only) β 20027.66ms β
β Average Time (HEAD) β 466.36ms β
β Average Time (split-aggr-refactor-only) β 465.76ms β
β Queries Faster β 3 β
β Queries Slower β 5 β
β Queries with No Change β 35 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**clickbench_partitioned β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 105.0s |
| Peak memory | 30.5 GiB |
| Avg memory | 22.8 GiB |
| CPU user | 1027.8s |
| CPU sys | 76.7s |
| Peak spill | 0 B |
**clickbench_partitioned β branch**
| Metric | Value |
|--------|-------|
| Wall time | 105.0s |
| Peak memory | 30.0 GiB |
| Avg memory | 22.7 GiB |
| CPU user | 1024.3s |
| CPU sys | 77.0s |
| 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622542859
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622519035)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4622519035-439-rf7jf 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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-refactor-only (bd75229ee6f492e63e802b3cb7c76aee792b85f1) to 48adae4 (merge-base) [diff](https://github.com/apache/datafusion/compare/48adae4a3db1bdefcd87a12996abd5e25b506a6f..bd75229ee6f492e63e802b3cb7c76aee792b85f1) 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622459179
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622443332)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4622443332-436-rmmwl 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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-refactor-only (09462b9f7b8ea1c3fcccc0b2966b3163470734f4) to 48adae4 (merge-base) [diff](https://github.com/apache/datafusion/compare/48adae4a3db1bdefcd87a12996abd5e25b506a6f..09462b9f7b8ea1c3fcccc0b2966b3163470734f4) 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622481531
Benchmark for [this request](https://github.com/apache/datafusion/pull/22729#issuecomment-4622443332) failed.
Last 20 lines of output:
<details><summary>Click to expand</summary>
```
Compiling datafusion-functions-window v53.1.0 (/workspace/datafusion-branch/datafusion/functions-window)
Compiling datafusion-physical-plan v53.1.0 (/workspace/datafusion-branch/datafusion/physical-plan)
Compiling datafusion-physical-expr-adapter v53.1.0 (/workspace/datafusion-branch/datafusion/physical-expr-adapter)
warning: ignoring -C extra-filename flag due to -o flag
error[E0428]: the name `Partial` is defined multiple times
--> datafusion/physical-plan/src/aggregates/hash_table.rs:42:1
|
40 | pub(super) struct Partial;
| -------------------------- previous definition of the type `Partial` here
41 | /// Marker for partial state -> final value aggregation.
42 | pub(super) struct Partial;
| ^^^^^^^^^^^^^^^^^^^^^^^^^^ `Partial` redefined here
|
= note: `Partial` must be defined only once in the type namespace of this module
For more information about this error, try `rustc --explain E0428`.
warning: `datafusion-physical-plan` (lib) generated 1 warning
error: could not compile `datafusion-physical-plan` (lib) due to 1 previous error; 1 warning emitted
warning: build failed, waiting for other jobs to finish...
```
</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: Split hash aggregation logic into separated streams [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622619050
> The purpose of this refactor is to make that boundary explicit. With this split, the key blocked state-management changes (Accumulator + GroupValues) should only interact with the new dedicated stream, rather than being wired into the complex state machine in row_hash.rs.
Make sense. Currently , it is actually very very hard to understand the state machine in `row_hash.rs`...
> Yes I agree. I was thinking maybe https://github.com/apache/datafusion/pull/15591 can be updated after this refactor, if we can ensure the blocked Accumulator and GroupValues can only get planned into the new aggregate streams in this refactor, and the old row_hash.rs won't get affected, that change would be easier to review.
Sounds good, #15591 still not apply the blocked logic into splling path and others, and I think really hard if we do it in the old logic...
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#discussion_r3356023077
##########
datafusion/physical-plan/src/aggregates/hash_aggregate.rs:
##########
@@ -0,0 +1,349 @@
+// 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.
+
+//! Grouped hash aggregation for simple multi-stage aggregation paths.
+//!
+//! This module handles the basic grouped two-stage paths:
+//!
+//! ```text
+//! input rows -> GROUP BY hash table -> accumulator state rows
+//! state rows -> GROUP BY hash table -> final aggregate rows
+//! ```
+//!
+//! `AggregateExec` keeps finite-memory, ordered, limit, grouping-set,
+//! `partial state -> partial state`, and single-stage aggregation on
+//! `GroupedHashAggregateStream` for now.
+
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::RecordBatch;
+use datafusion_common::Result;
+use datafusion_execution::TaskContext;
+use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use futures::stream::{Stream, StreamExt};
+
+use super::AggregateExec;
+use super::hash_table::{AggregateHashTable, InitialPartial, PartialFinal};
+use crate::metrics::{BaselineMetrics, MetricBuilder, RecordOutput, SpillMetrics};
+use crate::stream::EmptyRecordBatchStream;
+use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream, metrics};
+
+/// Hash aggregation uses a 2-stage (partial and final) hash aggregation, this stream
+/// is for the partial stage.
+///
+/// # Example
+///
+/// select k, avg(v) from t group by k;
+///
+/// ## Plan
+/// AggregateExec(stage=final)
+/// -- RepartitionExec(hash(k))
+/// ---- AggregateExec(stage=partial)
+///
+/// ## Partial Stage Behavior
+/// Input: raw rows
+/// Output: partial states for all groups (e.g. for avg(x), it's sum(x), count(x))
+///
+/// ## Final Stage Behavior
+/// Input: partial states
+/// Output: results for all groups (e.g. for avg(x), it's avg(x) calculated from the state)
+pub(crate) struct InitialPartialHashAggregateStream {
+ /// Output schema: group columns followed by partial aggregate state columns.
+ schema: SchemaRef,
+
+ /// Input batches containing raw rows, not partial aggregate state.
+ input: SendableRecordBatchStream,
+
+ /// Hash table state for this aggregate stream.
+ hash_table: AggregateHashTable<InitialPartial>,
+
+ /// Memory reservation for group keys and accumulators.
+ reservation: MemoryReservation,
+
+ /// Execution metrics shared with the aggregate plan node.
+ baseline_metrics: BaselineMetrics,
+
+ /// Tracks partial aggregation row reduction, matching `GroupedHashAggregateStream`.
+ reduction_factor: metrics::RatioMetrics,
+}
+
+/// Hash aggregation uses a 2-stage (partial and final) hash aggregation, this stream
+/// is for the final stage.
+///
+/// See [`InitialPartialHashAggregateStream`] for details.
+pub(crate) struct PartialFinalHashAggregateStream {
Review Comment:
I think it might be clearer to call them `PartialHashAggregateStream` and `FinalHashAggregateStream` π€. I'll update the names.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622541173
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622519035)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4622519035-438-xhr4s 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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-refactor-only (bd75229ee6f492e63e802b3cb7c76aee792b85f1) to 48adae4 (merge-base) [diff](https://github.com/apache/datafusion/compare/48adae4a3db1bdefcd87a12996abd5e25b506a6f..bd75229ee6f492e63e802b3cb7c76aee792b85f1) 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622433576
> The idea is incremental migration: we have to keep some duplication during the migration, but eventually the old GroupedHashAggregateStream can be deleted entirely.
I see -- the plan for eventually removing `GroupedHashAggregateStream` sounds good and makes sense
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on code in PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#discussion_r3359869455
##########
datafusion/physical-plan/src/aggregates/hash_table.rs:
##########
@@ -0,0 +1,615 @@
+// Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
Will clearer to organize it like instead of single `hash_table.rs`?
```
aggregator
- mod.rs
- partial.rs
- final.rs
```
##########
datafusion/physical-plan/src/aggregates/hash_table.rs:
##########
@@ -0,0 +1,615 @@
+// 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.
+
+use std::collections::HashMap;
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, AsArray, BooleanArray, new_null_array};
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::RecordBatch;
+use datafusion_common::{Result, internal_err};
+use datafusion_execution::memory_pool::proxy::VecAllocExt;
+use datafusion_expr::{EmitTo, GroupsAccumulator};
+
+use super::group_values::{GroupByMetrics, GroupValues, new_group_values};
+use super::order::GroupOrdering;
+use super::row_hash::create_group_accumulator;
+use super::{
+ AggregateExec, PhysicalGroupBy, aggregate_expressions, evaluate_group_by,
+ group_id_array, max_duplicate_ordinal,
+};
+use crate::PhysicalExpr;
+use crate::metrics::{MetricBuilder, MetricCategory};
+
+/// Marker for raw rows -> partial state aggregation.
+pub(super) struct Partial;
+/// Marker for partial state -> final value aggregation.
+pub(super) struct Final;
+
+/// Grouped hash table shared by the initial-partial and partial-final paths.
+///
+/// While building, it consumes input batches and updates group / accumulator
+/// state. While outputting, it incrementally output the materialized batches.
+///
+/// # Marker Type
+/// `AggrMode` is a zero-sized marker type used with `PhantomData` to keep the
+/// initial-partial and partial-final update logic in separate impl blocks. Different
+/// stages has different semantics and applicable optimizations, this makes the
+/// implementation easier to follow.
+pub(super) struct AggregateHashTable<AggrMode> {
Review Comment:
Seems `AggrMode` is not actually used.
I think it may be clearer like that?
- define `AggregatorCore`, and place the common methods in it
- define `PartialAggregator` and `FinalAggregator` to wrap `AggregatorCore`, and place the specific methods in them
##########
datafusion/sqllogictest/test_files/information_schema.slt:
##########
@@ -370,6 +371,7 @@ datafusion.execution.batch_size 8192 Default batch size while creating new batch
datafusion.execution.coalesce_batches true When set to true, record batches will be examined between each operator and small batches will be coalesced into larger batches. This is helpful when there are highly selective filters or joins that could produce tiny output batches. The target batch size is determined by the configuration setting
datafusion.execution.collect_statistics true Should DataFusion collect statistics when first creating a table. Has no effect after the table is created. Applies to the default `ListingTableProvider` in DataFusion. Defaults to true.
datafusion.execution.enable_ansi_mode false Whether to enable ANSI SQL mode. The flag is experimental and relevant only for DataFusion Spark built-in functions When `enable_ansi_mode` is set to `true`, the query engine follows ANSI SQL semantics for expressions, casting, and error handling. This means: - **Strict type coercion rules:** implicit casts between incompatible types are disallowed. - **Standard SQL arithmetic behavior:** operations such as division by zero, numeric overflow, or invalid casts raise runtime errors rather than returning `NULL` or adjusted values. - **Consistent ANSI behavior** for string concatenation, comparisons, and `NULL` handling. When `enable_ansi_mode` is `false` (the default), the engine uses a more permissive, non-ANSI mode designed for user convenience and backward compatibility. In this mode: - Implicit casts between types are allowed (e.g., string to integer when possible). - Arithmetic operations are more lenient β for example, `abs()` on
the minimum representable integer value returns the input value instead of raising overflow. - Division by zero or invalid casts may return `NULL` instead of failing. # Default `false` β ANSI SQL mode is disabled by default.
+datafusion.execution.enable_migration_aggregate true Temporary switch for aggregate stream implementations that are being migrated from `GroupedHashAggregateStream`. When set to true, DataFusion tries the migrated implementations when their preconditions are satisfied. When set to false, grouped aggregation falls back to `GroupedHashAggregateStream`. This option will be removed after the migration is finished. See <https://github.com/apache/datafusion/issues/22710> for details.
Review Comment:
Seems we can also add fuzzy case for it in `/Users/ruiqiucao/Desktop/github/datafusion/datafusion/core/tests/fuzz_cases/aggregate_fuzz.rs` based on this config option.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] refactor: Split hash aggregation logic into separated streams [datafusion]
Posted by "Rachelint (via GitHub)" <gi...@apache.org>.
Rachelint commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4618544977
> The existing challenge is that the current implementation is hard to extend and review. I want to clean things up through this refactor first, and then apply the actual change.
Current one seems the refactor of original `row_hash.rs`?
But in #155, code changes in `row_hash.rs` actually very few(154 lines), and main code changes are blocked version implementation of `accumulators` and `group values`, and test codes(3000+ lines).
I totally agree with we should refactor code in `row_hash.rs`. It is very very messy and hard to maintain as I complained before, due to we mix the `partial` and `final` logic in a `single stream`.
But seems it may not help very much to reduce codes about supporting blocked memory management?
I think maybe #15591 and refactor can be pushed forward in parallel, they don't have very much in common.
As said above only few codes are for supporting blocked mode in the aggr stream, large codes are for blocked version `accumulators` and `group values`.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "ariel-miculas (via GitHub)" <gi...@apache.org>.
ariel-miculas commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4611260579
I'm curious about the high-level vision: is the plan to close https://github.com/apache/datafusion/pull/15591 in favor of this new approach?
I would like the redesign of hash aggregation to take into account the memory constraints imposed by the finite memory pool, i.e. how does the implementation perform under OOM conditions.
* how do we improve memory accounting (see https://github.com/apache/datafusion/issues/22526).
* how do we avoid excessive memory allocations during OOM condition (see https://github.com/apache/datafusion/pull/22165)
* other issues such as https://github.com/apache/datafusion/issues/19906
Otherwise we'll end up with the same issues that exist now. E.g. EmitTo::First(n) wasn't designed for emitting a large portion of the existing groups, so it over-allocated when used for emitting early in partial aggregation OOM case.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622656053
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622519035)
**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-refactor-only
--------------------
Benchmark tpch_sf1.json
--------------------
βββββββββββββ³βββββββββββββββββββββββββββββββββ³βββββββββββββββββββββββββββββββββ³βββββββββββββββ
β Query β HEAD β split-aggr-refactor-only β Change β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 1 β 38.26 / 39.66 Β±1.63 / 42.79 ms β 38.24 / 38.80 Β±1.05 / 40.89 ms β no change β
β QQuery 2 β 18.55 / 18.97 Β±0.34 / 19.38 ms β 18.27 / 18.63 Β±0.24 / 18.97 ms β no change β
β QQuery 3 β 33.45 / 33.65 Β±0.22 / 34.06 ms β 31.05 / 32.26 Β±1.16 / 33.64 ms β no change β
β QQuery 4 β 17.11 / 17.59 Β±0.80 / 19.19 ms β 17.08 / 17.46 Β±0.50 / 18.45 ms β no change β
β QQuery 5 β 37.39 / 40.25 Β±1.72 / 42.68 ms β 39.33 / 40.00 Β±0.45 / 40.74 ms β no change β
β QQuery 6 β 15.90 / 16.56 Β±0.91 / 18.36 ms β 15.98 / 16.36 Β±0.39 / 17.11 ms β no change β
β QQuery 7 β 42.96 / 44.09 Β±0.66 / 44.75 ms β 42.92 / 46.15 Β±2.61 / 49.12 ms β no change β
β QQuery 8 β 42.62 / 42.90 Β±0.32 / 43.50 ms β 42.70 / 44.08 Β±1.25 / 46.03 ms β no change β
β QQuery 9 β 48.43 / 48.97 Β±0.68 / 50.22 ms β 48.56 / 49.10 Β±0.75 / 50.54 ms β no change β
β QQuery 10 β 41.65 / 41.90 Β±0.20 / 42.19 ms β 41.89 / 44.12 Β±1.82 / 47.04 ms β 1.05x slower β
β QQuery 11 β 12.86 / 13.23 Β±0.43 / 13.92 ms β 12.69 / 12.93 Β±0.20 / 13.18 ms β no change β
β QQuery 12 β 23.82 / 24.31 Β±0.33 / 24.85 ms β 23.54 / 24.32 Β±0.54 / 24.81 ms β no change β
β QQuery 13 β 32.00 / 33.62 Β±1.20 / 35.22 ms β 33.10 / 34.46 Β±0.98 / 35.62 ms β no change β
β QQuery 14 β 23.42 / 24.10 Β±0.94 / 25.97 ms β 23.52 / 23.66 Β±0.13 / 23.84 ms β no change β
β QQuery 15 β 30.94 / 31.38 Β±0.70 / 32.77 ms β 30.86 / 31.04 Β±0.13 / 31.17 ms β no change β
β QQuery 16 β 14.07 / 14.18 Β±0.07 / 14.26 ms β 13.83 / 14.13 Β±0.21 / 14.38 ms β no change β
β QQuery 17 β 72.42 / 72.64 Β±0.14 / 72.85 ms β 72.52 / 73.74 Β±1.31 / 75.78 ms β no change β
β QQuery 18 β 57.49 / 58.55 Β±0.66 / 59.43 ms β 57.86 / 59.77 Β±2.25 / 64.15 ms β no change β
β QQuery 19 β 32.77 / 32.97 Β±0.30 / 33.55 ms β 32.45 / 32.72 Β±0.23 / 33.12 ms β no change β
β QQuery 20 β 31.77 / 32.07 Β±0.21 / 32.32 ms β 31.80 / 32.30 Β±0.63 / 33.53 ms β no change β
β QQuery 21 β 55.26 / 56.55 Β±1.48 / 59.42 ms β 55.20 / 57.39 Β±1.43 / 59.19 ms β no change β
β QQuery 22 β 13.59 / 13.77 Β±0.12 / 13.97 ms β 13.45 / 13.85 Β±0.24 / 14.18 ms β no change β
βββββββββββββ΄βββββββββββββββββββββββββββββββββ΄βββββββββββββββββββββββββββββββββ΄βββββββββββββββ
βββββββββββββββββββββββββββββββββββββββββββ³βββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 751.90ms β
β Total Time (split-aggr-refactor-only) β 757.24ms β
β Average Time (HEAD) β 34.18ms β
β Average Time (split-aggr-refactor-only) β 34.42ms β
β Queries Faster β 0 β
β Queries Slower β 1 β
β Queries with No Change β 21 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββββββ΄βββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**tpch β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 5.7 GiB |
| Avg memory | 4.9 GiB |
| CPU user | 29.6s |
| CPU sys | 2.1s |
| Peak spill | 0 B |
**tpch β branch**
| Metric | Value |
|--------|-------|
| Wall time | 5.0s |
| Peak memory | 5.7 GiB |
| Avg memory | 4.9 GiB |
| CPU user | 29.6s |
| CPU sys | 2.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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622699441
π€ Benchmark completed (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622519035)
**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-refactor-only
--------------------
Benchmark clickbench_partitioned.json
--------------------
βββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββββββββββββββββββββββββββ³ββββββββββββββββ
β Query β HEAD β split-aggr-refactor-only β Change β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β QQuery 0 β 1.16 / 3.39 Β±4.31 / 12.02 ms β 1.16 / 3.39 Β±4.31 / 12.00 ms β no change β
β QQuery 1 β 12.42 / 12.90 Β±0.25 / 13.14 ms β 12.29 / 12.95 Β±0.36 / 13.24 ms β no change β
β QQuery 2 β 35.64 / 36.03 Β±0.31 / 36.51 ms β 36.04 / 36.40 Β±0.29 / 36.76 ms β no change β
β QQuery 3 β 30.70 / 31.85 Β±0.70 / 32.90 ms β 31.94 / 32.11 Β±0.11 / 32.27 ms β no change β
β QQuery 4 β 236.05 / 238.98 Β±2.67 / 243.54 ms β 241.42 / 248.12 Β±5.92 / 255.84 ms β no change β
β QQuery 5 β 277.09 / 281.52 Β±2.57 / 284.74 ms β 275.29 / 281.55 Β±4.56 / 287.78 ms β no change β
β QQuery 6 β 1.21 / 1.37 Β±0.24 / 1.84 ms β 1.21 / 1.38 Β±0.24 / 1.84 ms β no change β
β QQuery 7 β 13.85 / 13.93 Β±0.05 / 14.01 ms β 14.19 / 16.51 Β±4.18 / 24.87 ms β 1.19x slower β
β QQuery 8 β 335.81 / 346.55 Β±6.17 / 352.53 ms β 336.21 / 341.13 Β±3.24 / 345.70 ms β no change β
β QQuery 9 β 479.27 / 491.95 Β±7.18 / 499.60 ms β 471.71 / 484.51 Β±11.85 / 505.01 ms β no change β
β QQuery 10 β 72.50 / 73.60 Β±0.87 / 74.64 ms β 71.31 / 73.31 Β±1.54 / 75.87 ms β no change β
β QQuery 11 β 84.16 / 85.34 Β±0.76 / 86.34 ms β 83.36 / 83.84 Β±0.46 / 84.71 ms β no change β
β QQuery 12 β 275.38 / 283.85 Β±5.98 / 293.46 ms β 278.29 / 282.77 Β±4.23 / 289.44 ms β no change β
β QQuery 13 β 401.86 / 406.17 Β±6.61 / 419.31 ms β 471.01 / 497.73 Β±26.43 / 546.79 ms β 1.23x slower β
β QQuery 14 β 289.82 / 298.87 Β±5.98 / 305.35 ms β 287.89 / 295.81 Β±4.99 / 301.70 ms β no change β
β QQuery 15 β 286.39 / 289.62 Β±2.37 / 293.03 ms β 277.84 / 289.60 Β±7.48 / 298.88 ms β no change β
β QQuery 16 β 635.92 / 649.20 Β±7.04 / 655.51 ms β 630.17 / 640.83 Β±7.61 / 650.28 ms β no change β
β QQuery 17 β 628.78 / 653.98 Β±15.77 / 676.20 ms β 638.44 / 655.19 Β±18.06 / 682.33 ms β no change β
β QQuery 18 β 1335.41 / 1357.71 Β±17.75 / 1382.16 ms β 1304.71 / 1338.16 Β±27.19 / 1375.36 ms β no change β
β QQuery 19 β 27.71 / 28.04 Β±0.18 / 28.23 ms β 28.39 / 31.31 Β±4.77 / 40.80 ms β 1.12x slower β
β QQuery 20 β 525.01 / 535.77 Β±11.90 / 558.17 ms β 520.49 / 532.89 Β±8.13 / 543.50 ms β no change β
β QQuery 21 β 604.75 / 607.93 Β±2.57 / 612.34 ms β 609.63 / 619.10 Β±7.44 / 630.81 ms β no change β
β QQuery 22 β 1082.63 / 1089.41 Β±4.27 / 1094.27 ms β 1071.98 / 1094.74 Β±13.06 / 1109.99 ms β no change β
β QQuery 23 β 3326.57 / 3362.87 Β±42.03 / 3430.61 ms β 3243.97 / 3289.17 Β±30.34 / 3329.45 ms β no change β
β QQuery 24 β 41.67 / 42.97 Β±1.29 / 45.30 ms β 41.57 / 42.32 Β±0.87 / 43.97 ms β no change β
β QQuery 25 β 114.04 / 116.36 Β±2.52 / 121.21 ms β 112.95 / 115.26 Β±2.00 / 118.63 ms β no change β
β QQuery 26 β 42.70 / 44.36 Β±1.40 / 46.81 ms β 42.41 / 47.11 Β±8.76 / 64.61 ms β 1.06x slower β
β QQuery 27 β 677.42 / 686.60 Β±5.85 / 694.86 ms β 678.46 / 683.48 Β±4.84 / 692.68 ms β no change β
β QQuery 28 β 3084.72 / 3127.28 Β±36.96 / 3189.50 ms β 3083.47 / 3104.88 Β±17.69 / 3135.32 ms β no change β
β QQuery 29 β 40.53 / 42.38 Β±2.81 / 47.97 ms β 40.56 / 54.60 Β±19.67 / 91.78 ms β 1.29x slower β
β QQuery 30 β 314.91 / 318.78 Β±3.24 / 324.22 ms β 307.16 / 315.40 Β±6.41 / 325.44 ms β no change β
β QQuery 31 β 293.31 / 300.55 Β±7.77 / 315.18 ms β 350.77 / 358.28 Β±7.19 / 367.42 ms β 1.19x slower β
β QQuery 32 β 1009.72 / 1038.56 Β±19.96 / 1070.19 ms β 1646.84 / 1748.45 Β±52.59 / 1798.12 ms β 1.68x slower β
β QQuery 33 β 1530.63 / 1564.53 Β±29.09 / 1612.73 ms β 1496.35 / 1536.14 Β±26.47 / 1570.37 ms β no change β
β QQuery 34 β 1549.38 / 1592.19 Β±33.91 / 1650.26 ms β 1521.74 / 1554.32 Β±26.48 / 1595.89 ms β no change β
β QQuery 35 β 313.73 / 330.30 Β±22.47 / 374.09 ms β 288.14 / 320.48 Β±38.78 / 388.39 ms β no change β
β QQuery 36 β 72.61 / 77.57 Β±4.14 / 82.40 ms β 68.39 / 79.48 Β±10.82 / 98.66 ms β no change β
β QQuery 37 β 36.20 / 38.40 Β±3.02 / 44.23 ms β 35.82 / 38.34 Β±3.08 / 44.19 ms β no change β
β QQuery 38 β 41.57 / 52.54 Β±7.62 / 61.42 ms β 40.51 / 44.58 Β±3.42 / 50.23 ms β +1.18x faster β
β QQuery 39 β 149.20 / 155.30 Β±5.80 / 162.64 ms β 134.70 / 152.30 Β±9.62 / 163.92 ms β no change β
β QQuery 40 β 14.64 / 19.45 Β±9.38 / 38.21 ms β 14.42 / 15.26 Β±0.94 / 17.01 ms β +1.27x faster β
β QQuery 41 β 13.76 / 14.24 Β±0.27 / 14.56 ms β 13.87 / 14.07 Β±0.13 / 14.25 ms β no change β
β QQuery 42 β 13.71 / 15.15 Β±2.42 / 19.97 ms β 12.79 / 14.27 Β±2.16 / 18.55 ms β +1.06x faster β
βββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββββββββββββββββββββββββββ΄ββββββββββββββββ
βββββββββββββββββββββββββββββββββββββββββββ³βββββββββββββ
β Benchmark Summary β β
β‘βββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β Total Time (HEAD) β 20758.32ms β
β Total Time (split-aggr-refactor-only) β 21421.51ms β
β Average Time (HEAD) β 482.75ms β
β Average Time (split-aggr-refactor-only) β 498.17ms β
β Queries Faster β 3 β
β Queries Slower β 7 β
β Queries with No Change β 33 β
β Queries with Failure β 0 β
βββββββββββββββββββββββββββββββββββββββββββ΄βββββββββββββ
```
</p>
</details>
<details><summary>Resource Usage</summary>
**clickbench_partitioned β base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 105.0s |
| Peak memory | 30.0 GiB |
| Avg memory | 22.9 GiB |
| CPU user | 1060.8s |
| CPU sys | 81.9s |
| Peak spill | 0 B |
**clickbench_partitioned β branch**
| Metric | Value |
|--------|-------|
| Wall time | 110.0s |
| Peak memory | 35.8 GiB |
| Avg memory | 23.5 GiB |
| CPU user | 1098.6s |
| CPU sys | 84.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: Split hash aggregation logic into separated streams [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4621755262
I love the idea of refactoring to reduce the cognative load, BTW
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4629802758
For Clickbench result:
I tried to reproduce locally, and confirmed Q13, Q26, Q31, Q32 are slower.
I looked at the `explain analyze`, Q13, Q31, Q32 are expected due to https://github.com/apache/datafusion/pull/22729#issuecomment-4622592357
Q26 don't have aggregation, probably because this branch is behind main for some optimizations π€
If that look good, I will by default disable this split path, and enable it after #11627 is back, to prevent performance regression.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4638504482
@alamb @Rachelint Thanks for the review, it's ready for another look.
- Comments are addressed, except file split: there are more hash aggregation variant (like fused partial+final), so I think I can have a better idea to organize files when they're all done
- Regarding bench numbers, I think it's expected: https://github.com/apache/datafusion/pull/22729#issuecomment-4629802758
- Updated the EPIC for future follow-ups 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4638645164
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4638635412)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4638635412-463-qnz7q 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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-refactor-only (29d132bdd649e27c5925026f4287daa39ed84e96) to 48adae4 (merge-base) [diff](https://github.com/apache/datafusion/compare/48adae4a3db1bdefcd87a12996abd5e25b506a6f..29d132bdd649e27c5925026f4287daa39ed84e96) 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#discussion_r3347323349
##########
datafusion/physical-plan/src/aggregates/hash_aggregate.rs:
##########
@@ -0,0 +1,349 @@
+// 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.
+
+//! Grouped hash aggregation for simple multi-stage aggregation paths.
+//!
+//! This module handles the basic grouped two-stage paths:
+//!
+//! ```text
+//! input rows -> GROUP BY hash table -> accumulator state rows
+//! state rows -> GROUP BY hash table -> final aggregate rows
+//! ```
+//!
+//! `AggregateExec` keeps finite-memory, ordered, limit, grouping-set,
+//! `partial state -> partial state`, and single-stage aggregation on
+//! `GroupedHashAggregateStream` for now.
+
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use arrow::datatypes::SchemaRef;
+use arrow::record_batch::RecordBatch;
+use datafusion_common::Result;
+use datafusion_execution::TaskContext;
+use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation};
+use futures::stream::{Stream, StreamExt};
+
+use super::AggregateExec;
+use super::hash_table::{AggregateHashTable, InitialPartial, PartialFinal};
+use crate::metrics::{BaselineMetrics, MetricBuilder, RecordOutput, SpillMetrics};
+use crate::stream::EmptyRecordBatchStream;
+use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream, metrics};
+
+/// Hash aggregation uses a 2-stage (partial and final) hash aggregation, this stream
+/// is for the partial stage.
+///
+/// # Example
+///
+/// select k, avg(v) from t group by k;
+///
+/// ## Plan
+/// AggregateExec(stage=final)
+/// -- RepartitionExec(hash(k))
+/// ---- AggregateExec(stage=partial)
+///
+/// ## Partial Stage Behavior
+/// Input: raw rows
+/// Output: partial states for all groups (e.g. for avg(x), it's sum(x), count(x))
+///
+/// ## Final Stage Behavior
+/// Input: partial states
+/// Output: results for all groups (e.g. for avg(x), it's avg(x) calculated from the state)
+pub(crate) struct InitialPartialHashAggregateStream {
+ /// Output schema: group columns followed by partial aggregate state columns.
+ schema: SchemaRef,
+
+ /// Input batches containing raw rows, not partial aggregate state.
+ input: SendableRecordBatchStream,
+
+ /// Hash table state for this aggregate stream.
+ hash_table: AggregateHashTable<InitialPartial>,
+
+ /// Memory reservation for group keys and accumulators.
+ reservation: MemoryReservation,
+
+ /// Execution metrics shared with the aggregate plan node.
+ baseline_metrics: BaselineMetrics,
+
+ /// Tracks partial aggregation row reduction, matching `GroupedHashAggregateStream`.
+ reduction_factor: metrics::RatioMetrics,
+}
+
+/// Hash aggregation uses a 2-stage (partial and final) hash aggregation, this stream
+/// is for the final stage.
+///
+/// See [`InitialPartialHashAggregateStream`] for details.
+pub(crate) struct PartialFinalHashAggregateStream {
+ /// Output schema: group columns followed by final aggregate value columns.
+ schema: SchemaRef,
+
+ /// Input batches containing partial aggregate state rows.
+ input: SendableRecordBatchStream,
+
+ /// Hash table state for this aggregate stream.
+ hash_table: AggregateHashTable<PartialFinal>,
+
+ /// Execution metrics shared with the aggregate plan node.
+ baseline_metrics: BaselineMetrics,
+
+ /// Memory reservation for group keys and accumulators.
+ reservation: MemoryReservation,
+}
+
+impl InitialPartialHashAggregateStream {
+ pub fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ ) -> Result<Self> {
+ debug_assert_eq!(agg.mode, super::AggregateMode::Partial);
+ debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+
+ let schema = Arc::clone(&agg.schema);
+ let input = agg.input.execute(partition, Arc::clone(context))?;
+ let batch_size = context.session_config().batch_size();
+ let baseline_metrics = BaselineMetrics::new(&agg.metrics, partition);
+
+ // Preserve the existing aggregate metric surface for this plan node.
+ let _spill_metrics = SpillMetrics::new(&agg.metrics, partition);
+ let reduction_factor = MetricBuilder::new(&agg.metrics)
+ .with_type(metrics::MetricType::Summary)
+ .ratio_metrics("reduction_factor", partition);
+
+ let hash_table = AggregateHashTable::<InitialPartial>::new(
+ agg,
+ partition,
+ Arc::clone(&schema),
+ batch_size,
+ )?;
+
+ let reservation = MemoryConsumer::new(format!(
+ "InitialPartialHashAggregateStream[{partition}]"
+ ))
+ .register(context.memory_pool());
+
+ Ok(Self {
+ schema,
+ input,
+ hash_table,
+ baseline_metrics,
+ reservation,
+ reduction_factor,
+ })
+ }
+}
+
+impl Stream for InitialPartialHashAggregateStream {
+ type Item = Result<RecordBatch>;
+
+ fn poll_next(
+ mut self: std::pin::Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll<Option<Self::Item>> {
+ let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
+
+ loop {
+ if self.hash_table.is_done() {
+ let _ = self.reservation.try_resize(0);
+ return Poll::Ready(None);
+ } else if self.hash_table.is_building() {
+ match self.input.poll_next_unpin(cx) {
+ Poll::Pending => return Poll::Pending,
+ Poll::Ready(Some(Ok(batch))) => {
+ let timer = elapsed_compute.timer();
+ self.reduction_factor.add_total(batch.num_rows());
+ let result = self.hash_table.aggregate_batch(&batch);
+ timer.done();
+
+ if let Err(e) = result {
+ return Poll::Ready(Some(Err(e)));
+ }
+
+ // TODO: impl memory-limited aggr, when OOM directly send
+ // partial state to final aggregate stage
+ if let Err(e) =
+ self.reservation.try_resize(self.hash_table.memory_size())
+ {
+ return Poll::Ready(Some(Err(e)));
+ }
+ }
+ Poll::Ready(Some(Err(e))) => {
+ return Poll::Ready(Some(Err(e)));
+ }
+ Poll::Ready(None) => {
+ let input_schema = self.input.schema();
+ self.input = Box::pin(EmptyRecordBatchStream::new(input_schema));
+
+ let timer = elapsed_compute.timer();
+ let result = self.hash_table.start_output();
+ timer.done();
+
+ if let Err(e) = result {
+ return Poll::Ready(Some(Err(e)));
+ }
+ }
+ }
+ } else {
+ let timer = elapsed_compute.timer();
+ let result = self.hash_table.next_output_batch();
+ timer.done();
+
+ match result {
+ Ok(Some(batch)) => {
+ let _ =
+ self.reservation.try_resize(self.hash_table.memory_size());
+ self.reduction_factor.add_part(batch.num_rows());
+ debug_assert!(batch.num_rows() > 0);
+ return Poll::Ready(Some(Ok(
+ batch.record_output(&self.baseline_metrics)
+ )));
+ }
+ Ok(None) => {
+ let _ = self.reservation.try_resize(0);
+ return Poll::Ready(None);
+ }
+ Err(e) => return Poll::Ready(Some(Err(e))),
+ }
+ }
+ }
+ }
+}
+
+impl RecordBatchStream for InitialPartialHashAggregateStream {
+ fn schema(&self) -> SchemaRef {
+ Arc::clone(&self.schema)
+ }
+}
+
+impl PartialFinalHashAggregateStream {
+ pub fn new(
+ agg: &AggregateExec,
+ context: &Arc<TaskContext>,
+ partition: usize,
+ ) -> Result<Self> {
+ debug_assert!(matches!(
+ agg.mode,
+ super::AggregateMode::Final | super::AggregateMode::FinalPartitioned
+ ));
+ debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+
+ let schema = Arc::clone(&agg.schema);
+ let input = agg.input.execute(partition, Arc::clone(context))?;
+ let batch_size = context.session_config().batch_size();
+ let baseline_metrics = BaselineMetrics::new(&agg.metrics, partition);
+
+ // Preserve the existing aggregate metric surface for this plan node.
+ let _spill_metrics = SpillMetrics::new(&agg.metrics, partition);
+
+ let hash_table = AggregateHashTable::<PartialFinal>::new(
+ agg,
+ partition,
+ Arc::clone(&schema),
+ batch_size,
+ )?;
+
+ let reservation =
+ MemoryConsumer::new(format!("PartialFinalHashAggregateStream[{partition}]"))
+ .register(context.memory_pool());
+
+ Ok(Self {
+ schema,
+ input,
+ hash_table,
+ baseline_metrics,
+ reservation,
+ })
+ }
+}
+
+impl Stream for PartialFinalHashAggregateStream {
+ type Item = Result<RecordBatch>;
+
+ fn poll_next(
Review Comment:
The state machines are identical for now, but in follow-up work, such as skipping partial aggregation for high-cardinality inputs, their control flows will diverge. I think separating them improves clarity, as discussed in https://github.com/apache/datafusion/issues/22710.
Some duplication is inevitable, but that is the trade-off.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4610892906
cc @Dandandan @ariel-miculas @alamb, who have expressed interest before.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4618766692
> But seems it may not help very much to reduce codes about supporting blocked memory management?
That's true, the actual code change is about the same. The major thing get reduced is the cognitive overhead when maintaining this feature, I'll try to explain better:
I think the real pain point is:
1. To review and merge #15591, reviewers need to understand how the actual blocked state change interacts with `row_hash.rs`.
2. `row_hash.rs` is way too complex, so it is hard for reviewers to see the boundary between the actual blocked state-management changes and the existing hash-aggregation state machine.
The purpose of this refactor is to make that boundary explicit. With this split, the key blocked state-management changes (`Accumulator` + `GroupValues`) should only interact with the new dedicated stream, rather than being wired into the complex state machine in `row_hash.rs`.
> I think maybe #15591 and refactor can be pushed forward in parallel, they don't have very much in common.
Yes I agree. I was thinking maybe #15591 can be updated after this refactor, if we can ensure the blocked `Accumulator` and `GroupValues` can only get planned into the new aggregate streams in this refactor, and the old `row_hash.rs` won't get affected, that change would be easier to review.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4638635412
run benchmark clickbench_partitioned
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4621939197
> Thank you @2010YOUY01 -- this looks like a good idea, but I am slightly confused
>
> 1. The PR says it "splits" logic but I actually see two entirely new streams (and no reduction / refactoring of the existing streams)
> 2. I am not sure how the two new streams are related to the existing stream
@alamb Thank you for the review!
The idea is incremental migration: we have to keep some duplication during the migration, but eventually the old `GroupedHashAggregateStream` can be deleted entirely.
```rust
// Before
AggregateExec::execute() {
// One multiplexed stream handles all paths.
GroupedHashAggregateStream::new(...)
}
// Incremental migration process
AggregateExec::execute() {
match self.plan_stream() {
// Migrated path 1/5
StreamPlan::Partial => PartialHashAggregateStream::new(...),
// Migrated path 2/5
StreamPlan::Final => FinalHashAggregateStream::new(...),
// Paths not migrated yet still use the old implementation.
_ => GroupedHashAggregateStream::new(...),
}
}
// After all paths are migrated
AggregateExec::execute() {
match self.plan_stream() {
StreamPlan::Partial => PartialHashAggregateStream::new(...),
StreamPlan::Final => FinalHashAggregateStream::new(...),
// Migrated path 3/5
// ...
// No fallback remains; GroupedHashAggregateStream can be deleted.
}
}
```
The reasons for this approach are:
- Ideally, we would remove the migrated paths from `GroupedHashAggregateStream` as we go. However, I have found that code particularly difficult to modify safely.
- This incremental migration minimizes risk by keeping the existing implementation as a fallback while new paths are introduced. Once all paths have been migrated, we can delete the old implementation and validate the result with the existing test suite. The main challenge is that it is difficult to test each migrated path in isolation, and I can't think about a good idea to address that, so I expect the final deletion step to require careful review and validation.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#discussion_r3356042629
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -499,6 +506,8 @@ impl PartialEq for PhysicalGroupBy {
#[expect(clippy::large_enum_variant)]
enum StreamType {
AggregateStream(AggregateStream),
+ InitialPartialHash(InitialPartialHashAggregateStream),
+ PartialFinalHash(PartialFinalHashAggregateStream),
GroupedHash(GroupedHashAggregateStream),
Review Comment:
Added comments in [3184e00](https://github.com/apache/datafusion/pull/22729/commits/3184e0044c1468a4660f5c89063253d47077b344) for 1. input/output schema 2. Incremental migration strategy for why `GroupsHashAggregateStream` hasn't changed.
> For example is `GroupedHash` the same as InitialPartialHash?
`GroupedHash` is a reused path for `InitialPartial` and `ParitalFinal` and many others, this might explain why changing things inside `GroupedHash` is very hard.
--
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622457966
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622443332)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4622443332-435-vtqjg 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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-refactor-only (09462b9f7b8ea1c3fcccc0b2966b3163470734f4) to 48adae4 (merge-base) [diff](https://github.com/apache/datafusion/compare/48adae4a3db1bdefcd87a12996abd5e25b506a6f..09462b9f7b8ea1c3fcccc0b2966b3163470734f4) 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622443332
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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622547678
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622519035)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4622519035-440-24rrh 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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-refactor-only (bd75229ee6f492e63e802b3cb7c76aee792b85f1) to 48adae4 (merge-base) [diff](https://github.com/apache/datafusion/compare/48adae4a3db1bdefcd87a12996abd5e25b506a6f..bd75229ee6f492e63e802b3cb7c76aee792b85f1) 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: Split hash aggregation logic into separated streams [datafusion]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #22729:
URL: https://github.com/apache/datafusion/pull/22729#issuecomment-4622469509
π€ Benchmark running (GKE) | [trigger](https://github.com/apache/datafusion/pull/22729#issuecomment-4622443332)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4622443332-437-lhw7s 6.12.68+ #1 SMP Wed Apr 1 02:23:28 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-refactor-only (09462b9f7b8ea1c3fcccc0b2966b3163470734f4) to 48adae4 (merge-base) [diff](https://github.com/apache/datafusion/compare/48adae4a3db1bdefcd87a12996abd5e25b506a6f..09462b9f7b8ea1c3fcccc0b2966b3163470734f4) 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