make EmitTo::NextBlock simpler.

Author: Rachelint

datafusion/expr-common/src/groups_accumulator.rs

@@ -36,7 +36,7 @@ pub enum EmitTo {
     /// Emit next block in the blocked managed groups
     ///
     /// Similar as `Emit::All`, will also clear all existing group indexes
-    NextBlock(bool),
+    NextBlock,
 }
 
 impl EmitTo {
@@ -75,7 +75,7 @@ impl EmitTo {
                 std::mem::swap(v, &mut t);
                 t
             }
-            Self::NextBlock(_) => unreachable!("don't support take block in take_needed"),
+            Self::NextBlock => unreachable!("don't support take block in take_needed"),
         }
     }
 
@@ -85,7 +85,7 @@ impl EmitTo {
     ///
     fn take_needed_block<T>(&self, blocks: &mut VecDeque<Vec<T>>) -> Vec<T> {
         assert!(
-            matches!(self, Self::NextBlock(_)),
+            matches!(self, Self::NextBlock),
             "only support take block in take_needed_block"
         );
         blocks

datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/accumulate.rs

@@ -303,7 +303,7 @@ impl SeenValues for FlatSeenValues {
                 }
                 first_n_null
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("not support block emit in flat seen values")
             }
         };
@@ -415,7 +415,7 @@ impl SeenValues for BlockedSeenValues {
     }
 
     fn emit(&mut self, emit_to: EmitTo) -> NullBuffer {
-        assert!(matches!(emit_to, EmitTo::NextBlock(_)));
+        assert!(matches!(emit_to, EmitTo::NextBlock));
 
         let mut block = self
             .blocked_builders
@@ -542,7 +542,7 @@ impl NullStateAdapter {
                 let mut return_builder = BooleanBufferBuilder::new(0);
                 let num_blocks = null_state.seen_values.blocked_builders.len();
                 for _ in 0..num_blocks {
-                    let blocked_nulls = null_state.build(EmitTo::NextBlock(true));
+                    let blocked_nulls = null_state.build(EmitTo::NextBlock);
                     for bit in blocked_nulls.inner().iter() {
                         return_builder.append(bit);
                     }

datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/bool_op.rs

@@ -118,7 +118,7 @@ where
                 }
                 first_n
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this accumulator still not support blocked groups")
             }
         };

datafusion/functions-aggregate/src/correlation.rs

@@ -448,7 +448,7 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator {
         let n = match emit_to {
             EmitTo::All => self.count.len(),
             EmitTo::First(n) => n,
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this accumulator still not support blocked groups")
             }
         };
@@ -504,7 +504,7 @@ impl GroupsAccumulator for CorrelationGroupsAccumulator {
         let n = match emit_to {
             EmitTo::All => self.count.len(),
             EmitTo::First(n) => n,
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this accumulator still not support blocked groups")
             }
         };

datafusion/functions-aggregate/src/first_last.rs

@@ -404,7 +404,7 @@ where
                 self.size_of_orderings -=
                     result.iter().map(ScalarValue::size_of_vec).sum::<usize>()
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this accumulator still not support blocked groups")
             }
         }
@@ -431,7 +431,7 @@ where
                 }
                 first_n
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this group values still not support blocked groups")
             }
         }

datafusion/functions-aggregate/src/min_max/min_max_bytes.rs

@@ -511,7 +511,7 @@ impl MinMaxBytesState {
                 self.total_data_bytes -= first_data_capacity;
                 (first_data_capacity, first_min_maxes)
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this accumulator still not support blocked groups")
             }
         }

datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs

@@ -1154,7 +1154,7 @@ impl<const STREAMING: bool> GroupValues for GroupValuesColumn<STREAMING> {
 
                 output
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this group values still not support blocked groups")
             }
         };

datafusion/physical-plan/src/aggregates/group_values/row.rs

@@ -229,7 +229,7 @@ impl GroupValues for GroupValuesRows {
                 });
                 output
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this group values still not support blocked groups")
             }
         };

datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes.rs

@@ -116,7 +116,7 @@ impl<O: OffsetSizeTrait> GroupValues for GroupValuesByes<O> {
 
                 emit_group_values
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this group values still not support blocked groups")
             }
         };

datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes_view.rs

@@ -117,7 +117,7 @@ impl GroupValues for GroupValuesBytesView {
 
                 emit_group_values
             }
-            EmitTo::NextBlock(_) => {
+            EmitTo::NextBlock => {
                 unreachable!("this group values still not support blocked groups")
             }
         };