Unconditionally enable reduce reduction

Removes the feature flag and the reduce collation implementation that is
now unreachable.

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

src/compute-types/src/explain/text.rs

@@ -37,8 +37,7 @@ use crate::plan::join::delta_join::{DeltaPathPlan, DeltaStagePlan};
 use crate::plan::join::linear_join::LinearStagePlan;
 use crate::plan::join::{DeltaJoinPlan, JoinClosure, LinearJoinPlan};
 use crate::plan::reduce::{
-    AccumulablePlan, BasicPlan, BucketedPlan, CollationPlan, HierarchicalPlan, MonotonicPlan,
-    SingleBasicPlan,
+    AccumulablePlan, BasicPlan, BucketedPlan, HierarchicalPlan, MonotonicPlan, SingleBasicPlan,
 };
 use crate::plan::threshold::ThresholdPlan;
 use crate::plan::{AvailableCollections, LirId, Plan, PlanNode};
@@ -358,14 +357,6 @@ impl Plan {
                         ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
                         ctx.indent.reset();
                     }
-                    ReducePlan::Collation(plan) => {
-                        writeln!(
-                            f,
-                            "{}→Collated Multi-GroupAggregate{annotations}",
-                            ctx.indent
-                        )?;
-                        ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
-                    }
                 }
 
                 ctx.indented(|ctx| {
@@ -765,10 +756,6 @@ impl Plan {
                         writeln!(f, "{}Reduce::Basic{}", ctx.indent, annotations)?;
                         ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
                     }
-                    ReducePlan::Collation(plan) => {
-                        writeln!(f, "{}Reduce::Collation{}", ctx.indent, annotations)?;
-                        ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
-                    }
                 }
                 ctx.indented(|ctx| {
                     if let Some(key) = input_key {
@@ -1614,77 +1601,6 @@ impl BasicPlan {
     }
 }
 
-impl DisplayText<PlanRenderingContext<'_, Plan>> for CollationPlan {
-    fn fmt_text(
-        &self,
-        f: &mut fmt::Formatter<'_>,
-        ctx: &mut PlanRenderingContext<'_, Plan>,
-    ) -> fmt::Result {
-        if ctx.config.verbose_syntax {
-            self.fmt_verbose_text(f, ctx)
-        } else {
-            self.fmt_default_text(f, ctx)
-        }
-    }
-}
-
-impl CollationPlan {
-    #[allow(clippy::needless_pass_by_ref_mut)]
-    fn fmt_default_text(
-        &self,
-        f: &mut fmt::Formatter<'_>,
-        ctx: &mut PlanRenderingContext<'_, Plan>,
-    ) -> fmt::Result {
-        if let Some(plan) = &self.accumulable {
-            writeln!(f, "{}Accumulable sub-aggregation", ctx.indent)?;
-            ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
-        }
-        if let Some(plan) = &self.hierarchical {
-            writeln!(f, "{}Hierarchical sub-aggregation", ctx.indent)?;
-            ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
-        }
-        if let Some(plan) = &self.basic {
-            writeln!(f, "{}Non-incremental sub-aggregation", ctx.indent)?;
-            ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
-        }
-        Ok(())
-    }
-
-    fn fmt_verbose_text(
-        &self,
-        f: &mut fmt::Formatter<'_>,
-        ctx: &mut PlanRenderingContext<'_, Plan>,
-    ) -> fmt::Result {
-        {
-            use crate::plan::reduce::ReductionType;
-            let aggregate_types = &self
-                .aggregate_types
-                .iter()
-                .map(|reduction_type| match reduction_type {
-                    ReductionType::Accumulable => "a".to_string(),
-                    ReductionType::Hierarchical => "h".to_string(),
-                    ReductionType::Basic => "b".to_string(),
-                })
-                .collect::<Vec<_>>();
-            let aggregate_types = separated(", ", aggregate_types);
-            writeln!(f, "{}aggregate_types=[{}]", ctx.indent, aggregate_types)?;
-        }
-        if let Some(plan) = &self.accumulable {
-            writeln!(f, "{}accumulable", ctx.indent)?;
-            ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
-        }
-        if let Some(plan) = &self.hierarchical {
-            writeln!(f, "{}hierarchical", ctx.indent)?;
-            ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
-        }
-        if let Some(plan) = &self.basic {
-            writeln!(f, "{}basic", ctx.indent)?;
-            ctx.indented(|ctx| plan.fmt_text(f, ctx))?;
-        }
-        Ok(())
-    }
-}
-
 /// Helper struct for rendering an arrangement.
 struct Arrangement<'a> {
     key: &'a Vec<MirScalarExpr>,

src/compute-types/src/plan.rs

@@ -635,9 +635,6 @@ impl<T: timely::progress::Timestamp> Plan<T> {
                     PlanNode::Reduce { plan, .. } => {
                         // Upgrade non-monotonic hierarchical plans to monotonic with mandatory consolidation.
                         match plan {
-                            ReducePlan::Collation(collation) => {
-                                collation.as_monotonic(true);
-                            }
                             ReducePlan::Hierarchical(hierarchical) => {
                                 hierarchical.as_monotonic(true);
                             }

src/compute-types/src/plan/reduce.proto

@@ -65,7 +65,7 @@ use std::collections::BTreeMap;
 use mz_expr::{
     AggregateExpr, AggregateFunc, MapFilterProject, MirScalarExpr, permutation_for_arrangement,
 };
-use mz_ore::{assert_none, soft_assert_or_log};
+use mz_ore::soft_assert_or_log;
 use serde::{Deserialize, Serialize};
 
 use crate::plan::{AvailableCollections, bucketing_of_expected_group_size};
@@ -127,10 +127,6 @@ pub enum ReducePlan {
     Hierarchical(HierarchicalPlan),
     /// Plan for computing only basic aggregations.
     Basic(BasicPlan),
-    /// Plan for computing a mix of different kinds of aggregations.
-    /// We need to do extra work here to reassemble results back in the
-    /// requested order.
-    Collation(CollationPlan),
 }
 
 /// Plan for computing a set of accumulable aggregations.
@@ -351,63 +347,21 @@ impl ReducePlan {
         }
 
         // Convert each grouped list of reductions into a plan.
-        let plan: Vec<_> = reduction_types
-            .into_iter()
-            .map(|(typ, aggregates_list)| {
-                ReducePlan::create_inner(
-                    typ,
-                    aggregates_list,
-                    monotonic,
-                    expected_group_size,
-                    fused_unnest_list,
-                )
-            })
-            .collect();
-
-        // If we only have a single type of aggregation present we can
-        // render that directly
-        if plan.len() == 1 {
-            return plan[0].clone();
-        }
-
-        // Otherwise, we have to stitch reductions together.
-
-        // First, lets sanity check that we don't have an impossible number
-        // of reduction types.
-        assert!(plan.len() <= 3);
-
-        let mut collation: CollationPlan = Default::default();
-
-        // Construct a mapping from output_position -> reduction that we can
-        // use to reconstruct the output in the correct order.
-        let aggregate_types = aggregates
-            .iter()
-            .map(|a| reduction_type(&a.func))
-            .collect::<Vec<_>>();
-
-        collation.aggregate_types = aggregate_types;
-
-        for expr in plan.into_iter() {
-            match expr {
-                ReducePlan::Accumulable(e) => {
-                    assert_none!(collation.accumulable);
-                    collation.accumulable = Some(e);
-                }
-                ReducePlan::Hierarchical(e) => {
-                    assert_none!(collation.hierarchical);
-                    collation.hierarchical = Some(e);
-                }
-                ReducePlan::Basic(e) => {
-                    assert_none!(collation.basic);
-                    collation.basic = Some(e);
-                }
-                ReducePlan::Distinct | ReducePlan::Collation(_) => {
-                    panic!("Inner reduce plan was unsupported type!")
-                }
-            }
-        }
-
-        ReducePlan::Collation(collation)
+        let Some(plan) = reduction_types.pop_first().map(|(typ, aggregates_list)| {
+            ReducePlan::create_inner(
+                typ,
+                aggregates_list,
+                monotonic,
+                expected_group_size,
+                fused_unnest_list,
+            )
+        }) else {
+            panic!(
+                "Multiple reduction types ({reduction_types:?}) detected in ReducePlan::create_from. This indicates a bug in reduce reduction."
+            );
+        };
+
+        plan
     }
 
     /// Generate a plan for computing the specified type of aggregations.

src/compute-types/src/plan/reduce.rs

@@ -931,7 +931,6 @@ impl<'a, T> std::fmt::Display for RenderPlanExprHumanizer<'a, T> {
                     write!(f, ")")
                 }
                 ReducePlan::Basic(..) => write!(f, "Non-incremental GroupAggregate`"),
-                ReducePlan::Collation(..) => write!(f, "Collated Multi-GroupAggregate"),
             },
             TopK {
                 input: _,