diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 040d7e55c..890e33c75 100644 --- a/control_plane/src/physical/compiler.rs +++ b/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,8 +1694,55 @@ 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, + selected: &BTreeSet, + 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 { @@ -1703,16 +1750,23 @@ fn preserve_native_raw_counters(queries: &mut [PlanningQuery]) -> Result<(), Com 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]))", diff --git a/control_plane/src/physical/workload_cost.rs b/control_plane/src/physical/workload_cost.rs index d2baa2965..54feda38f 100644 --- a/control_plane/src/physical/workload_cost.rs +++ b/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 diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index 4cff1b520..ce1a2cd63 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/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::() + .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] diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index 06e68731f..b245b32d6 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/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")); diff --git a/docs/examples/asapquery-compatibility-demo-snapshot.json b/docs/examples/asapquery-compatibility-demo-snapshot.json index 6f255629b..a75dfb750 100644 --- a/docs/examples/asapquery-compatibility-demo-snapshot.json +++ b/docs/examples/asapquery-compatibility-demo-snapshot.json @@ -19,7 +19,7 @@ "time_selection": { "scope": "real_time", "lookback": 5000, "as_of": null } }, { - "query": "sum_over_time(asap_demo_gauge[5s])", + "query": "sum(sum_over_time(asap_demo_gauge[5s]))", "demand": { "fixed_interval": 1000 }, "requirements": { "accuracy": "implicit_exact", "response_latency": "unspecified" }, "predictability": { "predictable": { "known_at": null } }, diff --git a/docs/user_guide/asapquery-profile.md b/docs/user_guide/asapquery-profile.md index 9b6af24e0..ccc3c4728 100644 --- a/docs/user_guide/asapquery-profile.md +++ b/docs/user_guide/asapquery-profile.md @@ -102,6 +102,13 @@ Query serving uses only the installed `QueryPlan` DAG. A query absent from that DAG is a capability miss and goes to the exact Prometheus fallback; the backend does not search materialization candidates while serving. +Per-series window queries also retain exact fallback when the raw producer cannot +preserve every source label. For example, bare `sum_over_time(m[1m])` must return +one value per series; a pooled accumulator cannot replace those rows. Explicit +additive reductions such as `sum(sum_over_time(m[1m]))` and grouped variants can +still use warm state. The demo uses this explicit global sum and forwards its +bare quantile query to Prometheus. + Counter `rate` and `increase` queries currently use the exact Prometheus fallback in this raw-ingest profile. The backend does not install pooled counter state because independent series can reset or arrive at the same timestamp.