You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/15 03:12:51 UTC
(datafusion) branch gh-readonly-queue/main/pr-23525-4b68de61d37839ec40f12d3c1ad8995d5106bbf0 created (now 2a563487c1)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-23525-4b68de61d37839ec40f12d3c1ad8995d5106bbf0
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 2a563487c1 doc: More comments to aggregate planning overview (#23525)
This branch includes the following new commits:
new 2a563487c1 doc: More comments to aggregate planning overview (#23525)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: doc: More comments to aggregate planning overview (#23525)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23525-4b68de61d37839ec40f12d3c1ad8995d5106bbf0
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 2a563487c104435e894d7bc048ec06c593da4c8d
Author: Yongting You <20...@gmail.com>
AuthorDate: Wed Jul 15 11:12:28 2026 +0800
doc: More comments to aggregate planning overview (#23525)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
Part of https://github.com/apache/datafusion/issues/22710
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
There are currently around 10 specialized aggregation implementations
(streams such as AggregateStream). We have comments describing each
variant, but no high-level overview. This PR adds one.
It's mostly a aggregate planning overview, and brief intuitions about
existing aggregate streams, plus pointers.
## 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.
-->
## 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
2. 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.
-->
---
datafusion/physical-plan/src/aggregates/mod.rs | 133 ++++++++++++++++++++++++-
1 file changed, 132 insertions(+), 1 deletion(-)
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
index 7ae481e96c..732da32ab0 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -15,7 +15,132 @@
// specific language governing permissions and limitations
// under the License.
-//! Aggregates functionalities
+//! Aggregate functionality
+//!
+//! # Aggregate planning
+//!
+//! DataFusion selects different aggregate implementations (streams) based on the
+//! query shape and configuration. This section provides an overview of the
+//! available stream variants.
+//!
+//! See each stream's documentation for details.
+//!
+//! ## 1. Two-stage hash aggregation
+//!
+//! Two-stage hash aggregation is used for regular parallel execution.
+//!
+//! The input passes through three execution operators to produce the final
+//! aggregation result:
+//!
+//! 1. Partial aggregation reads the input and produces partial states. It
+//! aggregates independently within each partition, which usually reduces
+//! cardinality before the later shuffle.
+//! 2. Hash repartitioning on the group keys sends all partial states for each
+//! group to the same output partition for final aggregation.
+//! 3. Final aggregation reads the partial states, combines them, and emits the
+//! final results.
+//!
+//! ```text
+//! AggregateExec (final)
+//! RepartitionExec (hash by group keys)
+//! AggregateExec (partial)
+//! ```
+//!
+//! See [`PartialHashAggregateStream`] and [`FinalHashAggregateStream`] for details.
+//!
+//! ### Ordering optimization
+//!
+//! When the input is ordered by the group key, an ordered fast path is used. It
+//! uses a similar two-stage hash aggregation with an early-emission optimization.
+//!
+//! ```text
+//! AggregateExec (final, ordered)
+//! RepartitionExec (hash by group keys, order-preserving)
+//! AggregateExec (partial, ordered)
+//! ```
+//!
+//! See [`OrderedPartialAggregateStream`] and [`OrderedFinalAggregateStream`] for
+//! details.
+//!
+//! Related configuration:
+//!
+//! - [`datafusion.execution.target_partitions`](datafusion_common::config::ExecutionOptions::target_partitions)
+//! - [`datafusion.optimizer.repartition_aggregations`](datafusion_common::config::OptimizerOptions::repartition_aggregations)
+//! - [`datafusion.optimizer.prefer_existing_sort`](datafusion_common::config::OptimizerOptions::prefer_existing_sort)
+//!
+//! ## 2. Single-stage hash aggregation
+//!
+//! When there is a single partition, or the aggregation input is already
+//! key-partitioned (e.g., a data source has existing range partitioning),
+//! `Single` mode aggregation is used.
+//!
+//! It takes raw input and directly produces the final result.
+//!
+//! ```text
+//! AggregateExec (mode=Single or SinglePartitioned)
+//! input
+//! ```
+//!
+//! See [`SingleHashAggregateStream`] for details.
+//!
+//! Related configuration:
+//!
+//! - [`datafusion.execution.target_partitions`](datafusion_common::config::ExecutionOptions::target_partitions)
+//! - [`datafusion.optimizer.repartition_aggregations`](datafusion_common::config::OptimizerOptions::repartition_aggregations)
+//!
+//! ## 3. Aggregation without grouping expressions
+//!
+//! A global aggregate maintains one accumulator set per input partition rather
+//! than a hash table of groups. Partial stages compute local states and a final
+//! stage combines them into one output row:
+//!
+//! ```text
+//! AggregateExec (final, no-grouping)
+//! CoalescePartitionsExec
+//! AggregateExec (partial, no-grouping)
+//! ```
+//!
+//! Every stage without grouping expressions uses [`AggregateStream`]. This path
+//! is selected before the grouped-stream migration setting is considered.
+//!
+//! ## 4. Grouped TopK aggregation
+//!
+//! When a query only needs the best `N` groups, retaining every group in a hash
+//! table and sorting them afterward does unnecessary work. The optimizer pushes
+//! the sort limit and direction into the aggregate:
+//!
+//! ```text
+//! SortExec (fetch=N)
+//! AggregateExec (limit=N, order=...)
+//! input
+//! ```
+//!
+//! [`GroupedTopKAggregateStream`] keeps a bounded priority map for a single group
+//! key. It supports group-by-only queries and compatible `MIN` or `MAX`
+//! aggregates. An unordered group-by-only soft limit instead stays on the normal
+//! hash aggregation path.
+//!
+//! Related configuration:
+//!
+//! - [`datafusion.optimizer.enable_topk_aggregation`](datafusion_common::config::OptimizerOptions::enable_topk_aggregation)
+//! - [`datafusion.optimizer.enable_distinct_aggregation_soft_limit`](datafusion_common::config::OptimizerOptions::enable_distinct_aggregation_soft_limit)
+//!
+//! ## 5. Partial-reduce hash aggregation
+//!
+//! This implementation will not be planned by DataFusion SQL interface, it must be
+//! manually constructed at [`ExecutionPlan`] level.
+//!
+//! This mode is useful in a distributed setting.
+//!
+//! See [`PartialReduceHashAggregateStream`] for details.
+//!
+//! ## 6. Fallback grouped hash aggregation
+//!
+//! [`GroupedHashAggregateStream`] is the legacy implementation for several of the
+//! stream types above. It is being incrementally migrated to separate streams.
+//!
+//! See the issue for details: <https://github.com/apache/datafusion/issues/22710>
+#![expect(rustdoc::private_intra_doc_links)]
use std::borrow::Cow;
use std::sync::Arc;
@@ -1035,6 +1160,12 @@ impl AggregateExec {
));
}
+ // Select the stream type based on the query shape and configuration.
+ // For an overview, see the `Aggregate planning` section in this file's
+ // documentation.
+ //
+ // # Implementation Note
+ //
// `GroupedHashAggregateStream` is being incrementally refactored. See the
// tracking issue for details.
//
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org