avantgardnerio commented on code in PR #7192:
URL: https://github.com/apache/arrow-datafusion/pull/7192#discussion_r1302014199
##########
datafusion/core/src/physical_plan/aggregates/mod.rs:
##########
@@ -717,14 +725,38 @@ impl AggregateExec {
partition: usize,
context: Arc<TaskContext>,
) -> Result<StreamType> {
+ // no group by at all
if self.group_by.expr.is_empty() {
- Ok(StreamType::AggregateStream(AggregateStream::new(
+ return Ok(StreamType::AggregateStream(AggregateStream::new(
self, context, partition,
- )?))
+ )?));
+ }
+
+ // grouping by an expression that has a sort/limit upstream
+ if let Some(limit) = self.limit {
+ return Ok(StreamType::GroupedPriorityQueue(
+ GroupedTopKAggregateStream::new(self, context, partition,
limit)?,
+ ));
+ }
+
+ // grouping by something else and we need to just materialize all
results
+ Ok(StreamType::GroupedHash(GroupedHashAggregateStream::new(
+ self, context, partition,
+ )?))
+ }
+
+ /// Finds the DataType and SortDirection for this Aggregate, if there is
one
+ pub fn get_minmax_desc(&self) -> Option<(Field, bool)> {
+ let agg_expr = match self.aggr_expr.as_slice() {
+ [expr] => expr,
+ _ => return None,
+ };
+ if let Some(max) = agg_expr.as_any().downcast_ref::<Max>() {
+ Some((max.field().ok()?, true))
+ } else if let Some(min) = agg_expr.as_any().downcast_ref::<Min>() {
+ Some((min.field().ok()?, true))
Review Comment:
> is that intended?
No, and I'm surprised tests are passing, which means I'm missing some -
which I will add then fix.
--
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]
