refactor!: name summary builds by shape; one RowWeight for keyed builds

crates/executor/src/capability.rs

@@ -148,7 +148,13 @@ fn validate_hydra_cms(
     match (&input.weight, proof) {
         (SummaryInputExpr::Constant(w), NonNegativeWeightProof::UnitCount) if *w == 1.0 => Ok(()),
         (SummaryInputExpr::Constant(w), NonNegativeWeightProof::PositiveConstant) => {
-            hydra_constant_weight(*w).map(|_| ())
+            let weight = constant_row_weight(*w).map_err(|error| format!("HydraCms {error}"))?;
+            i32::try_from(weight.get()).map(|_| ()).map_err(|_| {
+                format!(
+                    "HydraCms constant weight must be at most i32::MAX (asap_sketchlib's \
+                     Hydra adds i32 weights), got {weight}"
+                )
+            })
         }
         (SummaryInputExpr::Column(_), _) => Ok(()),
         _ => Err(
@@ -159,16 +165,17 @@ fn validate_hydra_cms(
     }
 }
 
-/// A HydraCms constant weight as a positive integer at most `i32::MAX`
-/// (asap_sketchlib's Hydra adds `i32` weights); an error otherwise.
-pub(crate) fn hydra_constant_weight(weight: f64) -> Result<std::num::NonZeroU32, String> {
-    if weight.is_finite() && weight > 0.0 && weight.fract() == 0.0 && weight <= f64::from(i32::MAX)
+/// A `SummaryUpdate` constant weight as the positive integer a
+/// [`RowWeight::Constant`](crate::operators::RowWeight::Constant) holds
+/// (at most `u32::MAX`); an error otherwise. The all-groups (HydraCms) build
+/// further bounds it by `i32::MAX`.
+pub(crate) fn constant_row_weight(weight: f64) -> Result<std::num::NonZeroU32, String> {
+    if weight.is_finite() && weight > 0.0 && weight.fract() == 0.0 && weight <= f64::from(u32::MAX)
     {
         Ok(std::num::NonZeroU32::new(weight as u32).expect("weight > 0"))
     } else {
         Err(format!(
-            "HydraCms constant weight must be a positive integer within i32 (asap_sketchlib's \
-             Hydra adds i32 weights), got {weight}"
+            "constant weight must be a positive integer within u32, got {weight}"
         ))
     }
 }
@@ -194,7 +201,7 @@ pub(crate) fn hydra_cms_shape(
         },
     ) = family
     else {
-        return Err("shared summary grouping has no registered kernel".into());
+        return Err("a shared-grouping (Hydra) family has no registered kernel".into());
     };
     if kind.algorithm() != &SketchAlgorithm::Cms
         || kind.params()
@@ -285,7 +292,7 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> {
     if let SummaryFamilyType::Sketch(kind, grouping) = family {
         // Plain Count-Min is native as stored state only: it merges and reads
         // its bare count, but the DAG does not build it from rows. HydraCms
-        // is built by the shared summary operator.
+        // is built by the all-groups summary build.
         if let (A::Cms, SketchParams::Cms { width, depth }) = (kind.algorithm(), kind.params()) {
             if grouping != &GroupingStrategy::PerSubpopulationInstance {
                 return hydra_cms_shape(family).map(|_| ()).map_err(Error::Invalid);

crates/executor/src/operators/mod.rs

@@ -34,7 +34,7 @@ pub(crate) mod vector_window;
 pub use aggregate::Reduction;
 pub use series_window::SubquerySteps;
 pub use sort::SortKey;
-pub use summary::{HydraRowWeight, SummaryEvaluation};
+pub use summary::{RowWeight, SummaryEvaluation};
 #[derive(Clone, serde::Serialize, serde::Deserialize)]
 enum Kind {
     #[serde(skip)]
@@ -135,7 +135,9 @@ enum Kind {
         kind: planner_types::ir::operator::JoinKind,
         predicate: Box<crate::expressions::CompiledExpression>,
     },
-    SummaryBuild {
+    /// One summary state per group, updated from one value column per row
+    /// (built by [`Operator::per_group_summary_build`]).
+    PerGroupSummaryBuild {
         family: SummaryFamilyType,
         value: usize,
         time: Option<usize>,
@@ -144,21 +146,25 @@ enum Kind {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         filter: Option<Box<Expression>>,
     },
-    KeyedSummaryBuild {
+    /// One keyed state per group whose items take a row weight, e.g. a CMS
+    /// with heap (built by [`Operator::per_group_keyed_summary_build`]).
+    PerGroupKeyedSummaryBuild {
         family: SummaryFamilyType,
-        value: usize,
         items: Vec<usize>,
+        /// What each row adds to its item's count.
+        row_weight: RowWeight,
         groups: Vec<usize>,
         /// Rows for which this is not true update no state; their group is kept.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         filter: Option<Box<Expression>>,
     },
-    /// One HydraCms state shared by all groups, adding each row's item weight.
-    SharedSummaryBuild {
+    /// One state shared by all groups and emitted per group, today a
+    /// HydraCms grid (built by [`Operator::all_groups_summary_build`]).
+    AllGroupsSummaryBuild {
         family: SummaryFamilyType,
         item: usize,
         /// What each row adds to its item's count.
-        row_weight: HydraRowWeight,
+        row_weight: RowWeight,
         groups: Vec<usize>,
         /// Rows for which this is not true update no state; their group is kept.
         #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -311,9 +317,9 @@ impl PhysicalOperator<Batch, SchemaRef> for Operator {
                 | Kind::Window { .. }
                 | Kind::Join { .. }
                 | Kind::SemiJoin { .. }
-                | Kind::SummaryBuild { .. }
-                | Kind::KeyedSummaryBuild { .. }
-                | Kind::SharedSummaryBuild { .. }
+                | Kind::PerGroupSummaryBuild { .. }
+                | Kind::PerGroupKeyedSummaryBuild { .. }
+                | Kind::AllGroupsSummaryBuild { .. }
                 | Kind::SummaryMerge { .. }
                 | Kind::VectorToScalar { .. }
         )
@@ -364,9 +370,9 @@ impl PhysicalOperator<Batch, SchemaRef> for Operator {
             Kind::Window { .. } => "WindowAggregate",
             Kind::SemiJoin { .. } => "SemiJoin",
             Kind::Join { .. } => "RelationalJoin",
-            Kind::SummaryBuild { .. }
-            | Kind::KeyedSummaryBuild { .. }
-            | Kind::SharedSummaryBuild { .. } => "SummaryAgg",
+            Kind::PerGroupSummaryBuild { .. }
+            | Kind::PerGroupKeyedSummaryBuild { .. }
+            | Kind::AllGroupsSummaryBuild { .. } => "SummaryAgg",
             Kind::KeyedEvaluation { .. } => "SummaryEstimate",
             Kind::SummaryMerge { .. } => "SummaryMerge",
             Kind::Evaluation { .. } => "SummaryEvaluation",
@@ -418,10 +424,10 @@ impl PhysicalOperator<Batch, SchemaRef> for Operator {
             }
             Kind::Join { .. } | Kind::SemiJoin { .. } => joins::execute(self, inputs, context),
             Kind::SummaryMerge { .. } => summary::execute_merge(self, inputs, context),
-            Kind::SummaryBuild { .. }
+            Kind::PerGroupSummaryBuild { .. }
             | Kind::Evaluation { .. }
-            | Kind::KeyedSummaryBuild { .. }
-            | Kind::SharedSummaryBuild { .. }
+            | Kind::PerGroupKeyedSummaryBuild { .. }
+            | Kind::AllGroupsSummaryBuild { .. }
             | Kind::KeyedEvaluation { .. } => summary::execute(self, inputs, context),
         }
     }