refactor: Hydra row weight enum, TooManyCandidates count, cost_unit rename, stacked-shift pane fix

crates/devtools/src/bin/stage_pipeline.rs

@@ -598,7 +598,7 @@ fn plan_selection_json(
                 id.clone(),
                 json!({
                     "cost_per_second": cost.cost_per_second,
-                    "unit": cost.unit,
+                    "cost_unit": cost.cost_unit,
                     "source": cost.source,
                     "usage": usage_json(&cost.usage),
                     "per_node": per_node,

crates/executor/src/capability.rs

@@ -125,9 +125,10 @@ pub fn validate_summary_kernel(
 }
 
 /// The only shared grouping with a kernel: Hydra over Count-Min. Each update
-/// adds one item column's value to its group, with weight 1 (a row count) or
-/// a weight column proven non-negative (a Count-Min point count only
-/// overestimates when no weight is negative).
+/// adds one item column's value to its group, with weight 1 (a row count,
+/// `UnitCount`), a positive integer constant (`PositiveConstant`) or a weight
+/// column proven non-negative (a Count-Min point count only overestimates
+/// when no weight is negative).
 fn validate_hydra_cms(
     family: &SummaryFamilyType,
     input: &SummaryUpdate,
@@ -158,12 +159,12 @@ fn validate_hydra_cms(
     }
 }
 
-/// A HydraCms constant weight as the `i32` sketchlib's Hydra adds: a
-/// positive integer within `i32`; an error otherwise.
-pub(crate) fn hydra_constant_weight(weight: f64) -> Result<i32, String> {
+/// 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)
     {
-        Ok(weight as i32)
+        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 \

crates/executor/src/operators/mod.rs

@@ -34,14 +34,7 @@ pub(crate) mod vector_window;
 pub use aggregate::Reduction;
 pub use series_window::SubquerySteps;
 pub use sort::SortKey;
-pub use summary::SummaryEvaluation;
-/// A shared summary build's default row weight: one per row, a count.
-fn unit_weight() -> i32 {
-    1
-}
-fn is_unit_weight(weight: &i32) -> bool {
-    *weight == 1
-}
+pub use summary::{HydraRowWeight, SummaryEvaluation};
 #[derive(Clone, serde::Serialize, serde::Deserialize)]
 enum Kind {
     #[serde(skip)]
@@ -164,13 +157,8 @@ enum Kind {
     SharedSummaryBuild {
         family: SummaryFamilyType,
         item: usize,
-        /// The weight column; each row adds `constant_weight` without one.
-        #[serde(default, skip_serializing_if = "Option::is_none")]
-        weight: Option<usize>,
-        /// What each row adds when there is no weight column: 1 for a count,
-        /// or a positive integer constant weight.
-        #[serde(default = "unit_weight", skip_serializing_if = "is_unit_weight")]
-        constant_weight: i32,
+        /// What each row adds to its item's count.
+        row_weight: HydraRowWeight,
         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")]

crates/executor/src/operators/summary/mod.rs

@@ -1,5 +1,20 @@
 use super::*;
 use planner_types::ir::schema::{FamilyCapability, MergeInputRule};
+use std::num::NonZeroU32;
+
+/// What each row of a shared HydraCms build adds to its item's count.
+/// Produced by the physical planner from a validated `SummaryUpdate`
+/// weight and read once per row by the build.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub enum HydraRowWeight {
+    /// The value of this input column (Float64 or Int64, non-negative); a
+    /// row whose value is NULL adds nothing.
+    Column(usize),
+    /// This positive integer per row: 1 for a row count. At most `i32::MAX`,
+    /// since asap_sketchlib's Hydra adds `i32` weights;
+    /// [`Operator::shared_summary_build`] refuses a larger one.
+    Constant(NonZeroU32),
+}
 /// A summary evaluation: a sketch query, or an exact evaluation with typed parameters.
 #[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
 pub enum SummaryEvaluation {
@@ -63,20 +78,15 @@ impl Operator {
         })
     }
     /// One HydraCms state shared by every group, emitted as one row per group
-    /// that reads it. Each row adds its `weight` column's value (or
-    /// `constant_weight` without one, which must be positive) to its `item`
-    /// value; a row whose weight is NULL adds nothing.
+    /// that reads it. Each row adds its `row_weight` to its `item` value; a
+    /// row whose weight column is NULL adds nothing.
     pub fn shared_summary_build(
         input: SchemaRef,
         family: SummaryFamilyType,
         item: usize,
-        weight: Option<usize>,
-        constant_weight: i32,
+        row_weight: HydraRowWeight,
         groups: Vec<usize>,
     ) -> Result<Self, Error> {
-        if constant_weight <= 0 {
-            return Err(invalid("HydraCms constant weight must be positive"));
-        }
         crate::values::validate_family(&family)?;
         crate::factory::create_hydra_cms(&family).map_err(Error::Invalid)?;
         validate_groups(&input, &groups)?;
@@ -88,12 +98,22 @@ impl Operator {
                 "HydraCms items must be non-null Utf8, Int64 or Bool",
             ));
         }
-        if let Some(weight) = weight {
-            if !matches!(
-                plain(&input, weight)?.0,
-                DataType::Float64 | DataType::Int64
-            ) {
-                return Err(invalid("HydraCms weights must be Float64 or Int64"));
+        match row_weight {
+            HydraRowWeight::Column(weight) => {
+                if !matches!(
+                    plain(&input, weight)?.0,
+                    DataType::Float64 | DataType::Int64
+                ) {
+                    return Err(invalid("HydraCms weights must be Float64 or Int64"));
+                }
+            }
+            HydraRowWeight::Constant(weight) => {
+                if i32::try_from(weight.get()).is_err() {
+                    return Err(Error::Invalid(format!(
+                        "HydraCms constant weight must be at most i32::MAX (asap_sketchlib's \
+                         Hydra adds i32 weights), got {weight}"
+                    )));
+                }
             }
         }
         let mut fields = groups
@@ -110,8 +130,7 @@ impl Operator {
             kind: Kind::SharedSummaryBuild {
                 family,
                 item,
-                weight,
-                constant_weight,
+                row_weight,
                 groups,
                 filter: None,
             },
@@ -345,8 +364,7 @@ pub(super) fn execute<'a>(
         Kind::SharedSummaryBuild {
             family,
             item,
-            weight,
-            constant_weight,
+            row_weight,
             groups,
             filter,
         } => Ok(futures::stream::once(async move {
@@ -356,7 +374,7 @@ pub(super) fn execute<'a>(
                     input,
                     family,
                     *item,
-                    (*weight, *constant_weight),
+                    *row_weight,
                     groups,
                     filter.as_deref(),
                     &context,
@@ -800,7 +818,7 @@ async fn build_shared_summary(
     mut input: Input<'_, Batch>,
     family: &SummaryFamilyType,
     item: usize,
-    (weight, constant_weight): (Option<usize>, i32),
+    row_weight: HydraRowWeight,
     groups: &[usize],
     filter: Option<&Expression>,
     context: &RunContext,
@@ -834,10 +852,14 @@ async fn build_shared_summary(
             if !selected(filter, row)? {
                 continue;
             }
-            let weight = match weight.map(|i| &row[i]) {
-                None => constant_weight,
-                Some(Value::Null) => continue,
-                Some(value) => hydra_weight(value)?,
+            let weight = match row_weight {
+                HydraRowWeight::Constant(weight) => {
+                    i32::try_from(weight.get()).expect("validated HydraCms constant weight")
+                }
+                HydraRowWeight::Column(i) => match &row[i] {
+                    Value::Null => continue,
+                    value => hydra_weight(value)?,
+                },
             };
             let item = match &row[item] {
                 Value::Utf8(v) => v.to_string(),

crates/executor/src/operators/unchecked.rs

@@ -169,19 +169,11 @@ impl TryFrom<UncheckedOperator> for Operator {
             Kind::SharedSummaryBuild {
                 family,
                 item,
-                weight,
-                constant_weight,
+                row_weight,
                 groups,
                 filter,
             } => with_filter(
-                Operator::shared_summary_build(
-                    input(0)?,
-                    family,
-                    item,
-                    weight,
-                    constant_weight,
-                    groups,
-                )?,
+                Operator::shared_summary_build(input(0)?, family, item, row_weight, groups)?,
                 filter,
             )?,
             Kind::KeyedEvaluation { state, k } => {

crates/executor/src/physical_planner/mod.rs

@@ -2,7 +2,7 @@
 //! Compilation needs no readers; deployment resolves inputs after selection.
 //! A node without a native lowering fails compilation; the deployment runtime
 //! must execute it.
-use crate::operators::SummaryEvaluation;
+use crate::operators::{HydraRowWeight, SummaryEvaluation};
 use crate::summary_kernels::exact::ExactEvaluation;
 use crate::{
     operators::{Expression, Operator, Reduction, SortKey},
@@ -1027,20 +1027,20 @@ fn summary_build(
         };
         // Validated: the unit count, a positive integer constant or a
         // non-negative weight column.
-        let (weight, constant_weight) = match &update.weight {
-            SummaryInputExpr::Column(weight) => (Some(named_column(input, weight)?), 1),
-            SummaryInputExpr::Constant(weight) => (
-                None,
+        let row_weight = match &update.weight {
+            SummaryInputExpr::Column(weight) => {
+                HydraRowWeight::Column(named_column(input, weight)?)
+            }
+            SummaryInputExpr::Constant(weight) => HydraRowWeight::Constant(
                 crate::capability::hydra_constant_weight(*weight).map_err(Error::Invalid)?,
             ),
-            _ => (None, 1),
+            _ => unreachable!("validated HydraCms weight"),
         };
         return Operator::shared_summary_build(
             input.clone(),
             family.clone(),
             named_column(input, item)?,
-            weight,
-            constant_weight,
+            row_weight,
             groups(input, keys)?,
         );
     }

crates/executor/tests/hydra_cms_execution.rs

@@ -2,7 +2,7 @@
 //! planner and answers each group's item point counts from one shared grid.
 use asap_executor::{
     dag::{
-        operators::Operator,
+        operators::{HydraRowWeight, Operator},
         values::{Batch, SchemaRef, Value},
         Limits, RunContext, Scope,
     },
@@ -355,6 +355,55 @@ fn positive_constant_weight_adds_the_constant_per_row() {
     }
 }
 
+// The row weight is one serialized field: `{"Constant":2}` for a constant
+// weight, `{"Column":2}` for a weight column. A deserialized constant above
+// `i32::MAX` (asap_sketchlib's Hydra adds `i32` weights) is refused, as is
+// one passed to the constructor.
+#[test]
+fn row_weight_serializes_as_one_field_and_rejects_constants_above_i32() {
+    let constant = SummaryUpdate {
+        weight: SummaryInputExpr::Constant(2.0),
+        weight_domain: WeightDomain::NonNegative {
+            proof: NonNegativeWeightProof::PositiveConstant,
+        },
+        ..unit_count()
+    };
+    let column = weighted(non_negative());
+    for (update, expected) in [
+        (constant, serde_json::json!({"Constant": 2})),
+        (column, serde_json::json!({"Column": 2})),
+    ] {
+        let build = compile_node(&summary_agg(update, None), &[input_schema()]).unwrap();
+        let wire = serde_json::to_value(&build).unwrap();
+        let fields = &wire["kind"]["SharedSummaryBuild"];
+        assert_eq!(fields["row_weight"], expected, "{wire}");
+        assert!(fields.get("weight").is_none() && fields.get("constant_weight").is_none());
+        let restored: Operator = serde_json::from_value(wire.clone()).unwrap();
+        assert_eq!(serde_json::to_value(&restored).unwrap(), wire);
+    }
+    let build = |weight: u32| {
+        Operator::shared_summary_build(
+            input_schema(),
+            family(),
+            1,
+            HydraRowWeight::Constant(std::num::NonZeroU32::new(weight).unwrap()),
+            vec![0],
+        )
+    };
+    assert!(build(i32::MAX as u32).is_ok());
+    let error = build(i32::MAX as u32 + 1)
+        .err()
+        .expect("constant above i32::MAX");
+    assert!(error.to_string().contains("i32::MAX"), "{error}");
+    let mut wire = serde_json::to_value(build(2).unwrap()).unwrap();
+    wire["kind"]["SharedSummaryBuild"]["row_weight"] =
+        serde_json::json!({"Constant": i32::MAX as u32 + 1});
+    let error = serde_json::from_value::<Operator>(wire)
+        .err()
+        .expect("deserialized constant above i32::MAX");
+    assert!(error.to_string().contains("i32::MAX"), "{error}");
+}
+
 // A weight column of unknown sign is rejected at binding: a Count-Min point
 // count only overestimates when no weight is negative.
 #[test]