From 0855ad0e50704d6a5d5131bc84ea4ade0d2103a9 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 7 Sep 2026 16:12:26 -0600 Subject: [PATCH 1/2] Drive shared distributed acceptance with the real Collector update runtime --- data_plane/tests/backend_process_e2e.rs | 73 +++++++++++++++---------- 1 file changed, 44 insertions(+), 29 deletions(-) diff --git a/data_plane/tests/backend_process_e2e.rs b/data_plane/tests/backend_process_e2e.rs index 338f2e9e..beca3d17 100644 --- a/data_plane/tests/backend_process_e2e.rs +++ b/data_plane/tests/backend_process_e2e.rs @@ -16,7 +16,8 @@ use asap_otel_proto::tonic::metrics::v1::{ metric::Data, DdSketch, DdSketchDataPoint, DdSketchEncoding, Metric, ResourceMetrics, ScopeMetrics, }; -use asap_sketchlib::proto::sketchlib::DdSketchState; +use asap_precompute_rs::Precompute; +use asap_sketchlib::proto::sketchlib::{sketch_envelope, SketchEnvelope as ProtoEnvelope}; use control_plane::opamp::{ opamp_proto, CollectorPlanStatus, CollectorPlanStatusKind, COLLECTOR_PLAN_CAPABILITY, COLLECTOR_PLAN_MESSAGE, PLAN_STATUS_MESSAGE, @@ -75,10 +76,49 @@ fn ddsketch_export( plan: &serde_json::Value, sequence: u64, ) -> Vec { - let mut sketch = asap_sketchlib::DdSketch::new(alpha); + let decoded = asap_precompute_rs::CollectorPlan::from_json( + &serde_json::to_vec(plan).unwrap(), + "whole-e2e-collector", + ) + .unwrap(); + let mut configs = decoded.to_precompute_config_set().unwrap().configs; + assert_eq!( + configs.len(), + 1, + "two query roots must create only one producer" + ); + let config = configs.remove(0); + assert_eq!(config.sketch_params["relative_accuracy"], alpha); + let runtime = asap_precompute_rs::precompute::PrecomputeImpl::new( + Some(config), + Some(Box::new(move || { + Box::new(asap_precompute_rs::sketches::DDSketchWrapper::new(alpha)) + })), + Some(Box::new(asap_precompute_rs::sketches::DDSketchObserver)), + ); for value in values { - sketch.update(*value); + runtime + .observe(&asap_precompute_rs::Observation::new( + timestamp_ns / 1_000_000 - 500, + metric, + vec![], + vec![asap_precompute_rs::KeyValue::new("service", "whole-e2e")], + asap_precompute_rs::ObservationValue { + kind: asap_precompute_rs::ObservationValueKind::Float, + float: *value, + ..Default::default() + }, + )) + .unwrap(); } + let envelopes = runtime.tick(timestamp_ns / 1_000_000); + assert_eq!(runtime.stats().input_observations, values.len() as u64); + assert_eq!(envelopes.len(), 1); + assert_eq!(envelopes[0].count, values.len() as u64); + let wire = ProtoEnvelope::decode(envelopes[0].payload.as_slice()).unwrap(); + let Some(sketch_envelope::SketchState::Ddsketch(state)) = wire.sketch_state else { + panic!("expected actual Collector DDSketch state") + }; let materialization = plan["materializations"][0]["materialization"] .as_u64() .unwrap(); @@ -126,12 +166,7 @@ fn ddsketch_export( attributes, start_time_unix_nano: timestamp_ns.saturating_sub(1_000_000_000), time_unix_nano: timestamp_ns, - sketch: DdSketchState { - alpha: sketch.wire_alpha(), - store_counts: sketch.store_counts, - store_offset: sketch.store_offset, - } - .encode_to_vec(), + sketch: state.encode_to_vec(), encoding: DdSketchEncoding::DdsketchEncodingProto as i32, exemplars: Vec::new(), flags: 0, @@ -519,26 +554,6 @@ async fn production_control_plane_to_data_plane_otlp_to_promql() { let body = ingestion.text().await.unwrap(); assert!(status.is_success(), "OTLP rejected: {status}: {body}"); - // The sealed full frame already carries its exact window. A subsequent - // checkpoint exercises the same producer's next window independently. - let watermark_ns = sample_ns + window_ms * 1_000_000; - client - .post(format!("http://{otlp_http}/v1/metrics")) - .header("content-type", "application/x-protobuf") - .body(ddsketch_export( - "whole_process_e2e_latency_ms", - watermark_ns, - &[], - planned_alpha, - &collector_plan, - 2, - )) - .send() - .await - .expect("POST watermark OTLP to production data plane") - .error_for_status() - .expect("data plane accepted watermark"); - let query = "quantile_over_time(0.99, whole_process_e2e_latency_ms[1s])"; let mut last_response = serde_json::Value::Null; for _ in 0..50 { From 80beceff188c4c02e752ac56f68f743193adcf4a Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 7 Sep 2026 16:16:49 -0600 Subject: [PATCH 2/2] Fail closed when exact arithmetic requires different grouping or temporal bindings --- control_plane/src/physical/compiler.rs | 27 +++++++++++++ control_plane/src/query_plan.rs | 55 ++++++++++++++++++++++++-- 2 files changed, 78 insertions(+), 4 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 3617e724..929e20cd 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -3006,6 +3006,33 @@ mod tests { .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. diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index a74f702c..4d46306f 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -505,6 +505,11 @@ pub(crate) fn exact_value_executable(node: &SummaryNode) -> bool { ) && 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, _), @@ -513,17 +518,59 @@ pub(crate) fn exact_value_executable(node: &SummaryNode) -> bool { .. } => { if matches!(child.expr, SummaryExpr::KeepPreAsap(_)) { - matches!(kind, ExactKind::Sum | ExactKind::Increase | ExactKind::Rate) - || (matches!((kind, reduction), (ExactKind::Count, Reduction::PerEntity)) - && matches!(&child.expr, SummaryExpr::KeepPreAsap(expr) if matches!(expr.as_ref(), planner_types::pre_asap::QueryExpr::TimeRange { .. }))) + 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 { - matches!(kind, ExactKind::Sum) && exact_value_executable(child) + // 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, 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,