alamb commented on code in PR #25104:
URL: https://github.com/apache/datafusion/pull/25104#discussion_r4000846435
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1223,143 +1227,76 @@ 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 the legacy implementation of all the
- // branches below. It has been split into dedicated streams, and the
- // `enable_migration_aggregate` config option selects between the new
- // streams and the legacy implementation.
- //
- // See the `grouped_hash_stream` module documentation for the
- // deprecation schedule.
- if context
+ // `GroupedHashAggregateStream` is the legacy implementation of every
+ // grouped path below. It is only planned as an opt-in fallback, see
the
+ // `grouped_hash_stream` module documentation for the deprecation
+ // schedule.
+ if !context
.session_config()
.options()
.execution
.enable_migration_aggregate
{
- if self.should_use_ordered_partial_aggregate_stream(context) {
- return Ok(StreamType::OrderedPartialAggregate(
- OrderedPartialAggregateStream::new(self, context,
partition)?,
- ));
- }
+ return Ok(StreamType::GroupedHash(GroupedHashAggregateStream::new(
+ self, context, partition,
+ )?));
+ }
- if self.should_use_partial_hash_stream(context) {
- return
Ok(StreamType::PartialHash(PartialHashAggregateStream::new(
- self, context, partition,
- )?));
- }
+ // Grouping sets are expanded by the raw-input stages, so the stages
+ // consuming partial state must receive the expanded keys as a plain
+ // group by (see `PhysicalGroupBy::as_final`).
+ if self.mode.input_mode() == AggregateInputMode::Partial
+ && !self.group_by.is_single()
+ {
+ return internal_err!(
+ "Grouping sets must be expanded before {:?} aggregation, which
consumes partial state",
+ self.mode
+ );
+ }
- if self.should_use_partial_reduce_hash_stream(context) {
- return Ok(StreamType::PartialReduceHash(
- PartialReduceHashAggregateStream::new(self, context,
partition)?,
- ));
+ // Choose the execution path based on (aggregation mode, ordering).
+ //
+ // Note that `self.input_order_mode` represents both input ordering
and output
+ // order promise. See its comment for details.
+ use AggregateMode::*;
+ use InputOrderMode::*;
+ let stream = match (self.mode, &self.input_order_mode) {
Review Comment:
I really like how this uses the `match` arm to ensure all combinations are
properly covered
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -888,6 +888,10 @@ pub struct AggregateExec {
metrics: ExecutionPlanMetricsSet,
required_input_ordering: Option<OrderingRequirements>,
/// Describes how the input is ordered relative to the group by columns
+ ///
+ /// This field is also overloaded to mean "the output MUST preserve this
Review Comment:
Do you think we should make a separate flag for output order?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
