fix: review nits after #745: cap message from its constant, Hydra constant weight, fallible updater, stacked shifts, viewer ranks

crates/executor/src/capability.rs

@@ -146,8 +146,29 @@ 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(|_| ())
+        }
         (SummaryInputExpr::Column(_), _) => Ok(()),
-        _ => Err("HydraCms weight must be the unit count or a non-negative weight column".into()),
+        _ => Err(
+            "HydraCms weight must be the unit count, a positive constant or a non-negative \
+             weight column"
+                .into(),
+        ),
+    }
+}
+
+/// 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> {
+    if weight.is_finite() && weight > 0.0 && weight.fract() == 0.0 && weight <= f64::from(i32::MAX)
+    {
+        Ok(weight as i32)
+    } else {
+        Err(format!(
+            "HydraCms constant weight must be a positive integer within i32 (asap_sketchlib's \
+             Hydra adds i32 weights), got {weight}"
+        ))
     }
 }
 

crates/executor/src/operators/mod.rs

@@ -35,6 +35,13 @@ 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
+}
 #[derive(Clone, serde::Serialize, serde::Deserialize)]
 enum Kind {
     #[serde(skip)]
@@ -157,9 +164,13 @@ enum Kind {
     SharedSummaryBuild {
         family: SummaryFamilyType,
         item: usize,
-        /// The weight column; each row adds 1 without one.
+        /// 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,
         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

@@ -63,15 +63,20 @@ 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 (1 without
-    /// one) to its `item` value; a row whose weight is NULL adds nothing.
+    /// 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.
     pub fn shared_summary_build(
         input: SchemaRef,
         family: SummaryFamilyType,
         item: usize,
         weight: Option<usize>,
+        constant_weight: i32,
         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)?;
@@ -106,6 +111,7 @@ impl Operator {
                 family,
                 item,
                 weight,
+                constant_weight,
                 groups,
                 filter: None,
             },
@@ -340,13 +346,22 @@ pub(super) fn execute<'a>(
             family,
             item,
             weight,
+            constant_weight,
             groups,
             filter,
         } => Ok(futures::stream::once(async move {
             Batch::try_new(
                 output,
-                build_shared_summary(input, family, *item, *weight, groups, filter.as_deref(), &context)
-                    .await?,
+                build_shared_summary(
+                    input,
+                    family,
+                    *item,
+                    (*weight, *constant_weight),
+                    groups,
+                    filter.as_deref(),
+                    &context,
+                )
+                .await?,
             )
         })
         .boxed_local()),
@@ -785,7 +800,7 @@ async fn build_shared_summary(
     mut input: Input<'_, Batch>,
     family: &SummaryFamilyType,
     item: usize,
-    weight: Option<usize>,
+    (weight, constant_weight): (Option<usize>, i32),
     groups: &[usize],
     filter: Option<&Expression>,
     context: &RunContext,
@@ -820,7 +835,7 @@ async fn build_shared_summary(
                 continue;
             }
             let weight = match weight.map(|i| &row[i]) {
-                None => 1,
+                None => constant_weight,
                 Some(Value::Null) => continue,
                 Some(value) => hydra_weight(value)?,
             };

crates/executor/src/operators/unchecked.rs

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

crates/executor/src/physical_planner/mod.rs

@@ -430,12 +430,9 @@ fn compile_internal(
                         "per-entity summary requires a resolved raw time range",
                     ));
                 };
-                // A tumbling pane reads its window through a `TimeShift`; the
+                // A tumbling pane reads its window through `TimeShift`s; the
                 // deployment supplies the shifted raw rows.
-                let source = match child.non_asap() {
-                    Some(NonASAPOp::TimeShift { child, .. }) => child,
-                    _ => child,
-                };
+                let source = below_time_shifts(child, |node| node.non_asap(), Ok)?;
                 let Some(NonASAPOp::Scan { schema, .. }) = source.non_asap() else {
                     return Err(invalid("per-entity summary requires a resolved source"));
                 };
@@ -598,6 +595,21 @@ fn compile_internal(
     Ok(physical_dag)
 }
 
+/// The first node at or below `node` that is not a `TimeShift`, following
+/// each shift's child: a range reads its raw source through any number of
+/// stacked shifts. `operator` gives a node's non-ASAP operator (`None` for an
+/// ASAP one) and `resolve` a child's node; an error when a child is missing.
+pub(crate) fn below_time_shifts<'a, N, C: 'a>(
+    mut node: &'a N,
+    operator: impl Fn(&'a N) -> Option<&'a NonASAPOp<C>>,
+    resolve: impl Fn(&'a C) -> Result<&'a N, Error>,
+) -> Result<&'a N, Error> {
+    while let Some(NonASAPOp::TimeShift { child, .. }) = operator(node) {
+        node = resolve(child)?;
+    }
+    Ok(node)
+}
+
 /// The logical window of the counter evaluation `id`, as `(offset,
 /// lookback)`: it ends `offset` ms before the evaluation time and spans
 /// `lookback` ms. It is the union of the range windows the evaluation
@@ -1013,16 +1025,22 @@ fn summary_build(
         let Some(SummaryInputExpr::Column(item)) = &update.item else {
             unreachable!("validated HydraCms item column")
         };
-        // Validated: the unit count, or a non-negative weight column.
-        let weight = match &update.weight {
-            SummaryInputExpr::Column(weight) => Some(named_column(input, weight)?),
-            _ => None,
+        // 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,
+                crate::capability::hydra_constant_weight(*weight).map_err(Error::Invalid)?,
+            ),
+            _ => (None, 1),
         };
         return Operator::shared_summary_build(
             input.clone(),
             family.clone(),
             named_column(input, item)?,
             weight,
+            constant_weight,
             groups(input, keys)?,
         );
     }

crates/executor/src/physical_planner/precompute.rs

@@ -88,16 +88,22 @@ pub fn boundary_schema(
     node: &PhysicalASAPDAGNode,
     nodes: &BTreeMap<u64, &PhysicalASAPDAGNode>,
 ) -> Result<SchemaRef, Error> {
+    let resolve = |child: &usize| {
+        nodes
+            .get(&(*child as u64))
+            .copied()
+            .ok_or_else(|| invalid("missing precompute node"))
+    };
     let mut scan = node;
     if let Payload::NonASAP(NonASAPOp::TimeRange { child, .. }) = &scan.payload {
-        scan = nodes
-            .get(&(*child as u64))
-            .ok_or_else(|| invalid("missing precompute node"))?;
-        while let Payload::NonASAP(NonASAPOp::TimeShift { child, .. }) = &scan.payload {
-            scan = nodes
-                .get(&(*child as u64))
-                .ok_or_else(|| invalid("missing precompute node"))?;
-        }
+        scan = below_time_shifts(
+            resolve(child)?,
+            |node| match &node.payload {
+                Payload::NonASAP(operator) => Some(operator),
+                Payload::ASAP(_) => None,
+            },
+            resolve,
+        )?;
     }
     if !matches!(
         &scan.payload,