diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 09e25f1c..a764fc4e 100644 --- a/control_plane/src/physical/compiler.rs +++ b/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, ) -> 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. diff --git a/control_plane/src/planner_selection.rs b/control_plane/src/planner_selection.rs index efd85feb..a41ef105 100644 --- a/control_plane/src/planner_selection.rs +++ b/control_plane/src/planner_selection.rs @@ -159,7 +159,13 @@ pub fn select_workload( accuracy: AccuracyTarget, cost_model: &dyn CostModel, ) -> Result)>, 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> = 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() diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index ac1dbf78..4d46306f 100644 --- a/control_plane/src/query_plan.rs +++ b/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 { 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 { + 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, 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, diff --git a/data_plane/src/precompute_engine/operators/sum_accumulator.rs b/data_plane/src/precompute_engine/operators/sum_accumulator.rs index c1069d2a..c4dcf549 100644 --- a/data_plane/src/precompute_engine/operators/sum_accumulator.rs +++ b/data_plane/src/precompute_engine/operators/sum_accumulator.rs @@ -11,35 +11,53 @@ use asap_types::Statistic; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct SumAccumulator { pub sum: f64, + /// None for scalar-only payloads; a sum does not establish a sample count. + #[serde(default)] + pub observation_count: Option, } impl SumAccumulator { pub fn new() -> Self { - Self { sum: 0.0 } + Self { + sum: 0.0, + observation_count: Some(0), + } } pub fn with_sum(sum: f64) -> Self { - Self { sum } + Self { + sum, + observation_count: None, + } } pub fn update(&mut self, value: f64) { self.sum += value; + self.observation_count = self + .observation_count + .and_then(|count| count.checked_add(1)); } pub fn deserialize_from_json(data: &Value) -> Result> { let sum = data["sum"] .as_f64() .ok_or("Missing or invalid 'sum' field")?; - Ok(Self::with_sum(sum)) + Ok(Self { + sum, + observation_count: data.get("observation_count").and_then(Value::as_u64), + }) } pub fn deserialize_from_bytes(buffer: &[u8]) -> Result> { - if buffer.len() < 4 { - return Err("Buffer too short for f32".into()); + match buffer.len() { + // Legacy Python scalar sums carry no sample-count evidence. + 4 => Ok(Self::with_sum(f32::from_le_bytes(buffer.try_into()?) as f64)), + // Counted sums use the same fixed layout as the Collector Sum payload. + 16 => Self::from_sum_bytes(buffer), + len => { + Err(format!("Invalid persisted Sum payload length: {len} (want 4 or 16)").into()) + } } - // Python uses struct.pack(" Result> { if buffer.len() < 16 { return Err(format!("Sum payload too short: {} bytes (want 16)", buffer.len()).into()); } let sum = f64::from_le_bytes(buffer[0..8].try_into().unwrap()); - // count = u64::from_le_bytes(buffer[8..16]) — decoded position documented - // but not retained by the scalar-sum accumulator. - Ok(Self::with_sum(sum)) + let count = u64::from_le_bytes(buffer[8..16].try_into().unwrap()); + Ok(Self { + sum, + observation_count: Some(count), + }) } } @@ -74,13 +93,21 @@ impl Default for SumAccumulator { impl SerializableToSink for SumAccumulator { fn serialize_to_json(&self) -> Value { serde_json::json!({ - "sum": self.sum + "sum": self.sum, + "observation_count": self.observation_count }) } fn serialize_to_bytes(&self) -> Vec { - // Match Python's struct.pack(" { + let mut bytes = Vec::with_capacity(16); + bytes.extend_from_slice(&self.sum.to_le_bytes()); + bytes.extend_from_slice(&count.to_le_bytes()); + bytes + } + None => (self.sum as f32).to_le_bytes().to_vec(), + } } } @@ -136,10 +163,9 @@ impl AggregateCore for SumAccumulator { } fn aux_stats(&self) -> AuxStats { - // SumAccumulator tracks exactly one scalar — the sum. - // Count/min/max aren't retained by this type. AuxStats { sum: Some(self.sum), + count: self.observation_count, ..AuxStats::empty() } } @@ -191,6 +217,7 @@ impl SingleSubpopulationAggregateFactory for SumAccumulatorFactory { ) -> Result, Box> { let mut total_sum = 0.0; + let mut observation_count = Some(0u64); for acc in accumulators { if acc.type_name() != "SumAccumulator" { @@ -198,9 +225,14 @@ impl SingleSubpopulationAggregateFactory for SumAccumulatorFactory { } let sum_value = acc.query(Statistic::Sum, None)?; total_sum += sum_value; + observation_count = + observation_count.and_then(|total| total.checked_add(acc.aux_stats().count?)); } - Ok(Box::new(SumAccumulator::with_sum(total_sum))) + Ok(Box::new(SumAccumulator { + sum: total_sum, + observation_count, + })) } fn create_default(&self) -> Box { @@ -213,7 +245,13 @@ impl MergeableAccumulator for SumAccumulator { accumulators: Vec, ) -> Result> { let total_sum = accumulators.iter().map(|acc| acc.sum).sum(); - Ok(SumAccumulator::with_sum(total_sum)) + let observation_count = accumulators + .iter() + .try_fold(0u64, |total, acc| total.checked_add(acc.observation_count?)); + Ok(SumAccumulator { + sum: total_sum, + observation_count, + }) } } @@ -221,6 +259,63 @@ impl MergeableAccumulator for SumAccumulator { mod tests { use super::*; + // Sample counts must survive updates and merges independently of the sum. + #[test] + fn observation_count_survives_merge() { + let mut first = SumAccumulator::new(); + first.update(10.0); + first.update(20.0); + let mut second = SumAccumulator::new(); + second.update(100.0); + let merged = SumAccumulator::merge_accumulators(vec![first, second]).unwrap(); + assert_eq!(merged.sum, 130.0); + assert_eq!(merged.aux_stats().count, Some(3)); + } + + // A legacy scalar sum has no evidence of how many observations produced it. + #[test] + fn legacy_sum_does_not_invent_observation_count() { + let mut raw = SumAccumulator::new(); + raw.update(10.0); + let merged = + SumAccumulator::merge_accumulators(vec![raw, SumAccumulator::with_sum(20.0)]).unwrap(); + assert_eq!(merged.aux_stats().count, None); + } + + // Persistence retains known counts, including zero and the full u64 range. + #[test] + fn counted_sum_binary_round_trip() { + for count in [0, 3, u64::MAX] { + let acc = SumAccumulator { + sum: 1.0000000000001, + observation_count: Some(count), + }; + let bytes = acc.serialize_to_bytes(); + assert_eq!(bytes.len(), 16); + let restored = SumAccumulator::deserialize_from_bytes(&bytes).unwrap(); + assert_eq!(restored.sum, acc.sum); + assert_eq!(restored.observation_count, Some(count)); + } + } + + // Existing scalar-only files remain readable without inventing counts. + #[test] + fn legacy_binary_sum_has_unknown_count() { + let bytes = 42.5f32.to_le_bytes(); + let restored = SumAccumulator::deserialize_from_bytes(&bytes).unwrap(); + assert_eq!(restored.sum, 42.5); + assert_eq!(restored.observation_count, None); + assert_eq!(restored.serialize_to_bytes(), bytes); + } + + // Truncated counted payloads must not silently decode as scalar sums. + #[test] + fn persisted_sum_rejects_invalid_lengths() { + for len in [0, 3, 5, 8, 15, 17] { + assert!(SumAccumulator::deserialize_from_bytes(&vec![0; len]).is_err()); + } + } + #[test] fn test_sum_accumulator_creation() { let acc = SumAccumulator::new(); diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index b9a83c79..de53bf96 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs @@ -122,6 +122,7 @@ pub fn execute_query_plan_instant( #[derive(Clone)] enum PhysicalQueryOutput { + Scalar(f64), State(Vec<(BTreeMap, GroupState)>), Value(Vec<(BTreeMap, SummaryValue)>), } @@ -151,6 +152,19 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { inputs: &[Self::Output], ) -> Result { match node { + QueryPlanNode::Scalar { value } => Ok(PhysicalQueryOutput::Scalar(*value)), + QueryPlanNode::Binary { operator, .. } => { + let [lhs, rhs] = inputs else { + return Err(PhysicalNodeError::ExpectedState); + }; + binary_values(operator, lhs, rhs) + } + QueryPlanNode::ReduceSum { grouping, .. } => { + let [PhysicalQueryOutput::Value(values)] = inputs else { + return Err(PhysicalNodeError::ExpectedState); + }; + reduce_sum_values(grouping, values) + } QueryPlanNode::ReadMaterialization { binding } => { let groups = self .context @@ -234,6 +248,148 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { } } +fn arithmetic(operator: &planner_types::pre_asap::ArithmeticOpKind, left: f64, right: f64) -> f64 { + use planner_types::pre_asap::ArithmeticOpKind::*; + match operator { + Add => left + right, + Sub => left - right, + Mul => left * right, + Div => left / right, + Mod => left % right, + Pow => left.powf(right), + Atan2 => left.atan2(right), + } +} + +fn binary_values( + operator: &planner_types::pre_asap::ArithmeticOpKind, + lhs: &PhysicalQueryOutput, + rhs: &PhysicalQueryOutput, +) -> Result { + type Labels = BTreeMap; + type Samples = BTreeMap<(Labels, i64), (f64, Option<(u64, u64)>)>; + fn flatten(values: &[(Labels, SummaryValue)]) -> Result { + let mut result = BTreeMap::new(); + for (labels, value) in values { + let SummaryValue::Points(points, coverage) = value else { + return Err(PhysicalNodeError::ExpectedState); + }; + let mut labels = labels.clone(); + labels.remove("__name__"); + for (timestamp, value) in points { + if result + .insert((labels.clone(), *timestamp), (*value, *coverage)) + .is_some() + { + return Err(PhysicalNodeError::Fallback( + "ambiguous default vector matching".into(), + )); + } + } + } + Ok(result) + } + fn points(values: Samples) -> PhysicalQueryOutput { + PhysicalQueryOutput::Value( + values + .into_iter() + .map(|((labels, timestamp), (value, coverage))| { + ( + labels, + SummaryValue::Points(vec![(timestamp, value)], coverage), + ) + }) + .collect(), + ) + } + match (lhs, rhs) { + (PhysicalQueryOutput::Scalar(a), PhysicalQueryOutput::Scalar(b)) => { + Ok(PhysicalQueryOutput::Scalar(arithmetic(operator, *a, *b))) + } + (PhysicalQueryOutput::Value(values), PhysicalQueryOutput::Scalar(scalar)) => Ok(points( + flatten(values)? + .into_iter() + .map(|(key, (value, coverage))| { + (key, (arithmetic(operator, value, *scalar), coverage)) + }) + .collect(), + )), + (PhysicalQueryOutput::Scalar(scalar), PhysicalQueryOutput::Value(values)) => Ok(points( + flatten(values)? + .into_iter() + .map(|(key, (value, coverage))| { + (key, (arithmetic(operator, *scalar, value), coverage)) + }) + .collect(), + )), + (PhysicalQueryOutput::Value(left), PhysicalQueryOutput::Value(right)) => { + let right = flatten(right)?; + Ok(points( + flatten(left)? + .into_iter() + .filter_map(|(key, (left, coverage))| { + let (right, right_coverage) = right.get(&key)?; + Some(( + key, + ( + arithmetic(operator, left, *right), + intersect_coverage(coverage, *right_coverage), + ), + )) + }) + .collect(), + )) + } + _ => Err(PhysicalNodeError::ExpectedState), + } +} + +fn intersect_coverage(left: Option<(u64, u64)>, right: Option<(u64, u64)>) -> Option<(u64, u64)> { + let (left, right) = (left?, right?); + let result = (left.0.max(right.0), left.1.min(right.1)); + (result.0 <= result.1).then_some(result) +} + +fn reduce_sum_values( + grouping: &control_plane::query_plan::PhysicalGrouping, + values: &[(BTreeMap, SummaryValue)], +) -> Result { + let control_plane::query_plan::PhysicalGrouping::Reduce(keys) = grouping else { + return Ok(PhysicalQueryOutput::Value(values.to_vec())); + }; + let mut groups = BTreeMap::new(); + for (labels, value) in values { + let SummaryValue::Points(points, coverage) = value else { + return Err(PhysicalNodeError::ExpectedState); + }; + let labels = labels + .iter() + .filter(|(name, _)| keys.contains(name)) + .map(|(k, v)| (k.clone(), v.clone())) + .collect::>(); + for (timestamp, value) in points { + groups + .entry((labels.clone(), *timestamp)) + .and_modify(|(sum, cover): &mut (f64, Option<(u64, u64)>)| { + *sum += value; + *cover = intersect_coverage(*cover, *coverage); + }) + .or_insert((*value, *coverage)); + } + } + Ok(PhysicalQueryOutput::Value( + groups + .into_iter() + .map(|((labels, timestamp), (sum, coverage))| { + ( + labels, + SummaryValue::Points(vec![(timestamp, sum)], coverage), + ) + }) + .collect(), + )) +} + fn execute_physical_query_plan( index: &SketchStore, entry: &control_plane::query_plan::QueryPlanEntry, @@ -253,6 +409,9 @@ fn execute_physical_query_plan( let output = physical_dag::execute(entry, &runtime) .map_err(|error| LoweringSkip::ExecuteFailed(format!("{error:?}")))?; match output { + PhysicalQueryOutput::Scalar(_) => Err(LoweringSkip::ExecuteFailed( + "scalar-only query is not a warm vector result".into(), + )), PhysicalQueryOutput::Value(values) => { let mut coverage = None; let mut series = Vec::new(); @@ -401,6 +560,110 @@ pub(crate) fn fold_coverage(coverage: &mut Option<(u64, u64)>, next: Option<(u64 #[cfg(test)] mod tests { use super::*; + use planner_types::pre_asap::ArithmeticOpKind; + + fn exact_points(metric: &str, service: &str, value: f64) -> PhysicalQueryOutput { + PhysicalQueryOutput::Value(vec![( + BTreeMap::from([ + ("__name__".into(), metric.into()), + ("service".into(), service.into()), + ]), + SummaryValue::Points(vec![(2000, value)], Some((1000, 2000))), + )]) + } + + // Arithmetic follows default vector matching, not positional row pairing. + #[test] + fn exact_binary_matches_labels_and_preserves_scalar_orientation() { + let left = exact_points("sum", "api", 24.0); + let right = exact_points("count", "api", 3.0); + let PhysicalQueryOutput::Value(result) = + binary_values(&ArithmeticOpKind::Div, &left, &right).unwrap() + else { + panic!("expected vector") + }; + assert_eq!( + result[0].0, + BTreeMap::from([("service".into(), "api".into())]) + ); + let SummaryValue::Points(points, coverage) = &result[0].1 else { + panic!("expected points") + }; + assert_eq!(points, &vec![(2000, 8.0)]); + assert_eq!(*coverage, Some((1000, 2000))); + let PhysicalQueryOutput::Value(result) = binary_values( + &ArithmeticOpKind::Div, + &PhysicalQueryOutput::Scalar(48.0), + &left, + ) + .unwrap() else { + panic!("expected vector") + }; + let SummaryValue::Points(points, _) = &result[0].1 else { + panic!("expected points") + }; + assert_eq!(points, &vec![(2000, 2.0)]); + let PhysicalQueryOutput::Value(result) = binary_values( + &ArithmeticOpKind::Div, + &left, + &exact_points("count", "worker", 3.0), + ) + .unwrap() else { + panic!("expected vector") + }; + assert!(result.is_empty()); + } + + // Zero denominators remain IEEE results; duplicate matches must fall back. + #[test] + fn exact_binary_handles_zero_and_rejects_ambiguous_matches() { + assert!(arithmetic(&ArithmeticOpKind::Div, 1.0, 0.0).is_infinite()); + assert!(arithmetic(&ArithmeticOpKind::Div, 0.0, 0.0).is_nan()); + let PhysicalQueryOutput::Value(mut values) = exact_points("a", "api", 1.0) else { + unreachable!() + }; + values.push(values[0].clone()); + assert!(binary_values( + &ArithmeticOpKind::Add, + &PhysicalQueryOutput::Value(values), + &PhysicalQueryOutput::Scalar(1.0) + ) + .is_err()); + assert_eq!( + intersect_coverage(Some((1000, 2000)), Some((1500, 2500))), + Some((1500, 2000)) + ); + assert_eq!(intersect_coverage(Some((1000, 2000)), None), None); + } + + // Rollup adds partial counts rather than counting the number of series. + #[test] + fn exact_rollup_adds_uneven_observation_counts() { + let values = [("a", 1.0), ("b", 3.0)] + .into_iter() + .map(|(instance, value)| { + ( + BTreeMap::from([ + ("service".into(), "api".into()), + ("instance".into(), instance.into()), + ]), + SummaryValue::Points(vec![(2000, value)], Some((1000, 2000))), + ) + }) + .collect::>(); + let PhysicalQueryOutput::Value(result) = reduce_sum_values( + &control_plane::query_plan::PhysicalGrouping::Reduce(vec!["service".into()]), + &values, + ) + .unwrap() else { + panic!("expected vector") + }; + assert_eq!(result.len(), 1); + let SummaryValue::Points(points, _) = &result[0].1 else { + panic!("expected points") + }; + assert_eq!(points, &vec![(2000, 4.0)]); + } use crate::storage_engines::sketch_db::data::{AggKind, SketchConfig}; use crate::storage_engines::sketch_db::index::{ AccuracyBound, Capability, SketchAlgorithm, SketchInstanceMetadata, SketchSampleState, diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index 722eac4c..c20c8e9b 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs @@ -233,6 +233,9 @@ impl GroupState { return None; }; let stat = match (readout, agg_type) { + (control_plane::query_plan::ExactReadout::Count, AggregationType::Sum) => { + asap_types::Statistic::Count + } ( control_plane::query_plan::ExactReadout::Sum, AggregationType::Sum | AggregationType::MultipleSum, @@ -260,7 +263,11 @@ impl GroupState { ("range_start_ms".to_string(), range_start_ms.to_string()), ("range_end_ms".to_string(), range_end_ms.to_string()), ]); - merged?.query_statistic(stat, key, &query_kwargs).ok() + let merged = merged?; + if readout == control_plane::query_plan::ExactReadout::Count { + return merged.aux_stats().count.map(|count| count as f64); + } + merged.query_statistic(stat, key, &query_kwargs).ok() } /// Coverage analog of `exact_value` — folds `(min_window_end_ms, diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 02f02804..ffec3adc 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -3924,6 +3924,70 @@ mod tests { drop(p); } + // Raw sample counts must survive production durable storage. + #[test] + fn raw_count_survives_disk_eviction() { + use crate::storage_engines::types::AggregationType; + let tmp = tempfile::TempDir::new().unwrap(); + let idx = Arc::new(SketchStore::new()); + // Register an ExactAgg(Sum) sid keyed by `zone`. + let mut m = meta(8001); + m.metric_name = "http_requests_total".into(); + m.group_by_keys = ["zone".to_string()].into_iter().collect(); + m.agg_kind = AggKind::ExactAgg { + agg_type: AggregationType::Sum, + parameters_canonical: String::new(), + spatial_filter_canonical: String::new(), + }; + idx.register(m); + let p = idx + .start_persistence(durable_cfg(tmp.path().to_path_buf())) + .unwrap(); + + let lv_zone = |v: &str| { + let mut x = BTreeMap::new(); + x.insert("zone".to_string(), v.to_string()); + x + }; + for i in 0..10u64 { + let s = i * 30_000; + idx.append_precompute( + 8001, + lv_zone("z0"), + (s, s + 30_000), + Box::new({ + let mut acc = crate::precompute_engine::operators::SumAccumulator::new(); + acc.update((i + 1) as f64); + acc.update(10.0); + acc + }), + ); + } + assert!( + wait_until( + || idx.approx_memory_bytes() == 0 && idx.list_sealed_epochs_len() == 0, + std::time::Duration::from_secs(5), + ), + "exact-agg windows never fully evicted" + ); + // Query the EVICTED portion [0, 150_000) — must come back from disk. + let series = idx.query_exact_agg_range(8001, 0, 150_000); + assert!( + !series.is_empty(), + "exact-agg query returned no result after flush and eviction" + ); + let (_label, samples) = &series[0]; + assert!( + samples.contains_key(&30_000), + "evicted exact-agg window missing from disk" + ); + let stats = samples[&30_000].aux_stats(); + assert_eq!(stats.count, Some(2)); + assert_eq!(stats.sum, Some(11.0)); + assert_eq!(stats.sum.unwrap() / stats.count.unwrap() as f64, 5.5); + drop(p); + } + /// BUG #3: the memory diagnostic + the flusher's memory-pressure /// trigger must account for `current_epoch`, not just sealed epochs. /// On origin/main `approx_memory_bytes()` sums ONLY sealed epochs, so diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs index eefc1b34..02478793 100644 --- a/data_plane/tests/asapquery_compatibility_process_e2e.rs +++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs @@ -108,6 +108,185 @@ fn is_warm(response: &Value) -> bool { }) } +// Three registered consumers must observe one raw SUM/count producer, including +// uneven instance sample counts and Remote Write retries. +#[tokio::test] +async fn shared_exact_dashboard_executes_selected_workload() { + let fallback_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let fallback_address = fallback_listener.local_addr().unwrap(); + tokio::spawn(async move { + axum::serve(fallback_listener, Router::new() + .route("/-/healthy", get(|| async { "healthy" })) + .route("/api/v1/query", get(|| async { Json(serde_json::json!({"status":"success","data":{"resultType":"vector","result":[]}})) }))) + .await.unwrap(); + }); + let output_dir = tempfile::tempdir().unwrap(); + let mut snapshot: Value = serde_json::from_str(include_str!( + "../../docs/examples/asapquery-compatibility-demo-snapshot.json" + )) + .unwrap(); + let sum = "sum by (service) (sum_over_time(asap_demo_gauge[5s]))"; + let count = "sum by (service) (count_over_time(asap_demo_gauge[5s]))"; + let mean = format!("{sum} / {count}"); + let mut entry = snapshot["query_workload"]["repeating_queries"][2].clone(); + snapshot["query_workload"]["repeating_queries"] = Value::Array( + [sum, count, mean.as_str()] + .into_iter() + .map(|query| { + entry["query"] = query.into(); + entry.clone() + }) + .collect(), + ); + let typed: control_plane::physical::compiler::BackendLocalPlanningSnapshot = + serde_json::from_value(snapshot.clone()).unwrap(); + let plan = typed.compile().unwrap(); + assert_eq!(plan.precompute_plan.materializations.len(), 1); + assert_eq!(plan.query_plan.entries.len(), 3); + let snapshot_path = output_dir.path().join("snapshot.json"); + std::fs::write(&snapshot_path, serde_json::to_vec(&snapshot).unwrap()).unwrap(); + let port = unused_port(); + let child = Command::new(env!("CARGO_BIN_EXE_data_plane")) + .args(["--profile", "asapquery", "--planning-snapshot"]) + .arg(&snapshot_path) + .args([ + "--forward-unsupported-queries", + "--prometheus-server", + &format!("http://{fallback_address}"), + ]) + .args(["--http-port", &port.to_string(), "--output-dir"]) + .arg(output_dir.path()) + .args([ + "--precompute-allowed-lateness-ms", + "0", + "--precompute-flush-interval-ms", + "25", + ]) + .stdout(Stdio::null()) + .stderr(Stdio::inherit()) + .spawn() + .unwrap(); + let mut child = ChildGuard(child); + let client = reqwest::Client::new(); + let backend = format!("http://127.0.0.1:{port}"); + wait_until_ready(&client, &format!("{backend}/api/v1/health"), &mut child.0).await; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as i64; + let base = now - now.rem_euclid(5000) - 20000; + let labeled = |service: &str, instance: &str, samples: &[(i64, f64)]| { + let mut item = series("asap_demo_gauge", samples); + item.labels.extend([ + Label { + name: "service".into(), + value: service.into(), + }, + Label { + name: "instance".into(), + value: instance.into(), + }, + ]); + item + }; + let request = WriteRequest { + timeseries: vec![ + labeled("api", "a", &[(base + 500, 10.0)]), + labeled( + "api", + "b", + &[(base + 600, 2.0), (base + 1700, 4.0), (base + 2900, 8.0)], + ), + labeled("worker", "c", &[(base + 700, 9.0), (base + 1900, 15.0)]), + ], + }; + assert_eq!(remote_write(&client, &backend, &request).await, 204); + assert_eq!(remote_write(&client, &backend, &request).await, 204); + let advance = WriteRequest { + timeseries: vec![ + labeled("api", "a", &[(base + 5500, 100.0)]), + labeled("api", "b", &[(base + 5500, 100.0)]), + labeled("worker", "c", &[(base + 5500, 100.0)]), + ], + }; + assert_eq!(remote_write(&client, &backend, &advance).await, 204); + let final_advance = WriteRequest { + timeseries: vec![ + labeled("api", "a", &[(base + 10500, 1000.0)]), + labeled("api", "b", &[(base + 10500, 1000.0)]), + labeled("worker", "c", &[(base + 10500, 1000.0)]), + ], + }; + assert_eq!(remote_write(&client, &backend, &final_advance).await, 204); + for (query, expected) in [ + (sum, [24.0, 24.0]), + (count, [4.0, 2.0]), + (mean.as_str(), [6.0, 12.0]), + ] { + let result = wait_for_warm_instant( + &client, + &backend, + query, + (base + 5000) as f64 / 1000.0, + &output_dir.path().join("query_engine.log"), + ) + .await; + let rows = result["data"]["result"].as_array().unwrap(); + assert_eq!(rows.len(), 2, "{query}: {result}"); + for row in rows { + let service = row["metric"]["service"].as_str().unwrap(); + assert_eq!(row["metric"].as_object().unwrap().len(), 1); + let value: f64 = row["value"][1].as_str().unwrap().parse().unwrap(); + assert_eq!( + value, + expected[usize::from(service == "worker")], + "{query}: {result}" + ); + assert_eq!( + row["value"][0].as_f64().unwrap(), + (base + 5000) as f64 / 1000.0 + ); + } + } + let result: Value = client + .get(format!("{backend}/api/v1/query_range")) + .query(&[ + ("query", mean.clone()), + ("start", ((base + 5000) as f64 / 1000.0).to_string()), + ("end", ((base + 10000) as f64 / 1000.0).to_string()), + ("step", "5".into()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!(is_warm(&result), "{result}"); + for row in result["data"]["result"].as_array().unwrap() { + let values = row["values"].as_array().unwrap(); + assert_eq!(values.len(), 2, "{result}"); + assert_eq!(values[1][1], "100", "{result}"); + } + // Unaligned intervals cannot be answered by whole tumbling states. + let result: Value = client + .get(format!("{backend}/api/v1/query")) + .query(&[ + ("query", mean), + ("time", ((base + 5001) as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert!( + !is_warm(&result), + "partial interval was incorrectly warm: {result}" + ); +} + async fn wait_for_warm_instant( client: &reqwest::Client, base: &str,