refactor(hash-aggr): Simplify aggregate hash table with tempated functions#23324
Open
2010YOUY01 wants to merge 2 commits into
Open
refactor(hash-aggr): Simplify aggregate hash table with tempated functions#233242010YOUY01 wants to merge 2 commits into
2010YOUY01 wants to merge 2 commits into
Conversation
Contributor
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing split-aggr-simplify-ht-template (a1c9b6a) to 166c040 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing split-aggr-simplify-ht-template (a1c9b6a) to 166c040 (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing split-aggr-simplify-ht-template (a1c9b6a) to 166c040 (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
Rachelint
approved these changes
Jul 7, 2026
Rachelint
left a comment
Rachelint
left a comment
Contributor
There was a problem hiding this comment.
Thanks @2010YOUY01 , It seems actually clear after removing duplication through this approach
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
Contributor
|
Some thoughts about later refactoring:
What do you think about this idea @2010YOUY01 @alamb |
Contributor
I think it sounds good. I think there are some other major parts too:
Looking at the names of the modules, in this directory I also feel the structure is somewhat lost 
I made a PR to try and name them more consistently: |
alamb
approved these changes
Jul 7, 2026
alamb
left a comment
alamb
left a comment
Contributor
There was a problem hiding this comment.
Thank yoU @2010YOUY01 and @Rachelint -- this looks very nice to me
I have a few stylistic / simplification suggestions, but nothing I think is required or could not be done as a follow on
| /// Each aggregation mode chooses a different `aggregate_fn` according to its | ||
| /// semantics. For example, partial aggregation takes raw inputs, and update them | ||
| /// into stored partial states, so [`GroupsAccumulator::update_batch`] is used. | ||
| pub(super) fn aggregate_batch_inner<F>( |
Contributor
There was a problem hiding this comment.
can you please document somewhere what these arguments are? Specifically the &[usize] and usize arguments?
Something like
/// Function used by [`AggregateHashTable::aggregate_batch_inner`] to update one
/// accumulator with one evaluated input batch.
///
/// Arguments:
/// * accumulator to update.
/// * accumulator's evaluated arguments and optional filter.
/// * `&[usize]` with one group index per input row, mapping each row to its interned group
/// * total number of groups currently interned in that buffer, including newly interned groups
pub(super) type AggregateBatchFn = fn(
&mut AggregateAccumulator,
&EvaluatedAccumulatorArgs,
&[usize],
usize,
) -> Result<()>;So then this would be something like
pub(super) fn aggregate_batch_inner<F: AggregateBatchFn>(
&mut self,
batch: &RecordBatch,
mut aggregate_fn: F
) -> Result<()>
Comment on lines
+72
to
+74
| |acc, values, group_indices, total_num_groups| { | ||
| acc.merge_batch(values, group_indices, total_num_groups) | ||
| }, |
Contributor
There was a problem hiding this comment.
I think you can use the method name directly here
pub(in crate::aggregates) fn aggregate_batch(
&mut self,
batch: &RecordBatch,
) -> Result<()> {
self.aggregate_batch_inner(batch, AggregateAccumulator::merge_batch)
}| self.state = AggregateHashTableState::OutputtingMaterialized(output); | ||
| } | ||
| batch | ||
| self.next_output_batch_inner(|acc, emit_to, output| { |
Contributor
There was a problem hiding this comment.
likewise here I think you can just use the method name directly
self.next_output_batch_inner(Self::materialize_partial_reduce_output)| /// | ||
| /// This is a temporary solution until blocked state management is implemented: | ||
| /// Issue: <https://github.com/apache/datafusion/issues/7065> | ||
| pub(super) fn next_output_batch_inner<F>( |
Contributor
There was a problem hiding this comment.
Similarly, here I think a new typedef would help
/// Function used by [`AggregateHashTable::next_output_batch_inner`] to
/// materialize all currently buffered output for one aggregation mode.
///
/// Arguments
/// * hash table that is producing output
/// *The outputting buffer moved out of the table state
/// * output schema for the materialized [`RecordBatch`].
pub(super) type NextOutputBatchFn<AggrMode> = fn(
&AggregateHashTable<AggrMode>,
AggregateHashTableBuffer,
SchemaRef,
) -> Result<MaterializedAggregateOutput>;| Ok(()) | ||
| self.aggregate_batch_inner( | ||
| batch, | ||
| |acc, values, group_indices, total_num_groups| { |
Contributor
There was a problem hiding this comment.
here too
self.aggregate_batch_inner(batch, AggregateAccumulator::merge_batch)| let mut output = state.group_values.emit(emit_to)?; | ||
|
|
||
| for acc in state.accumulators.iter_mut() { | ||
| self.next_output_batch_inner(|acc, emit_to, output| { |
Contributor
There was a problem hiding this comment.
self.next_output_batch_inner(Self::materialize_partial_output)| } | ||
|
|
||
| Ok(()) | ||
| self.aggregate_batch_inner( |
Contributor
There was a problem hiding this comment.
self.aggregate_batch_inner(batch, HashAggregateAccumulator::update_batch)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Part of [EPIC] Split Aggregation Logic into Dedicated Streams #22710
An alternative for and closes refactor(hash-aggr): Simplify aggregate hash table #23309
Rationale for this change
See #23309 and #23309 (comment) for background.
I prefer this approach, but I want to point out the tradeoff for this PR's approach: the shared utility includes a complex lambda function argument, this makes them harder to extend. But I think it's okay since most functionality has been implemented for the refactor, and there are not likely to have new functional requirements.
What changes are included in this PR?
Are these changes tested?
Are there any user-facing changes?