@@ -28,13 +28,12 @@ use datafusion_common::assert_or_internal_err;
 use datafusion_execution::memory_pool::proxy::VecAllocExt;
 use datafusion_expr::{AggregateMetrics, EmitTo};
 
-use crate::InputOrderMode;
 use crate::PhysicalExpr;
 use crate::aggregates::group_values::{
     AccumulatorPhase, AggregateAccumulatorMetrics, AggregateArgumentMetrics,
     GroupByMetrics, GroupValues, new_group_values,
 };
-use crate::aggregates::order::GroupOrdering;
+use crate::aggregates::order::{GroupCompletionMode, GroupOrdering};
 use crate::aggregates::{
     AggregateExec, AggregateMode, PhysicalGroupBy, aggregate_expressions,
     evaluate_group_by,
@@ -115,7 +114,7 @@ impl OrderedAggregateTableMetrics {
 /// `OrderedAggrMode` selects the aggregate semantics. For example,
 /// `OrderedAggregateTable::<PartialMarker>::new(...)` consumes raw rows
 /// and emits partial states, while
-/// `OrderedAggregateTable::<FinalMarker>::new_with_input_order(...)`
+/// `OrderedAggregateTable::<FinalMarker>::new_with_group_completion(...)`
 /// consumes partial states and emits final values.
 ///
 /// Shared methods live on `impl<T>`; single/partial/final behavior lives on
@@ -189,7 +188,7 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
         output_schema: SchemaRef,
         state_schema: SchemaRef,
         batch_size: usize,
-        input_order_mode: &InputOrderMode,
+        group_completion_mode: &GroupCompletionMode,
         aggregate_mode: &AggregateMode,
         filters: Vec<Option<Arc<dyn PhysicalExpr>>>,
         metrics: OrderedAggregateTableMetrics,
@@ -199,7 +198,8 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
             "OrderedAggregateTable requires config batch_size >= 1"
         );
 
-        let group_ordering = GroupOrdering::try_new(input_order_mode)?;
+        let group_ordering =
+            GroupOrdering::try_new_for_group_completion(group_completion_mode)?;
         let group_schema = agg.group_by.group_schema(input_schema)?;
         let group_values = new_group_values(group_schema, &group_ordering)?;
         let aggregate_arguments = aggregate_expressions(
 
@@ -25,8 +25,8 @@ use arrow::datatypes::SchemaRef;
 use arrow::record_batch::RecordBatch;
 use datafusion_common::Result;
 
-use crate::InputOrderMode;
 use crate::aggregates::aggregate_hash_table::FinalMarker;
+use crate::aggregates::order::GroupCompletionMode;
 use crate::aggregates::{AggregateExec, AggregateMode, group_values::AccumulatorPhase};
 
 use super::common::HashAggregateAccumulator;
@@ -42,12 +42,12 @@ use super::common_ordered::{OrderedAggregateTable, OrderedAggregateTableMetrics}
 ///
 /// See comments at [`OrderedAggregateTable`] for details.
 impl OrderedAggregateTable<FinalMarker> {
-    pub(in crate::aggregates) fn new_with_input_order(
+    pub(in crate::aggregates) fn new_with_group_completion(
         agg: &AggregateExec,
         input_schema: &SchemaRef,
         output_schema: SchemaRef,
         batch_size: usize,
-        input_order_mode: &InputOrderMode,
+        group_completion_mode: &GroupCompletionMode,
         metrics: OrderedAggregateTableMetrics,
     ) -> Result<Self> {
         Self::new_for_mode(
@@ -56,7 +56,7 @@ impl OrderedAggregateTable<FinalMarker> {
             output_schema,
             Arc::clone(input_schema),
             batch_size,
-            input_order_mode,
+            group_completion_mode,
             &AggregateMode::Final,
             vec![None; agg.aggr_expr.len()],
             metrics,
 
@@ -68,7 +68,7 @@ impl OrderedAggregateTable<PartialMarker> {
             output_schema,
             state_schema,
             batch_size,
-            &agg.input_order_mode,
+            &agg.group_completion_mode,
             &AggregateMode::Partial,
             agg.filter_expr.iter().cloned().collect(),
             metrics,
 
@@ -59,7 +59,7 @@ impl OrderedAggregateTable<SingleMarker> {
             output_schema,
             state_schema,
             batch_size,
-            &agg.input_order_mode,
+            &agg.group_completion_mode,
             &agg.mode,
             agg.filter_expr.iter().cloned().collect(),
             metrics,
 
@@ -528,7 +528,8 @@ impl GroupedHashAggregateStream {
             .collect::<Vec<_>>()
             .join(", ");
         let name = format!("GroupedHashAggregateStream[{partition}] ({agg_fn_names})");
-        let group_ordering = GroupOrdering::try_new(&agg.input_order_mode)?;
+        let group_ordering =
+            GroupOrdering::try_new_for_group_completion(&agg.group_completion_mode)?;
         let oom_mode = match (agg.mode, &group_ordering) {
             // In partial aggregation mode, always prefer to emit incomplete results early.
             (AggregateMode::Partial, _) => OutOfMemoryMode::EmitEarly,
 
@@ -37,13 +37,14 @@ use super::aggregate_hash_table::{
     AggregateHashTable, FinalMarker, OrderedAggregateTableMetrics, PartialMarker,
     PartialSkipMarker,
 };
+use super::order::GroupCompletionMode;
 use super::skip_partial::SkipAggregationProbe;
 use super::spill::AggregateSpill;
 use crate::metrics::{
     BaselineMetrics, MetricBuilder, MetricCategory, RecordOutput, SpillMetrics,
 };
 use crate::stream::{EmptyRecordBatchStream, RecordBatchStreamAdapter};
-use crate::{InputOrderMode, SendableRecordBatchStream, metrics};
+use crate::{SendableRecordBatchStream, metrics};
 
 /// Hash aggregation is implemented in two stages: partial and final. This
 /// stream implements the partial stage.
@@ -217,7 +218,7 @@ impl PartialHashAggregateStream {
         partition: usize,
     ) -> Result<Self> {
         debug_assert_eq!(agg.mode, super::AggregateMode::Partial);
-        debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+        debug_assert_eq!(agg.group_completion_mode, GroupCompletionMode::None);
 
         let schema = Arc::clone(&agg.schema);
         let input = agg.input.execute(partition, Arc::clone(context))?;
@@ -570,7 +571,7 @@ impl FinalHashAggregateStream {
             agg.mode,
             super::AggregateMode::Final | super::AggregateMode::FinalPartitioned
         ));
-        debug_assert_eq!(agg.input_order_mode, InputOrderMode::Linear);
+        debug_assert_eq!(agg.group_completion_mode, GroupCompletionMode::None);
 
         let input = agg.input.execute(partition, Arc::clone(context))?;
         Self::new_with_input(agg, context, partition, input)
@@ -604,7 +605,7 @@ impl FinalHashAggregateStream {
                 context,
                 partition,
                 batch_size,
-                &InputOrderMode::Linear,
+                &GroupCompletionMode::None,
                 &input_schema,
                 spill_metrics,
             )?))
@@ -873,6 +874,7 @@ mod tests {
     use std::time::Duration;
 
     use super::*;
+    use crate::InputOrderMode;
     use crate::aggregates::{AggregateMode, PhysicalGroupBy};
     use crate::common::collect;
     use crate::execution_plan::ExecutionPlan;
 
@@ -206,6 +206,7 @@ use datafusion_physical_expr_common::sort_expr::{
 use datafusion_expr::utils::AggregateOrderSensitivity;
 use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
 use itertools::Itertools;
+use order::GroupCompletionMode;
 use topk::hash_table::is_supported_hash_key_type;
 use topk::heap::is_supported_heap_type;
 
@@ -895,6 +896,12 @@ pub struct AggregateExec {
     /// input order". When that is not possible, the constructor overwrites it
     /// with the unordered variant [`InputOrderMode::Linear`].
     input_order_mode: InputOrderMode,
+    /// Describes when the executor can determine that groups are complete.
+    ///
+    /// Input ordering describes a subset of the cases in which groups can be
+    /// safely emitted before the input ends. Full group completion requires only
+    /// that rows for each complete grouping tuple are contiguous.
+    group_completion_mode: GroupCompletionMode,
     cache: Arc<PlanProperties>,
     /// During initialization, if the plan supports dynamic filtering (see [`AggrDynFilter`]),
     /// it is set to `Some(..)` regardless of whether it can be pushed down to a child node.
@@ -919,6 +926,7 @@ impl AggregateExec {
             required_input_ordering: self.required_input_ordering.clone(),
             metrics: ExecutionPlanMetricsSet::new(),
             input_order_mode: self.input_order_mode.clone(),
+            group_completion_mode: self.group_completion_mode.clone(),
             cache: Arc::clone(&self.cache),
             mode: self.mode,
             group_by: Arc::clone(&self.group_by),
@@ -939,6 +947,7 @@ impl AggregateExec {
             required_input_ordering: self.required_input_ordering.clone(),
             metrics: ExecutionPlanMetricsSet::new(),
             input_order_mode: self.input_order_mode.clone(),
+            group_completion_mode: self.group_completion_mode.clone(),
             cache: Arc::clone(&self.cache),
             mode: self.mode,
             group_by: Arc::clone(&self.group_by),
@@ -1072,6 +1081,8 @@ impl AggregateExec {
             input_order_mode = InputOrderMode::Linear;
         }
 
+        let group_completion_mode = GroupCompletionMode::from(&input_order_mode);
+
         // construct a map from the input expression to the output expression of the Aggregation group by
         let group_expr_mapping =
             ProjectionMapping::try_new(group_by.expr.clone(), &input.schema())?;
@@ -1102,6 +1113,7 @@ impl AggregateExec {
             required_input_ordering,
             limit_options: None,
             input_order_mode,
+            group_completion_mode,
             cache: Arc::new(cache),
             dynamic_filter: None,
         };
@@ -1284,47 +1296,51 @@ impl AggregateExec {
             );
         }
 
-        // 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.
+        // Choose the execution path based on aggregation mode and when groups
+        // are known to be complete.
         use AggregateMode::*;
-        use InputOrderMode::*;
-        let stream = match (self.mode, &self.input_order_mode) {
-            (Partial, Linear) => StreamType::PartialHash(
+        let stream = match (self.mode, &self.group_completion_mode) {
+            (Partial, GroupCompletionMode::None) => StreamType::PartialHash(
                 PartialHashAggregateStream::new(self, context, partition)?,
             ),
-            (Partial, Sorted | PartiallySorted(_)) => {
+            (Partial, GroupCompletionMode::Partial(_) | GroupCompletionMode::Full) => {
                 StreamType::OrderedPartialAggregate(OrderedPartialAggregateStream::new(
                     self, context, partition,
                 )?)
             }
-            (PartialReduce, Linear) => StreamType::PartialReduceHash(
+            (PartialReduce, GroupCompletionMode::None) => StreamType::PartialReduceHash(
                 PartialReduceHashAggregateStream::new(self, context, partition)?,
             ),
-            (PartialReduce, Sorted | PartiallySorted(_)) => {
-                // See the comment above: the builder enforces `Linear` order for
-                // `PartialReduce` mode.
+            (
+                PartialReduce,
+                GroupCompletionMode::Partial(_) | GroupCompletionMode::Full,
+            ) => {
                 return internal_err!(
-                    "PartialReduce aggregation must use InputOrderMode::Linear"
+                    "PartialReduce aggregation must use GroupCompletionMode::None"
                 );
             }
-            (Final | FinalPartitioned, Linear) => StreamType::FinalHash(
-                FinalHashAggregateStream::new(self, context, partition)?,
-            ),
-            (Final | FinalPartitioned, Sorted | PartiallySorted(_)) => {
-                StreamType::OrderedFinalAggregate(OrderedFinalAggregateStream::new(
+            (Final | FinalPartitioned, GroupCompletionMode::None) => {
+                StreamType::FinalHash(FinalHashAggregateStream::new(
                     self, context, partition,
                 )?)
             }
-            (Single | SinglePartitioned, Linear) => StreamType::SingleHash(
-                SingleHashAggregateStream::new(self, context, partition)?,
-            ),
-            (Single | SinglePartitioned, Sorted | PartiallySorted(_)) => {
-                StreamType::OrderedSingleAggregate(OrderedSingleAggregateStream::new(
+            (
+                Final | FinalPartitioned,
+                GroupCompletionMode::Partial(_) | GroupCompletionMode::Full,
+            ) => StreamType::OrderedFinalAggregate(OrderedFinalAggregateStream::new(
+                self, context, partition,
+            )?),
+            (Single | SinglePartitioned, GroupCompletionMode::None) => {
+                StreamType::SingleHash(SingleHashAggregateStream::new(
                     self, context, partition,
                 )?)
             }
+            (
+                Single | SinglePartitioned,
+                GroupCompletionMode::Partial(_) | GroupCompletionMode::Full,
+            ) => StreamType::OrderedSingleAggregate(OrderedSingleAggregateStream::new(
+                self, context, partition,
+            )?),
         };
         Ok(stream)
     }
@@ -2352,6 +2368,8 @@ impl ExecutionPlan for AggregateExec {
             required_input_ordering: _,
             // Derived at construction from the input ordering and `group_by`.
             input_order_mode: _,
+            // Derived at construction from `input_order_mode`.
+            group_completion_mode: _,
             // Derived at construction by `Self::compute_properties`.
             cache: _,
             dynamic_filter,
@@ -5042,6 +5060,10 @@ mod tests {
             aggregate.input_order_mode(),
             InputOrderMode::PartiallySorted(_)
         ));
+        assert_eq!(
+            aggregate.group_completion_mode,
+            GroupCompletionMode::Partial(vec![0])
+        );
 
         let task_ctx = new_migrated_hash_ctx(2);
         let stream = aggregate.execute_typed(0, &task_ctx)?;
@@ -5491,6 +5513,7 @@ mod tests {
         )?;
 
         assert_eq!(aggregate.input_order_mode(), &InputOrderMode::Linear);
+        assert_eq!(aggregate.group_completion_mode, GroupCompletionMode::None);
         // This captures the behavior before #24438. When the source can declare
         // `(key, time_bin)` group-contiguous, the corresponding case can use
         // `EmissionType::Incremental`.
@@ -5651,6 +5674,10 @@ mod tests {
             Arc::clone(&schema),
         )?;
         assert_eq!(final_aggregate.input_order_mode(), &InputOrderMode::Sorted);
+        assert_eq!(
+            final_aggregate.group_completion_mode,
+            GroupCompletionMode::Full
+        );
 
         let task_ctx = new_migrated_hash_ctx(2);
         let stream = final_aggregate.execute_typed(0, &task_ctx)?;