ipdates

datafusion/core/src/physical_plan/aggregates/order.rs

@@ -83,7 +83,22 @@ impl GroupOrdering {
 
     /// Returns the index of the last completed group, if any
     pub fn last_completed(&self) -> Option<usize> {
-        todo!()
+        match self.state {
+            OrderState::Start => None,
+            OrderState::InProgress(group_index) if group_index == 0 => None,
+            OrderState::InProgress(group_index) => Some(group_index - 1),
+            OrderState::Complete => todo!(),
+        }
+    }
+
+    /// returns true if any groups are completed and thus can be emitted
+    pub fn can_emit(&self) -> bool {
+        match self.state {
+            OrderState::Start => false,
+            OrderState::InProgress(group_index) if group_index == 0 => false,
+            OrderState::InProgress(_) => true,
+            OrderState::Complete => true,
+        }
     }
 
     /// Note that the input is complete so any outstanding groups are done as well

datafusion/core/src/physical_plan/aggregates/row_hash.rs

@@ -678,15 +678,12 @@ impl GroupedHashAggregateStream {
         Ok(output)
     }
 
-    /// Can any of the hash table be emitted?
+    /// Can at least part of the hash table be emitted?
     fn can_emit(&self) -> bool {
-        let some_groups_done = self
-            .group_ordering
+        self.group_ordering
             .as_ref()
-            .map(|group_ordering| group_ordering.last_completed().is_some())
-            .unwrap_or(false);
-
-        self.input_done || some_groups_done
+            .map(|group_ordering| group_ordering.can_emit())
+            .unwrap_or(self.input_done)
     }
 
     /// Update state when the last input is seen

datafusion/physical-expr/src/aggregate/count.rs

@@ -188,7 +188,7 @@ impl GroupsAccumulator for CountGroupsAccumulator {
     // return arrays for counts
     fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
         if emit_to.is_some() {
-            panic!("emit_to handling not yet implemented");
+            panic!("emit_to handling not yet implemented: {emit_to:?}");
         }
         let counts = std::mem::take(&mut self.counts);
         let counts: PrimitiveArray<Int64Type> = Int64Array::from(counts); // zero copy, no nulls