Execute shared exact dashboard readouts from the selected workload DAG

control_plane/src/physical/compiler.rs

@@ -1723,6 +1723,23 @@ impl PhysicalCompiler {
                     query_id: query.query_id.clone(),
                     reason,
                 })?;
+            if environment.target == PhysicalDeploymentTarget::DistributedCollectors
+                && selected.iter().any(|state| {
+                    matches!(
+                        state.family,
+                        SummaryFamilyType::ExactAggregate(
+                            planner_types::post_asap::ExactKind::Count,
+                            _
+                        )
+                    )
+                })
+            {
+                return Err(CompileError::Query {
+                    query_id: query.query_id.clone(),
+                    reason: "observation-count readout requires the backend-local raw producer"
+                        .into(),
+                });
+            }
             if selected.is_empty() {
                 continue;
             }
@@ -2357,6 +2374,29 @@ fn collect_selected_materializations(
         selected: &mut Vec<SelectedMaterialization>,
     ) -> Result<(), String> {
         match &node.expr {
+            SummaryExpr::BinaryOp { lhs, rhs, .. }
+                if crate::query_plan::exact_value_executable(node) =>
+            {
+                walk(lhs, readout, selected)?;
+                walk(rhs, readout, selected)?;
+            }
+            SummaryExpr::SummaryAgg {
+                child,
+                family:
+                    SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Sum, _),
+                ..
+            } if !matches!(child.expr, SummaryExpr::KeepPreAsap(_))
+                && crate::query_plan::exact_value_executable(node) =>
+            {
+                walk(child, readout, selected)?;
+            }
+            SummaryExpr::SummaryAgg { child, .. }
+                if !matches!(child.expr, SummaryExpr::KeepPreAsap(_)) => {}
+            SummaryExpr::SummaryAgg {
+                family:
+                    SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Count, _),
+                ..
+            } if !crate::query_plan::exact_value_executable(node) => {}
             SummaryExpr::SummaryEstimate {
                 summary_input,
                 query,
@@ -2420,6 +2460,14 @@ fn collect_selected_materializations(
 
 fn physical_materialization_family(family: &SummaryFamilyType) -> SummaryFamilyType {
     match family {
+        SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Count, _) => {
+            // The local raw Sum updater retains the observation count alongside
+            // its sum. Both logical states can use this one concrete producer.
+            SummaryFamilyType::ExactAggregate(
+                planner_types::post_asap::ExactKind::Sum,
+                planner_types::post_asap::ExactParams::Sum,
+            )
+        }
         SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Rate, _) => {
             SummaryFamilyType::ExactAggregate(
                 planner_types::post_asap::ExactKind::Increase,
@@ -2673,6 +2721,68 @@ mod tests {
         }
     }
 
+    #[test]
+    fn exact_dashboard_binds_sum_and_count_to_one_local_producer() {
+        // Both dashboard roots use one packed raw accumulator, with explicit readouts.
+        let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
+            "../../../docs/examples/asapquery-planning-snapshot.json"
+        ))
+        .unwrap();
+        let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap();
+        entries[0].query = Query("sum by (service) (sum_over_time(m[1m]))".into());
+        entries[0].requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
+        let mut mean = entries[0].clone();
+        mean.query = Query(
+            "sum by (service) (sum_over_time(m[1m])) / sum by (service) (count_over_time(m[1m]))"
+                .into(),
+        );
+        entries.push(mean);
+        let bundle = snapshot.compile().unwrap();
+        assert_eq!(bundle.precompute_plan.materializations.len(), 1);
+        assert_eq!(bundle.query_plan.entries.len(), 2);
+        for entry in bundle.query_plan.entries.values() {
+            assert!(
+                !entry.nodes.values().any(|node| matches!(
+                    node,
+                    crate::query_plan::QueryPlanNode::ExactFallback { .. }
+                )),
+                "{entry:?}"
+            );
+            assert_eq!(entry.materialization_bindings().len(), 1);
+        }
+        assert!(bundle.query_plan.entries.values().any(|entry| entry
+            .nodes
+            .values()
+            .any(|node| matches!(node, crate::query_plan::QueryPlanNode::Binary { .. }))));
+    }
+
+    // Grouping must not move through non-additive arithmetic during physical
+    // packing: SUM(instance SUM / instance COUNT) is not pooled SUM / COUNT.
+    #[test]
+    fn non_additive_entity_reduction_does_not_bind_pooled_state() {
+        for query in [
+            "sum by (service) (sum_over_time(m[1m]) / count_over_time(m[1m]))",
+            "sum by (service) (sum_over_time(m[1m])) / sum by (region) (count_over_time(m[1m]))",
+            "sum(m) / sum_over_time(m[1m])",
+            "sum_over_time(m[1m]) / count_over_time(m[5m])",
+            "sum_over_time(m[1m] offset 1h) / count_over_time(m[1m] offset 1h)",
+        ] {
+            let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
+                "../../../docs/examples/asapquery-planning-snapshot.json"
+            ))
+            .unwrap();
+            let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap();
+            entries[0].query = Query(query.into());
+            entries[0].requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
+            let bundle = snapshot.compile().unwrap();
+            assert!(bundle.precompute_plan.materializations.is_empty());
+            assert!(bundle.query_plan.entries.values().all(|entry| matches!(
+                entry.nodes[&entry.root],
+                crate::query_plan::QueryPlanNode::ExactFallback { .. }
+            )));
+        }
+    }
+
     #[test]
     fn canonical_snapshot_preserves_shared_bindings_after_serialization() {
         // Two different registered readouts survive publication with one state.

control_plane/src/planner_selection.rs

@@ -159,7 +159,13 @@ pub fn select_workload(
     accuracy: AccuracyTarget,
     cost_model: &dyn CostModel,
 ) -> Result<Vec<(usize, Rc<SummaryNode>)>, SelectionError> {
-    let strategies = asap_aware_mapping::default_strategies_with(cost_model);
+    // Canonical CSE still runs inside search_workload_with_targets. Do not
+    // offer CSE's per-invocation recompute alternative: this runtime currently
+    // provisions continuously maintained, content-addressed state only.
+    let strategies: Vec<Box<dyn ReplacementStrategy + '_>> = vec![
+        Box::new(SketchAlgorithmStrategy::new(cost_model)),
+        Box::new(asap_aware_mapping::SemanticEquivalentRewriteStrategy),
+    ];
     let space = asap_aware_mapping::search_workload_with_targets(
         roots
             .into_iter()

control_plane/src/query_plan.rs

@@ -137,6 +137,9 @@ impl QueryPlanEntry {
             )));
         }
         for (id, node) in &self.nodes {
+            if matches!(node, QueryPlanNode::Scalar { value } if !value.is_finite()) {
+                return Err(QueryPlanError::Invalid("non-finite scalar constant".into()));
+            }
             for input in node.inputs() {
                 if !self.nodes.contains_key(input) {
                     return Err(QueryPlanError::Invalid(format!(
@@ -234,6 +237,17 @@ pub enum PhysicalGrouping {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)]
 pub enum QueryPlanNode {
+    Scalar {
+        value: f64,
+    },
+    Binary {
+        inputs: [QueryNodeId; 2],
+        operator: planner_types::pre_asap::ArithmeticOpKind,
+    },
+    ReduceSum {
+        input: QueryNodeId,
+        grouping: PhysicalGrouping,
+    },
     ReadMaterialization {
         binding: MaterializationBinding,
     },
@@ -256,10 +270,13 @@ pub enum QueryPlanNode {
 impl QueryPlanNode {
     pub fn inputs(&self) -> &[QueryNodeId] {
         match self {
-            Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => &[],
-            Self::SummaryEstimate { input, .. } | Self::ExactReadout { input, .. } => {
-                std::slice::from_ref(input)
+            Self::Scalar { .. } | Self::ReadMaterialization { .. } | Self::ExactFallback { .. } => {
+                &[]
             }
+            Self::Binary { inputs, .. } => inputs,
+            Self::ReduceSum { input, .. }
+            | Self::SummaryEstimate { input, .. }
+            | Self::ExactReadout { input, .. } => std::slice::from_ref(input),
             Self::SummaryMerge { inputs } => inputs,
         }
     }
@@ -269,6 +286,7 @@ impl QueryPlanNode {
 #[serde(rename_all = "snake_case")]
 pub enum ExactReadout {
     Sum,
+    Count,
     Increase,
     Rate,
 }
@@ -331,6 +349,49 @@ where
         self.next_id += 1;
         self.seen.insert(identity, id);
         let physical = match &node.expr {
+            SummaryExpr::BinaryOp { lhs, rhs, operator } if exact_value_executable(node) => {
+                let planner_types::pre_asap::BinaryOpKind::Arithmetic(operator) = &operator.kind
+                else {
+                    unreachable!()
+                };
+                QueryPlanNode::Binary {
+                    inputs: [self.lower(lhs)?, self.lower(rhs)?],
+                    operator: operator.clone(),
+                }
+            }
+            SummaryExpr::KeepPreAsap(expr) if scalar_literal(expr).is_some() => {
+                QueryPlanNode::Scalar {
+                    value: scalar_literal(expr).unwrap(),
+                }
+            }
+            SummaryExpr::SummaryAgg {
+                family:
+                    SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Sum, _),
+                child,
+                reduction,
+                ..
+            } if !matches!(child.expr, SummaryExpr::KeepPreAsap(_))
+                && exact_value_executable(node) =>
+            {
+                QueryPlanNode::ReduceSum {
+                    input: self.lower(child)?,
+                    grouping: physical_grouping(reduction, child)?,
+                }
+            }
+            SummaryExpr::SummaryAgg { child, .. }
+                if !matches!(child.expr, SummaryExpr::KeepPreAsap(_)) =>
+            {
+                QueryPlanNode::ExactFallback {
+                    reason: "unsupported exact operation over summary output".into(),
+                }
+            }
+            SummaryExpr::SummaryAgg {
+                family:
+                    SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Count, _),
+                ..
+            } if !exact_value_executable(node) => QueryPlanNode::ExactFallback {
+                reason: "only temporal observation counts are supported".into(),
+            },
             SummaryExpr::BinaryOp { .. } | SummaryExpr::KeepPreAsap(_) => {
                 QueryPlanNode::ExactFallback {
                     reason: "post-ASAP node requires exact execution".into(),
@@ -346,10 +407,16 @@ where
                     let mut binding = (self.bind)(node, family)?;
                     binding.output_grouping = physical_grouping(reduction, child)?;
                     if let Some(readout) = exact_readout(family) {
-                        let input = QueryNodeId(self.next_id);
-                        self.next_id += 1;
-                        self.nodes
-                            .insert(input, QueryPlanNode::ReadMaterialization { binding });
+                        let existing = self.nodes.iter().find_map(|(id, node)| {
+                            matches!(node, QueryPlanNode::ReadMaterialization { binding: other } if other == &binding).then_some(*id)
+                        });
+                        let input = existing.unwrap_or_else(|| {
+                            let input = QueryNodeId(self.next_id);
+                            self.next_id += 1;
+                            self.nodes
+                                .insert(input, QueryPlanNode::ReadMaterialization { binding });
+                            input
+                        });
                         QueryPlanNode::ExactReadout { input, readout }
                     } else {
                         QueryPlanNode::ReadMaterialization { binding }
@@ -399,12 +466,111 @@ fn exact_readout(family: &SummaryFamilyType) -> Option<ExactReadout> {
     use planner_types::post_asap::ExactKind;
     match family {
         SummaryFamilyType::ExactAggregate(ExactKind::Sum, _) => Some(ExactReadout::Sum),
+        SummaryFamilyType::ExactAggregate(ExactKind::Count, _) => Some(ExactReadout::Count),
         SummaryFamilyType::ExactAggregate(ExactKind::Increase, _) => Some(ExactReadout::Increase),
         SummaryFamilyType::ExactAggregate(ExactKind::Rate, _) => Some(ExactReadout::Rate),
         _ => None,
     }
 }
 
+fn scalar_literal(expr: &planner_types::pre_asap::QueryExpr) -> Option<f64> {
+    use planner_types::pre_asap::{QueryExpr, ScalarValue};
+    let value = match expr {
+        QueryExpr::PromqlScalarBridge(child) => return scalar_literal(child),
+        QueryExpr::Literal(ScalarValue::Float64(value)) => *value,
+        QueryExpr::Literal(ScalarValue::Int64(value)) => *value as f64,
+        _ => return None,
+    };
+    value.is_finite().then_some(value)
+}
+
+/// The current exact arithmetic adapter is deliberately narrower than PromQL:
+/// default vector matching, scalar literals and additive temporal readouts.
+/// Unsupported operands make the complete expression fall back.
+pub(crate) fn exact_value_executable(node: &SummaryNode) -> bool {
+    use planner_types::post_asap::ExactKind;
+    if !node
+        .guarantee
+        .as_ref()
+        .is_some_and(|guarantee| guarantee.is_exact())
+    {
+        return false;
+    }
+    match &node.expr {
+        SummaryExpr::KeepPreAsap(expr) => scalar_literal(expr).is_some(),
+        SummaryExpr::BinaryOp { lhs, rhs, operator } => {
+            matches!(
+                operator.kind,
+                planner_types::pre_asap::BinaryOpKind::Arithmetic(_)
+            ) && operator.vector_match.is_none()
+                && exact_value_executable(lhs)
+                && exact_value_executable(rhs)
+                && value_grouping(node).is_ok()
+                && match (value_source(lhs), value_source(rhs)) {
+                    (Some(left), Some(right)) => left == right,
+                    _ => true,
+                }
+        }
+        SummaryExpr::SummaryAgg {
+            family: SummaryFamilyType::ExactAggregate(kind, _),
+            child,
+            reduction,
+            ..
+        } => {
+            if matches!(child.expr, SummaryExpr::KeepPreAsap(_)) {
+                matches!(&child.expr, SummaryExpr::KeepPreAsap(expr) if matches!(expr.as_ref(), planner_types::pre_asap::QueryExpr::TimeRange { child, .. } if matches!(child.as_ref(), planner_types::pre_asap::QueryExpr::Scan { .. })))
+                    && matches!(reduction, Reduction::PerEntity)
+                    && matches!(
+                        kind,
+                        ExactKind::Sum | ExactKind::Count | ExactKind::Increase | ExactKind::Rate
+                    )
+            } else {
+                // Raw producer grouping may move through additive reductions,
+                // but never through division or other value arithmetic.
+                matches!(kind, ExactKind::Sum)
+                    && matches!(child.expr, SummaryExpr::SummaryAgg { .. })
+                    && exact_value_executable(child)
+            }
+        }
+        _ => false,
+    }
+}
+
+fn value_grouping(node: &SummaryNode) -> Result<Option<PhysicalGrouping>, QueryPlanError> {
+    match &node.expr {
+        SummaryExpr::KeepPreAsap(_) => Ok(None),
+        SummaryExpr::SummaryAgg {
+            reduction, child, ..
+        } => physical_grouping(reduction, child).map(Some),
+        SummaryExpr::BinaryOp { lhs, rhs, .. } => {
+            let left = value_grouping(lhs)?;
+            let right = value_grouping(rhs)?;
+            match (left, right) {
+                (Some(left), Some(right)) if left != right => Err(QueryPlanError::Invalid(
+                    "arithmetic operands require different producer grouping contracts".into(),
+                )),
+                (left, right) => Ok(left.or(right)),
+            }
+        }
+        _ => Err(QueryPlanError::Invalid(
+            "unsupported exact value grouping".into(),
+        )),
+    }
+}
+
+// The MVP QueryPlan evaluates all operands over one interval. Different
+// selectors/windows need per-operand time binding before they can be warm.
+fn value_source(node: &SummaryNode) -> Option<&planner_types::pre_asap::QueryExpr> {
+    match &node.expr {
+        SummaryExpr::SummaryAgg { child, .. } => match &child.expr {
+            SummaryExpr::KeepPreAsap(expr) => Some(expr),
+            _ => value_source(child),
+        },
+        SummaryExpr::BinaryOp { lhs, rhs, .. } => value_source(lhs).or_else(|| value_source(rhs)),
+        _ => None,
+    }
+}
+
 fn physical_grouping(
     reduction: &Reduction,
     child: &SummaryNode,