fix: preserve per-series semantics for raw window queries

control_plane/src/physical/compiler.rs

@@ -1682,7 +1682,7 @@ impl BackendLocalPlanningSnapshot {
             });
         }
         select_workload_roots(&mut queries, canonical_roots, &topk_evidence_by_id)?;
-        preserve_native_raw_counters(&mut queries)?;
+        preserve_native_unsafe_raw_roots(&mut queries)?;
         Ok((
             PlanningRequest {
                 queries,
@@ -1694,25 +1694,79 @@ impl BackendLocalPlanningSnapshot {
     }
 }
 
-/// Preserve native execution for counters until raw producers retain series identity.
-fn preserve_native_raw_counters(queries: &mut [PlanningQuery]) -> Result<(), CompileError> {
+/// Raw accumulators do not retain arbitrary source labels. Preserve native semantics
+/// unless the selected DAG explicitly authorizes pooling the source entities.
+fn has_unsafe_raw_entity_leaf(
+    node: &Rc<SummaryNode>,
+    selected: &BTreeSet<usize>,
+    pooling: bool,
+) -> bool {
+    use planner_types::post_asap::ExactKind;
+    use planner_types::pre_asap::Reduction;
+    match &node.expr {
+        SummaryExpr::SummaryAgg {
+            child,
+            reduction,
+            family,
+            ..
+        } => {
+            if selected.contains(&(Rc::as_ptr(node) as usize)) {
+                return matches!(reduction, Reduction::PerEntity) && !pooling;
+            }
+            let additive_reduction = matches!(reduction, Reduction::Reduce(_))
+                && matches!(family, SummaryFamilyType::ExactAggregate(ExactKind::Sum, _))
+                && matches!(
+                    &child.expr,
+                    SummaryExpr::SummaryAgg {
+                        family: SummaryFamilyType::ExactAggregate(
+                            ExactKind::Sum | ExactKind::Count,
+                            _
+                        ),
+                        ..
+                    }
+                );
+            has_unsafe_raw_entity_leaf(child, selected, additive_reduction)
+        }
+        SummaryExpr::BinaryOp { lhs, rhs, .. } => {
+            has_unsafe_raw_entity_leaf(lhs, selected, false)
+                || has_unsafe_raw_entity_leaf(rhs, selected, false)
+        }
+        SummaryExpr::SummaryEstimate { summary_input, .. } => {
+            has_unsafe_raw_entity_leaf(summary_input, selected, false)
+        }
+        SummaryExpr::SummaryMerge { children } => children
+            .iter()
+            .any(|child| has_unsafe_raw_entity_leaf(child, selected, false)),
+        _ => false,
+    }
+}
+
+/// Preserve native execution for raw states that cannot preserve source semantics.
+fn preserve_native_unsafe_raw_roots(queries: &mut [PlanningQuery]) -> Result<(), CompileError> {
     for query in queries {
         let selected = collect_selected_materializations(&query.post_asap).map_err(|reason| {
             CompileError::Query {
                 query_id: query.query_id.clone(),
                 reason,
             }
         })?;
-        if selected.iter().any(|state| {
-            matches!(
-                state.family,
-                SummaryFamilyType::ExactAggregate(
-                    planner_types::post_asap::ExactKind::Increase
-                        | planner_types::post_asap::ExactKind::Rate,
-                    _
+        let unsafe_entities = has_unsafe_raw_entity_leaf(
+            &query.post_asap,
+            &selected.iter().map(|state| state.node_identity).collect(),
+            false,
+        );
+        if unsafe_entities
+            || selected.iter().any(|state| {
+                matches!(
+                    state.family,
+                    SummaryFamilyType::ExactAggregate(
+                        planner_types::post_asap::ExactKind::Increase
+                            | planner_types::post_asap::ExactKind::Rate,
+                        _
+                    )
                 )
-            )
-        }) {
+            })
+        {
             let parsed = crate::query_parser::parse_query_expr_canonical(
                 &query.query_string,
                 query.accuracy.clone(),
@@ -1755,7 +1809,7 @@ impl PhysicalCompiler {
         }
 
         if environment.target == PhysicalDeploymentTarget::BackendLocalRemoteWrite {
-            preserve_native_raw_counters(&mut request.queries)?;
+            preserve_native_unsafe_raw_roots(&mut request.queries)?;
         }
 
         let roots = request
@@ -2799,6 +2853,41 @@ fn stable_workload_plan_id(
 mod tests {
     use super::*;
 
+    #[test]
+    fn raw_per_entity_state_requires_explicit_additive_reduction() {
+        for query in [
+            "sum_over_time(m[1m])",
+            "quantile_over_time(0.99, m[1m])",
+            "sum_over_time(m[1m]) / count_over_time(m[1m])",
+        ] {
+            let mut environment = environment(10_000);
+            environment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+            environment.collector_ids.clear();
+            let plan = PhysicalCompiler
+                .compile(request("per-entity", query), environment)
+                .unwrap();
+            assert!(
+                plan.precompute_plan.materializations.is_empty(),
+                "{query} pooled source entities"
+            );
+        }
+        for query in [
+            "sum(sum_over_time(m[1m]))",
+            "sum by (job) (sum_over_time(m[1m]))",
+        ] {
+            let mut environment = environment(10_000);
+            environment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+            environment.collector_ids.clear();
+            let plan = PhysicalCompiler
+                .compile(request("reduced", query), environment)
+                .unwrap();
+            assert!(
+                !plan.precompute_plan.materializations.is_empty(),
+                "{query} lost safe additive state"
+            );
+        }
+    }
+
     // A pooled raw counter cannot distinguish same-timestamp series or independent resets.
     #[test]
     fn backend_local_rejects_pooled_counter_materialization() {
@@ -2844,7 +2933,7 @@ mod tests {
         let mut workload = request("counter", "sum(rate(m[1m]))");
         workload
             .queries
-            .extend(request("gauge", "sum_over_time(g[1m])").queries);
+            .extend(request("gauge", "sum(sum_over_time(g[1m]))").queries);
         let mut environment = environment(10_000);
         environment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
         environment.collector_ids.clear();
@@ -2858,7 +2947,7 @@ mod tests {
             .is_empty());
         assert_eq!(
             plan.query_plan
-                .lookup("sum_over_time(g[1m])")
+                .lookup("sum(sum_over_time(g[1m]))")
                 .unwrap()
                 .materialization_bindings()
                 .len(),
@@ -3096,10 +3185,16 @@ mod tests {
             PhysicalDeploymentTarget::DistributedCollectors,
             PhysicalDeploymentTarget::BackendLocalRemoteWrite,
         ] {
-            let mut workload = request("q90", "quantile_over_time(0.90, m[1m])");
-            workload
-                .queries
-                .extend(request("q99", "quantile_over_time(0.99, m[1m])").queries);
+            let (first, second) = if target == PhysicalDeploymentTarget::BackendLocalRemoteWrite {
+                ("sum(sum_over_time(m[1m]))", "sum(sum_over_time(m[1m])) * 2")
+            } else {
+                (
+                    "quantile_over_time(0.90, m[1m])",
+                    "quantile_over_time(0.99, m[1m])",
+                )
+            };
+            let mut workload = request("first", first);
+            workload.queries.extend(request("second", second).queries);
             let mut env = environment(10_000);
             env.target = target;
             if target == PhysicalDeploymentTarget::BackendLocalRemoteWrite {
@@ -3270,8 +3365,9 @@ mod tests {
         ))
         .unwrap();
         let entries = snapshot.query_workload.repeating_queries.as_mut().unwrap();
+        entries[0].query = Query("sum(sum_over_time(m[1m]))".into());
         let mut second = entries[0].clone();
-        second.query = Query("quantile_over_time(0.90, m[1m])".into());
+        second.query = Query("sum(count_over_time(m[1m]))".into());
         entries.push(second);
         let bundle = snapshot.compile().unwrap();
         assert_eq!(bundle.query_plan.entries.len(), 2);
@@ -3547,7 +3643,7 @@ mod tests {
             ),
             (
                 "q-sum",
-                "sum_over_time(m[1m])",
+                "sum(sum_over_time(m[1m]))",
                 crate::query_plan::ExactReadout::Sum,
             ),
         ] {
@@ -3573,7 +3669,7 @@ mod tests {
             assert!(plan.collector_plans.is_empty(), "{promql}");
             let entry = plan.query_plan.entries.values().next().unwrap();
             assert!(matches!(
-                entry.nodes.get(&entry.root),
+                entry.nodes.values().find(|node| matches!(node, crate::query_plan::QueryPlanNode::ExactReadout { .. })),
                 Some(crate::query_plan::QueryPlanNode::ExactReadout { readout, .. })
                     if *readout == expected_readout
             ));
@@ -3620,11 +3716,11 @@ mod tests {
         assert!(plan.collector_plans.is_empty());
         assert!(plan.transmission_plan.rules.is_empty());
         assert_eq!(plan.query_plan.entries.len(), 6);
-        assert_eq!(plan.precompute_plan.materializations.len(), 4);
+        assert_eq!(plan.precompute_plan.materializations.len(), 3);
         for query in [
             "rate(asap_demo_counter_total[5s])",
             "increase(asap_demo_counter_total[5s])",
-            "sum_over_time(asap_demo_gauge[5s])",
+            "sum(sum_over_time(asap_demo_gauge[5s]))",
             "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
             "topk(1, sum_over_time(asap_demo_gauge[5s]))",
             "topk(1, count_over_time(asap_demo_gauge[5s]))",

control_plane/src/physical/workload_cost.rs

@@ -461,10 +461,13 @@ mod tests {
     use super::*;
 
     fn fixture() -> BackendLocalPlanningSnapshot {
-        serde_json::from_str(include_str!(
+        let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
             "../../../docs/examples/asapquery-planning-snapshot.json"
         ))
-        .unwrap()
+        .unwrap();
+        snapshot.query_workload.repeating_queries.as_mut().unwrap()[0].query =
+            planner_types::workload::Query("sum(sum_over_time(m[1m]))".into());
+        snapshot
     }
 
     fn quoted() -> (
@@ -655,7 +658,7 @@ mod tests {
         let mut shared = request.clone();
         let mut second = shared.queries[0].clone();
         second.query_id = "second-consumer".into();
-        second.query_string = "quantile_over_time(0.5, m[1m])".into();
+        second.query_string = "sum(count_over_time(m[1m]))".into();
         shared.queries.push(second);
         let roots = shared
             .queries

data_plane/src/precompute_engine/worker.rs

@@ -3259,6 +3259,59 @@ aggregations:
     // sketch (`sketch_panes`) paths.
     // -----------------------------------------------------------------------
 
+    // A pooled Sum is correct only for an explicit cross-entity reduction.
+    #[test]
+    fn pooled_sum_does_not_preserve_per_entity_output_rows() {
+        use crate::precompute_engine::operators::SumAccumulator;
+        let config = make_agg_config(
+            1,
+            "gauge",
+            AggregationType::SingleSubpopulation,
+            "Sum",
+            10,
+            0,
+            vec![],
+        );
+        let sink = Arc::new(CapturingOutputSink::new());
+        let mut worker = make_worker(
+            HashMap::from([(1, config)]),
+            sink.clone(),
+            false,
+            0,
+            LateDataPolicy::Drop,
+        );
+        worker
+            .process_group_samples(
+                1,
+                PolicyFingerprint(1),
+                "",
+                vec![
+                    ("gauge{job=\"api\"}".into(), 1000, 10.0),
+                    ("gauge{job=\"api\"}".into(), 2000, 30.0),
+                    ("gauge{job=\"worker\"}".into(), 1000, 200.0),
+                ],
+            )
+            .unwrap();
+        worker.force_close_all().unwrap();
+        let captured = sink.drain();
+        assert_eq!(captured.len(), 1);
+        assert_eq!(
+            captured[0]
+                .1
+                .as_any()
+                .downcast_ref::<SumAccumulator>()
+                .unwrap()
+                .sum,
+            240.0
+        );
+        // sum_over_time must instead emit api=40 and worker=200 separately.
+        assert_ne!(
+            captured.len(),
+            2,
+            "this producer does not retain entity rows"
+        );
+    }
+
     // This single-series updater cannot implement a grouped sum of counter increases.
     // The physical compiler rejects raw counter producers until series state is preserved.
     #[test]

data_plane/tests/asapquery_compatibility_process_e2e.rs

@@ -1,7 +1,7 @@
 //! Black-box acceptance test for the collector-free ASAPQuery profile.
 //!
 //! Starts the production binary from a canonical workload snapshot, ingests
-//! only Prometheus Remote Write v1, exercises safe warm families and counter fallback
+//! only Prometheus Remote Write v1, exercises safe warm families and per-series fallback
 //! through instant and range APIs, and verifies exact fallback request parity.
 
 use std::collections::HashMap;
@@ -834,11 +834,12 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     let first_eval = (base + 5_000) as f64 / 1_000.0;
     let second_eval = (base + 10_000) as f64 / 1_000.0;
     let backend_log = output_dir.path().join("query_engine.log");
-    // Counter roots retain the complete exact request until independent series
-    // reset/timestamp state is represented by the backend raw producer.
+    // Counter and bare per-series quantile roots retain the complete exact request
+    // until raw producers can preserve the required per-series state.
     for query in [
         "rate(asap_demo_counter_total[5s])",
         "increase(asap_demo_counter_total[5s])",
+        "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
     ] {
         let instant: Value = client
             .get(format!("{backend}/api/v1/query"))
@@ -874,15 +875,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     let sum = wait_for_warm_instant(
         &client,
         &backend,
-        "sum_over_time(asap_demo_gauge[5s])",
-        first_eval,
-        &backend_log,
-    )
-    .await;
-    let quantile = wait_for_warm_instant(
-        &client,
-        &backend,
-        "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
+        "sum(sum_over_time(asap_demo_gauge[5s]))",
         first_eval,
         &backend_log,
     )
@@ -927,15 +920,9 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         "asap_demo_gauge{job=\"api\"}"
     );
     assert!((first_value(&sum, "value").expect("sum value") - 240.0).abs() < 1e-9);
-    let quantile_value = first_value(&quantile, "value").expect("quantile value");
-    assert!(
-        (19.0..=31.0).contains(&quantile_value),
-        "unexpected p50: {quantile_value}; response={quantile}"
-    );
 
     for query in [
-        "sum_over_time(asap_demo_gauge[5s])",
-        "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
+        "sum(sum_over_time(asap_demo_gauge[5s]))",
         "topk(1, sum_over_time(asap_demo_gauge[5s]))",
         "topk(1, count_over_time(asap_demo_gauge[5s]))",
     ] {
@@ -1045,7 +1032,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
 
     // Readiness polling may briefly reach the exact fallback before a newly
     // closed warm window is visible. Every planned query above was required
-    // to converge to a warm answer; isolate the explicit fallback assertions.
+    // to converge to its declared warm or exact tier; isolate further fallback assertions.
     fallback_calls.lock().await.clear();
 
     let fallback_instant: Value = client
@@ -1156,7 +1143,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     let materializations = status["materializations"]
         .as_array()
         .expect("materialization statuses");
-    assert_eq!(materializations.len(), 4);
+    assert_eq!(materializations.len(), 3);
     assert!(materializations
         .iter()
         .all(|entry| entry["phase"] == "serving"));