diff --git a/Cargo.lock b/Cargo.lock index e510f894..2884c716 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -307,6 +307,7 @@ dependencies = [ name = "asap-aware-mapping" version = "0.1.0" dependencies = [ + "asap-frontend-promql", "asap-types", "asap_sketchlib", "serde", @@ -350,6 +351,7 @@ dependencies = [ name = "asap-frontend-sql" version = "0.1.0" dependencies = [ + "asap-aware-mapping", "asap-sql-function-catalog", "asap-types", "datafusion", diff --git a/crates/asap-aware-mapping/Cargo.toml b/crates/asap-aware-mapping/Cargo.toml index c87f50dd..b2c03ed1 100644 --- a/crates/asap-aware-mapping/Cargo.toml +++ b/crates/asap-aware-mapping/Cargo.toml @@ -13,3 +13,6 @@ asap-types = { path = "../types" } thiserror = "2" serde = { version = "1", features = ["derive"] } serde_json = "1" + +[dev-dependencies] +asap-frontend-promql = { path = "../frontend-promql" } diff --git a/crates/asap-aware-mapping/src/accuracy.rs b/crates/asap-aware-mapping/src/accuracy.rs index 3dec32b2..235dbd8f 100644 --- a/crates/asap-aware-mapping/src/accuracy.rs +++ b/crates/asap-aware-mapping/src/accuracy.rs @@ -590,7 +590,7 @@ impl DefaultAccuracyModel { } let mut provenance = Vec::new(); let count = row_count(stats, &mut provenance); - let exact_local = ResultGuarantee::exact("ExactAggregate(MinMax)"); + let exact_local = ResultGuarantee::exact("ExactAggregate(Max)"); provenance.extend(composed_provenance( op, inputs, @@ -887,6 +887,43 @@ impl AccuracyModel for DefaultAccuracyModel { let same_metric = |metric: ErrorMetric| approximate.iter().all(|g| g.metric == metric); match op { + CompositionOperator::CheckedRelativeDivision => { + if inputs.len() != 2 || local.is_some() || !same_metric(ErrorMetric::RelativeValue) + { + return Err(unsupported( + "checked division requires two exact/relative-value operands".into(), + )); + } + let a = inputs[0] + .bound + .evaluate() + .ok_or_else(|| unsupported("unknown numerator bound".into()))?; + let b = inputs[1] + .bound + .evaluate() + .ok_or_else(|| unsupported("unknown denominator bound".into()))?; + if !(0.0..1.0).contains(&b) || a < 0.0 || !a.is_finite() { + return Err(unsupported("invalid relative division bounds".into())); + } + Ok(ResultGuarantee { + metric: ErrorMetric::RelativeValue, + bound: BoundExpr::Constant { + value: (a + b) / (1.0 - b) + 4.0 * f64::EPSILON, + }, + failure_probability: ProbabilityExpr::UnionBound { + terms: inputs + .iter() + .map(|g| g.failure_probability.clone()) + .collect(), + }, + provenance: composed_provenance( + op, + inputs, + &ResultGuarantee::exact("checked floating-point division"), + "checked_relative_division_union_bound", + ), + }) + } CompositionOperator::ApproximateAggregate => { let local = local.ok_or_else(|| { unsupported("approximate operator has no local guarantee to compose".into()) @@ -1126,6 +1163,25 @@ mod tests { use asap_types::post_asap::{GroupingStrategy, SketchKind}; use asap_types::workload::{DataDistribution, DataWorkload, Evidence, EvidenceSource}; + // Rank error cannot certify a numeric ratio; a same-sketch identity is not cancellation evidence. + #[test] + fn checked_division_propagates_value_bounds_and_rejects_rank_bounds() { + let op = CompositionOperator::CheckedRelativeDivision; + let inputs = [rel(0.01), rel(0.01)]; + let g = DefaultAccuracyModel + .propagate(&op, &inputs, None, &Default::default()) + .unwrap(); + assert!((g.bound.evaluate().unwrap() - 0.02 / 0.99).abs() < 1e-14); + let mut rank = inputs[0].clone(); + rank.metric = ErrorMetric::Rank; + assert!(DefaultAccuracyModel + .propagate(&op, &[rank.clone(), rank], None, &Default::default()) + .is_err()); + assert!(DefaultAccuracyModel + .propagate(&op, &[rel(0.01), rel(1.0)], None, &Default::default()) + .is_err()); + } + fn abs(bound: f64, delta: f64) -> ResultGuarantee { ResultGuarantee { metric: ErrorMetric::AbsoluteValue, diff --git a/crates/asap-aware-mapping/src/function_rules.rs b/crates/asap-aware-mapping/src/function_rules.rs index 82ad4e20..0b96e018 100644 --- a/crates/asap-aware-mapping/src/function_rules.rs +++ b/crates/asap-aware-mapping/src/function_rules.rs @@ -15,9 +15,13 @@ pub(crate) fn function_rules(intent: &AggIntent) -> Option { CompositionOperator::ExactSum, Some((ExactKind::Sum, ExactParams::Sum)), ), - AggIntent::Min { .. } | AggIntent::Max { .. } => ( + AggIntent::Min { .. } => ( CompositionOperator::ExactExtremum, - Some((ExactKind::MinMax, ExactParams::MinMax)), + Some((ExactKind::Min, ExactParams::Min)), + ), + AggIntent::Max { .. } => ( + CompositionOperator::ExactExtremum, + Some((ExactKind::Max, ExactParams::Max)), ), AggIntent::Avg { .. } => (CompositionOperator::ExactAverage, None), AggIntent::Rate => ( @@ -39,3 +43,20 @@ pub(crate) fn function_rules(intent: &AggIntent) -> Option { accumulator, }) } + +#[cfg(test)] +mod tests { + use super::*; + // The maintained extrema state must encode the direction independently of query text. + #[test] + fn minimum_and_maximum_have_distinct_accumulator_contracts() { + assert_ne!( + function_rules(&AggIntent::Min { col: None }) + .unwrap() + .accumulator, + function_rules(&AggIntent::Max { col: None }) + .unwrap() + .accumulator + ); + } +} diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index da96958a..b3262c28 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -254,3 +254,5 @@ pub use summary_maintenance_lifecycle::{ WorkloadDemand, }; pub use topk_reuse::TopKLimitReuseStrategy; + +pub mod maintained_population; diff --git a/crates/asap-aware-mapping/src/maintained_population.rs b/crates/asap-aware-mapping/src/maintained_population.rs new file mode 100644 index 00000000..ba9de4f6 --- /dev/null +++ b/crates/asap-aware-mapping/src/maintained_population.rs @@ -0,0 +1,459 @@ +//! Shared maintained-population candidates over canonical relational IR. +use crate::replacement::{ + Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG, +}; +use asap_types::post_asap::{ + maintained_population::*, ExecutionTiming, ResultGuarantee, SummaryExpr, SummaryFamilyType, + SummaryField, SummaryNode, SummarySchema, ValueOperation, +}; +use asap_types::pre_asap::{ + AggIntent, CompareOpKind, DataType, QueryExpr, Reduction, ScalarValue, Schema, Source, +}; +use std::rc::Rc; + +fn plain(schema: Schema) -> SummarySchema { + SummarySchema { + time_index: schema.time_index, + fields: schema + .columns + .into_iter() + .map(|c| SummaryField { + name: c.name, + dtype: SummaryFamilyType::Plain(c.dtype), + nullable: c.nullable, + }) + .collect(), + } +} + +fn strip_projection(mut root: &QueryExpr) -> &QueryExpr { + while let QueryExpr::Project { child, .. } = root { + root = child; + } + root +} + +fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadout, Rc)> { + let root = strip_projection(root); + let (source, grouping, readout, value_column) = match root { + QueryExpr::Aggregate { + child, + reduction: Reduction::Reduce(grouping), + measures, + having: None, + .. + } => { + let [intent] = measures.as_slice() else { + return None; + }; + let (col, readout) = match intent { + AggIntent::Quantile { q, col, .. } if q.is_finite() => { + (*col, PopulationReadout::Quantile { q: *q }) + } + AggIntent::Sum { col } => (*col, PopulationReadout::Sum), + AggIntent::Count { .. } => (None, PopulationReadout::Count), + AggIntent::Avg { col } => (*col, PopulationReadout::Average), + _ => return None, + }; + let schema = child.output_schema().ok()?; + if col.is_some_and(|c| schema.columns.get(c).is_none()) { + return None; + } + (child, grouping, readout, col) + } + QueryExpr::Limit { + n, + offset: 0, + child, + } => { + let QueryExpr::Sort { + child, + keys, + partition_by, + } = child.as_ref() + else { + return None; + }; + let [key] = keys.as_slice() else { + return None; + }; + let QueryExpr::Column(col) = &key.expr else { + return None; + }; + if key.ascending { + return None; + } + ( + child, + partition_by, + PopulationReadout::TopK { k: *n }, + Some(*col), + ) + } + _ => return None, + }; + if let QueryExpr::Scan { + source: Source::Table { .. }, + schema, + .. + } = source.as_ref() + { + let value_column = value_column.or_else(|| { + schema + .columns + .iter() + .position(|c| c.dtype == DataType::Float64 && !c.nullable) + })?; + let population = MaintainedPopulation { + input: PopulationInput::Rows { + input: Rc::clone(source), + value_column, + grouping: grouping.clone(), + }, + max_k: 0, + quantiles: false, + }; + if !schema.closed || !population.matches_input(source) { + return None; + } + return Some((population, readout, Rc::clone(source))); + } + let QueryExpr::Scan { + source: Source::TimeSeries { metric }, + predicates, + schema, + } = source.as_ref() + else { + return None; + }; + if value_column.is_some_and(|c| schema.columns.get(c).is_none_or(|c| c.name != "value")) { + return None; + } + // Open time-series schemas distinguish instant PromQL populations from table rows. + if metric.is_empty() || schema.closed || schema.time_index.is_none() { + return None; + } + let label = |col: usize| -> Option { + let c = schema.columns.get(col)?; + (c.dtype == DataType::Utf8).then(|| c.name.clone()) + }; + let mut matchers = Vec::new(); + for predicate in predicates { + let QueryExpr::Compare { left, op, right } = predicate.0.as_ref() else { + return None; + }; + let (QueryExpr::Column(col), QueryExpr::Literal(ScalarValue::Utf8(value))) = + (left.as_ref(), right.as_ref()) + else { + return None; + }; + let operation = match op { + CompareOpKind::Eq => CurrentSeriesMatch::Equal, + CompareOpKind::Ne => CurrentSeriesMatch::NotEqual, + CompareOpKind::Regex => CurrentSeriesMatch::Regex, + CompareOpKind::NotRegex => CurrentSeriesMatch::NotRegex, + _ => return None, + }; + matchers.push(CurrentSeriesMatcher { + label: label(*col)?, + value: value.clone(), + operation, + }); + } + matchers.sort(); + matchers.dedup(); + let mut labels = grouping + .keys() + .iter() + .map(|c| label(*c)) + .collect::>>()?; + labels.sort(); + labels.dedup(); + Some(( + MaintainedPopulation { + input: PopulationInput::CurrentSeries(CurrentSeriesInput { + metric: metric.clone(), + matchers, + grouping: labels, + without: grouping.is_without(), + lookback_ms: 300_000, + }), + max_k: 0, + quantiles: false, + }, + readout, + Rc::clone(source), + )) +} + +/// Workload-aware rule: compatible readouts share one retractable population. +/// Deployments opt in by registering this strategy when they can maintain complete +/// population updates and price the maintenance/readout boundary. +/// The population is exact; max_k bounds the shared readout cache, not its members. +pub struct MaintainedPopulationStrategy { + roots: Vec>, +} +impl MaintainedPopulationStrategy { + pub fn new(roots: &[Rc]) -> Self { + Self { + roots: roots.to_vec(), + } + } + pub fn candidate(&self, root: &Rc) -> Option> { + if let QueryExpr::Project { + cols, + qualifier, + child, + } = root.as_ref() + { + let child = self.candidate(child)?; + return Some(Rc::new(SummaryNode { + guarantee: child.guarantee.clone(), + schema: plain(root.output_schema().ok()?), + expr: SummaryExpr::ValueOperation { + child, + operation: ValueOperation::Project { + cols: cols.clone(), + qualifier: qualifier.clone(), + }, + timing: ExecutionTiming::ReadTime, + }, + })); + } + let (mut population, readout, source) = recognize(root)?; + let identity = population.clone(); + for other in self.roots.iter().chain(std::iter::once(root)) { + if let Some((p, r, _)) = recognize(other) { + if p == identity { + match r { + PopulationReadout::Quantile { .. } => population.quantiles = true, + PopulationReadout::TopK { k } => population.max_k = population.max_k.max(k), + PopulationReadout::Sum + | PopulationReadout::Count + | PopulationReadout::Average => {} + } + } + } + } + let input_schema = plain(source.output_schema().ok()?); + let scan = Rc::new(SummaryNode { + expr: SummaryExpr::KeepPreAsap(source), + schema: input_schema.clone(), + guarantee: Some(ResultGuarantee::exact("source samples")), + }); + let maintained = Rc::new(SummaryNode { + expr: SummaryExpr::ValueOperation { + child: scan, + operation: ValueOperation::MaintainPopulation { population }, + timing: ExecutionTiming::MaintenanceTime, + }, + schema: input_schema, + guarantee: Some(ResultGuarantee::exact( + "exact members under the declared population semantics", + )), + }); + Some(Rc::new(SummaryNode { + expr: SummaryExpr::ValueOperation { + child: maintained, + operation: ValueOperation::ReadPopulation { readout }, + timing: ExecutionTiming::ReadTime, + }, + schema: plain(root.output_schema().ok()?), + guarantee: Some(ResultGuarantee::exact("exact current-population readout")), + })) + } +} +impl ReplacementStrategy for MaintainedPopulationStrategy { + fn matches(&self, target: &TargetSubDAG<'_>) -> bool { + recognize(target.root).is_some() + } + fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec { + self.candidate(target.root) + .map(|node| ReplacementSubDAG { + strategy: "MaintainedPopulationStrategy", + replacement: Replacement::Summary(node), + provenance: ReplacementProvenance::SummaryImplementation, + rationale: + "share an exact maintained population across compatible aggregate readouts" + .into(), + }) + .into_iter() + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use asap_types::post_asap::{compile_executable_dag, share_common_summary_subtrees}; + fn lower(q: &str) -> Rc { + Rc::new( + asap_frontend_promql::lower_promql(q, asap_types::types::AccuracyTarget::Exact) + .unwrap(), + ) + } + + // Instant scalar aggregations share the same retractable series population. + #[test] + fn instant_sum_count_average_are_typed_current_series_candidates() { + let roots: Vec<_> = [ + "sum(a)", + "count(a)", + "avg(a)", + "sum by(job)(a)", + "count by(job)(a)", + "avg by(job)(a)", + ] + .map(lower) + .into(); + let rule = MaintainedPopulationStrategy::new(&roots); + for root in roots { + let candidate = rule + .candidate(&root) + .expect("current-series rule candidate"); + compile_executable_dag(&candidate).expect("typed executable DAG"); + } + } + + // Different readout parameters retain one shared maintenance producer in the DAG. + #[test] + fn quantiles_and_topk_share_a_planner_population() { + let roots: Vec<_> = [ + "quantile by(job)(0.5,a)", + "quantile by(job)(0.99,a)", + "topk by(job)(1,a)", + "topk by(job)(5,a)", + ] + .map(lower) + .into(); + let strategy = MaintainedPopulationStrategy::new(&roots); + let space = crate::search_workload_with( + roots + .iter() + .enumerate() + .map(|(i, r)| (i, Rc::clone(r))) + .collect(), + &[Box::new(MaintainedPopulationStrategy::new(&roots))], + ); + assert!(space + .groups() + .flat_map(|g| &g.candidates) + .any(|c| c.strategy == "MaintainedPopulationStrategy")); + let plans = share_common_summary_subtrees( + roots + .iter() + .enumerate() + .map(|(i, r)| (i, strategy.candidate(r).unwrap())) + .collect(), + ); + let mut producers = Vec::new(); + for (_, plan) in &plans { + compile_executable_dag(plan).unwrap(); + let SummaryExpr::ValueOperation { + child, + operation: ValueOperation::ReadPopulation { .. }, + .. + } = &plan.expr + else { + panic!("missing typed readout") + }; + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainPopulation { population }, + .. + } = &child.expr + else { + panic!("missing maintained population") + }; + assert_eq!(population.max_k, 5); + assert!(population.quantiles); + producers.push(Rc::as_ptr(child)); + } + assert!(producers.iter().all(|p| *p == producers[0])); + } + + // Source/group/matcher identity separates populations; temporal/nested operations are not instant populations. + #[test] + fn rule_respects_population_semantics() { + let roots: Vec<_> = [ + "topk(5,a)", + "topk(10,b)", + "quantile by(job)(0.5,a)", + "quantile(0.9,a{job=\"api\"})", + ] + .map(lower) + .into(); + let strategy = MaintainedPopulationStrategy::new(&roots); + let (p, _, _) = recognize(&roots[0]).unwrap(); + assert!(matches!(p.input, PopulationInput::CurrentSeries(ref s) if s.grouping.is_empty())); + let candidate = strategy.candidate(&roots[0]).unwrap(); + let SummaryExpr::ValueOperation { child, .. } = &candidate.expr else { + unreachable!() + }; + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainPopulation { population }, + .. + } = &child.expr + else { + unreachable!() + }; + assert_eq!(population.max_k, 5); + assert!(!population.quantiles); + for q in [ + "quantile_over_time(0.5,a[1m])", + "quantile(0.5,sum by(job)(a))", + "topk(5,a offset 1m)", + "topk(5,a @ 100)", + "bottomk(5,a)", + ] { + assert!( + strategy.candidate(&lower(q)).is_none(), + "unexpected current population for {q}" + ); + } + let q = lower("quantile without(instance)(0.5,a{job=~\"api.*\"})"); + let (p, _, _) = recognize(&q).unwrap(); + let PopulationInput::CurrentSeries(p) = p.input else { + panic!("series input") + }; + assert!(p.without); + assert_eq!(p.grouping, ["instance"]); + assert_eq!(p.matchers[0].operation, CurrentSeriesMatch::Regex); + } + // A readout cannot reinterpret arbitrary rows as maintained state or exceed its producer's contract. + #[test] + fn malformed_population_dags_fail_closed() { + let root = lower("topk(5,a)"); + let strategy = MaintainedPopulationStrategy::new(std::slice::from_ref(&root)); + let candidate = strategy.candidate(&root).unwrap(); + let mut bad = (*candidate).clone(); + let SummaryExpr::ValueOperation { operation, .. } = &mut bad.expr else { + unreachable!() + }; + *operation = ValueOperation::ReadPopulation { + readout: PopulationReadout::TopK { k: 6 }, + }; + assert!(compile_executable_dag(&Rc::new(bad.clone())).is_err()); + let SummaryExpr::ValueOperation { + child, operation, .. + } = &mut bad.expr + else { + unreachable!() + }; + *operation = ValueOperation::ReadPopulation { + readout: PopulationReadout::TopK { k: 5 }, + }; + let producer = Rc::make_mut(child); + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainPopulation { population }, + .. + } = &mut producer.expr + else { + unreachable!() + }; + let PopulationInput::CurrentSeries(spec) = &mut population.input else { + unreachable!() + }; + spec.metric = "b".into(); + assert!(compile_executable_dag(&Rc::new(bad)).is_err()); + } +} diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 484095b6..075f13a1 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -632,7 +632,7 @@ pub trait ReplacementStrategy { #[derive(Debug, Clone, PartialEq)] pub enum Implementation { /// An exact **mergeable** accumulator (partial state ≡ the value - /// itself: `Sum` / `Count` / `MinMax` / `Rate` / `Increase`). The + /// itself: `Sum` / `Count` / `Min` / `Max` / `Rate` / `Increase`). The /// built state *is* the answer already — no `SummaryEstimate` readout /// step. ExactAggregate { @@ -1366,6 +1366,25 @@ impl<'a> SketchAlgorithmStrategy<'a> { /// differs — see [`realize_child_with`]). fn propose_with(&self, root: &Rc, intent_override: Option<&AggIntent>) -> Proposals { let mut proposals = Proposals::default(); + if let Ok(Some(node)) = exact_topk_over_temporal_values(root, self.models) { + proposals.candidates.push(ReplacementSubDAG { + replacement: Replacement::Summary(node), + strategy: "SketchAlgorithmStrategy", + provenance: ReplacementProvenance::SummaryImplementation, + rationale: "select exact Top-K from independently maintained temporal values" + .into(), + }); + } + if intent_override.is_none() { + if let Ok(Some(node)) = realize_temporal_average(root, self.models, None) { + proposals.candidates.push(ReplacementSubDAG { + replacement: Replacement::Summary(node), + strategy: "SketchAlgorithmStrategy", + provenance: ReplacementProvenance::SummaryImplementation, + rationale: "read temporal average from sum/count only within the finite arithmetic domain; otherwise execute the original average".into(), + }); + } + } if intent_override.is_none() && is_supported_exact_binary(root) { if let Ok(Some(node)) = realize_binary(root, self.models, None) { proposals.candidates.push(ReplacementSubDAG { @@ -1663,11 +1682,93 @@ pub(crate) fn realize_child( /// re-splitting for its own approximate children) under the allocated /// budget. A child whose declared target is `Exact` keeps it: an allocation /// never approximates something the caller declared exact. +fn exact_topk_over_temporal_values( + root: &Rc, + models: Models<'_>, +) -> Result>, ImplementError> { + let QueryExpr::Aggregate { + reduction, + measures, + output_names, + having: None, + child, + } = root.as_ref() + else { + return Ok(None); + }; + if !matches!( + measures.as_slice(), + [AggIntent::TopK { + accuracy: AccuracyTarget::Exact, + .. + }] + ) { + return Ok(None); + } + let QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + child: input, + .. + } = child.as_ref() + else { + return Ok(None); + }; + if !matches!(input.as_ref(), QueryExpr::TimeRange { .. }) { + return Ok(None); + } + let values = realize_child_with(child, models, Some(&AccuracyTarget::Exact))?; + if matches!(values.expr, SummaryExpr::KeepPreAsap(_)) + || !values + .guarantee + .as_ref() + .is_some_and(ResultGuarantee::is_exact) + { + return Ok(None); + } + let values = finalize_exact_accumulator(values, child)?; + let node = Rc::new(SummaryNode { + guarantee: values.guarantee.clone(), + schema: lift(&root.output_schema()?), + expr: SummaryExpr::ValueOperation { + child: values, + operation: ValueOperation::Exact(ExactOperation::Aggregate { + reduction: reduction.clone(), + measures: measures.clone(), + output_names: output_names.clone(), + having: None, + }), + timing: ExecutionTiming::ReadTime, + }, + }); + validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?; + Ok(Some(node)) +} + +fn realize_temporal_average( + root: &Rc, + models: Models<'_>, + target: Option<&AccuracyTarget>, +) -> Result>, ImplementError> { + let Some(components) = crate::rewrite::temporal_average_components(root) else { + return Ok(None); + }; + let mut node = realize_child_with(&components, models, target)?; + let SummaryExpr::BinaryOp { operator, .. } = &mut Rc::make_mut(&mut node).expr else { + return Ok(None); + }; + operator.checked_finite_division = true; + validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?; + Ok(Some(node)) +} + pub(crate) fn realize_child_with( root: &Rc, models: Models<'_>, end_to_end_target: Option<&AccuracyTarget>, ) -> Result, ImplementError> { + if let Some(node) = realize_temporal_average(root, models, end_to_end_target)? { + return Ok(node); + } if let Some(composed) = realize_binary(root, models, end_to_end_target)? { return Ok(composed); } @@ -1818,6 +1919,20 @@ fn realize_binary( } } } + // Only the domain-proven ratio path may consume approximate operands. + // Runtime finite/nonzero checks alone do not establish quantile error bounds. + if ratio_domains.is_none() + && matches!(op, BinaryOpKind::Arithmetic(ArithmeticOpKind::Div)) + && [&lhs_node, &rhs_node].iter().any(|node| { + !node + .guarantee + .as_ref() + .is_some_and(ResultGuarantee::is_exact) + }) + { + return Ok(None); + } + let lhs_accelerated = lhs_scalar || !matches!(lhs_node.expr, SummaryExpr::KeepPreAsap(_)); let rhs_accelerated = rhs_scalar || !matches!(rhs_node.expr, SummaryExpr::KeepPreAsap(_)); if !lhs_accelerated || !rhs_accelerated { @@ -1866,6 +1981,8 @@ fn realize_binary( lhs: lhs_node, rhs: rhs_node, operator: asap_types::post_asap::BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, kind: op.clone(), vector_match: vector_match.clone(), }, @@ -2332,8 +2449,15 @@ fn realize_physical_summary_input( /// `SummaryEstimate` readout when `estimate` is set. // Retain the exact expression and schema while placing its value production // on the update path. Read-time consumers keep their original shared nodes. -fn maintenance_exact_values(node: Rc) -> Rc { +fn maintenance_exact_values(node: Rc) -> Option> { let expr = match &node.expr { + // These guards can fall back at read time, but cannot recover a parent + // sketch after an invalid value has entered its maintained state. + SummaryExpr::BinaryOp { operator, .. } + if operator.checked_finite_division || operator.checked_relative_division => + { + return None; + } SummaryExpr::BinaryOp { lhs, rhs, operator, .. } if operator.vector_match.is_none() @@ -2347,8 +2471,8 @@ fn maintenance_exact_values(node: Rc) -> Rc { .is_some_and(ResultGuarantee::is_exact) => { SummaryExpr::BinaryOp { - lhs: maintenance_exact_values(lhs.clone()), - rhs: maintenance_exact_values(rhs.clone()), + lhs: maintenance_exact_values(lhs.clone())?, + rhs: maintenance_exact_values(rhs.clone())?, operator: operator.clone(), timing: ExecutionTiming::MaintenanceTime, } @@ -2371,13 +2495,13 @@ fn maintenance_exact_values(node: Rc) -> Rc { timing: ExecutionTiming::MaintenanceTime, } } - _ => return node, + _ => return Some(node), }; - Rc::new(SummaryNode { + Some(Rc::new(SummaryNode { expr, schema: node.schema.clone(), guarantee: node.guarantee.clone(), - }) + })) } #[allow(clippy::too_many_arguments)] @@ -2443,7 +2567,10 @@ fn construct_summary_agg( // an exact scalar accumulator currently stores its value directly. let bound_child = finalize_exact_accumulator_at(bound_child, &input.child, ExecutionTiming::MaintenanceTime)?; - let bound_child = maintenance_exact_values(bound_child); + let bound_child = match maintenance_exact_values(bound_child) { + Some(child) => child, + None => keep_pre_asap(&input.child)?, + }; // ── Guarantee (issue #172) ────────────────────────────────────────── // Derived *before* the node exists, so an illegal composition is never @@ -5663,6 +5790,78 @@ mod tests { })) } + // Finite samples can overflow a sum although their native average is finite. + #[test] + fn temporal_average_requires_finite_division_guard() { + let root = Rc::new( + asap_frontend_promql::lower_promql("avg_over_time(a[5m])", AccuracyTarget::Exact) + .unwrap(), + ); + let candidates = + SketchAlgorithmStrategy::default_cost_model().replacements(&TargetSubDAG::new(&root)); + let operator = candidates + .iter() + .find_map(|c| match &c.replacement { + Replacement::Summary(node) => match &node.expr { + SummaryExpr::BinaryOp { operator, .. } => Some(operator), + _ => None, + }, + _ => None, + }) + .expect("maintained average candidate"); + assert!(operator.checked_finite_division); + assert!( + crate::rewrite::SemanticEquivalentRewriteStrategy + .replacements(&TargetSubDAG::new(&root)) + .is_empty(), + "an unconditional pre-ASAP rewrite would bypass the runtime guard" + ); + } + + // Exact Top-K consumes the Planner's maintained temporal values. + #[test] + fn exact_temporal_topk_has_a_maintained_value_candidate() { + for query in [ + "topk(5, sum_over_time(a[5m]))", + "topk by(job)(5, count_over_time(a[5m]))", + ] { + let root = + Rc::new(asap_frontend_promql::lower_promql(query, AccuracyTarget::Exact).unwrap()); + let models = Models::with_default_accuracy(&crate::cost_model::DefaultCostModel); + let node = exact_topk_over_temporal_values(&root, models) + .unwrap() + .expect("exact Top-K candidate"); + assert!(node.guarantee.as_ref().unwrap().is_exact()); + assert!(matches!( + node.expr, + SummaryExpr::ValueOperation { + operation: ValueOperation::Exact(ExactOperation::Aggregate { .. }), + .. + } + )); + asap_types::post_asap::compile_executable_dag(&node).unwrap(); + } + } + + // Runtime division guards do not establish the input quantiles' error bounds. + #[test] + fn quantile_ratio_without_input_proof_keeps_native_execution() { + let target = AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }; + for query in [ + "quantile_over_time(0.5,a[5m]) / quantile_over_time(0.9,a[5m])", + "avg_over_time(a[5m]) / quantile_over_time(0.5,a[5m])", + ] { + let root = Rc::new(asap_frontend_promql::lower_promql(query, target.clone()).unwrap()); + let models = Models::with_default_accuracy(&crate::cost_model::DefaultCostModel); + assert!(realize_binary(&root, models, Some(&target)) + .unwrap() + .is_none()); + } + } + #[test] fn relational_join_predicate_requires_and_normalizes_cross_input_columns() { let forward = normalize_cross_input_equi_predicate(&equi_pred(1, 3), 2, 4) @@ -5777,8 +5976,8 @@ mod tests { ), // exact mergeable accumulators (A::Sum { col: None }, Acc(E::Sum)), - (A::Min { col: None }, Acc(E::MinMax)), - (A::Max { col: None }, Acc(E::MinMax)), + (A::Min { col: None }, Acc(E::Min)), + (A::Max { col: None }, Acc(E::Max)), (A::Rate, Acc(E::Rate)), (A::IRate, Acc(E::IRate)), (A::Increase, Acc(E::Increase)), diff --git a/crates/asap-aware-mapping/src/rewrite.rs b/crates/asap-aware-mapping/src/rewrite.rs index 459f2e8a..1c481d86 100644 --- a/crates/asap-aware-mapping/src/rewrite.rs +++ b/crates/asap-aware-mapping/src/rewrite.rs @@ -26,20 +26,12 @@ //! that reshaping — see "Non-goals" below for why it does not also decide //! whether the reshaping is worth it. //! -//! ## Scope: `by(...)` grouping only (issue #253's own scope note) +//! ## Scope //! -//! [`AvgToSumOverCountStrategy::matches`] additionally requires -//! `Reduction::Reduce(by)` with `by` an ordinary (non-`without`) grouping — -//! narrower than [`SketchAlgorithmStrategy`]'s `bindable_intent`, which is -//! `Reduction`-agnostic. Two concrete reasons, not stylistic ones: +//! Ordinary `by(...)` averages use a schema-preserving projection. Temporal +//! Float64 temporal averages require a typed finite-division guard; their +//! sum/count components are never exported as an unconditional logical rewrite. //! -//! - **`Reduction::PerEntity`** (`rate`/`increase`/`*_over_time`) is -//! single-measure by construction — -//! [`aggregate_output_schema`](asap_types::pre_asap::query_expr::aggregate_output_schema) -//! `debug_assert!`s exactly one measure for it. This rewrite's entire -//! point is introducing a *second* measure (`Count` alongside `Sum`) -//! under the same node, which would violate that invariant outright, not -//! just drift a schema detail. //! - **`without(...)` grouping** leaves an `Aggregate`'s own output schema //! *open* (`closed: false`, see `without_output_schema`), while the //! `Project` this strategy always wraps the rewrite in forces @@ -135,6 +127,52 @@ fn avg_rewrite_target(node: &QueryExpr) -> Option<(usize, Option)> { /// types a `Div` of two `Int64` operands as `Int64` — the explicit operand /// `Cast` is what keeps both the division and rewritten `avg` column /// `Float64` the way the original always was, not an incidental extra step). +// These are conditional physical components, never an unconditional Rewrite. +// The caller must attach the finite-division execution guard before admission. +pub(crate) fn temporal_average_components(root: &Rc) -> Option> { + let QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures, + child, + having: None, + .. + } = root.as_ref() + else { + return None; + }; + let [AggIntent::Avg { col }] = measures.as_slice() else { + return None; + }; + if !matches!(child.as_ref(), QueryExpr::TimeRange { .. }) { + return None; + } + let schema = child.output_schema().ok()?; + let value = schema + .columns + .get(col.or_else(|| schema.column_id("value"))?)?; + if value.nullable || value.dtype != DataType::Float64 { + return None; + } + let aggregate = |intent| { + Rc::new(QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::clone(child), + }) + }; + let rewritten = Rc::new(QueryExpr::BinaryOp { + op: BinaryOpKind::Arithmetic(ArithmeticOpKind::Div), + lhs: aggregate(AggIntent::Sum { col: *col }), + rhs: aggregate(AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }), + vector_match: None, + }); + (root.output_schema().ok()? == rewritten.output_schema().ok()?).then_some(rewritten) +} + fn build_rewrite(root: &Rc) -> Option> { let (group_count, col) = avg_rewrite_target(root)?; let QueryExpr::Aggregate { @@ -378,6 +416,28 @@ mod tests { } } + // Temporal averages expose two single-measure children without closing labels. + #[test] + fn temporal_average_components_preserves_schema_and_exposes_sum_count() { + let root = Rc::new( + asap_frontend_promql::lower_promql( + "avg_over_time(a{job=\"api\"}[5m])", + AccuracyTarget::Exact, + ) + .unwrap(), + ); + assert!(SemanticEquivalentRewriteStrategy + .replacements(&TargetSubDAG::new(&root)) + .is_empty()); + let rewritten = + temporal_average_components(&root).expect("conditional sum/count components"); + assert_eq!( + root.output_schema().unwrap(), + rewritten.output_schema().unwrap() + ); + assert!(matches!(rewritten.as_ref(), QueryExpr::BinaryOp { .. })); + } + // ── matches ────────────────────────────────────────────────────────── #[test] diff --git a/crates/asap-aware-mapping/src/rollup.rs b/crates/asap-aware-mapping/src/rollup.rs index a5abae8a..351515dd 100644 --- a/crates/asap-aware-mapping/src/rollup.rs +++ b/crates/asap-aware-mapping/src/rollup.rs @@ -143,7 +143,7 @@ fn bindable_grouped_aggregate( /// reasoning behind each arm. `None` for any intent this module does not /// (yet) know a correct combinator for — including every intent /// `agg_is_mergeable` permits but this module doesn't specifically handle -/// (`Rate`, and everything outside the `Sum`/`Count`/`MinMax`/`Increase` +/// (`Rate`, and everything outside the `Sum`/`Count`/`Min`/`Max`/`Increase` /// vocabulary `agg_is_mergeable`'s own doc names) — so `is_legal_rollup_source` /// (which calls this) is *strictly narrower* than `agg_is_mergeable` alone, /// deliberately: `agg_is_mergeable` answers "does *some* partial-state merge diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs index 00ef21d8..9cfc370d 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs @@ -3004,6 +3004,8 @@ mod tests { lhs: Rc::clone(&operand), rhs: operand, operator: asap_types::post_asap::BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, kind: asap_types::pre_asap::BinaryOpKind::Arithmetic( asap_types::pre_asap::ArithmeticOpKind::Add, ), diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs index 5fc02f72..391a6795 100644 --- a/crates/frontend-promql/tests/promql_lowering.rs +++ b/crates/frontend-promql/tests/promql_lowering.rs @@ -380,6 +380,55 @@ fn count_over_rate_keeps_both_levels() { )); } +#[test] +fn count_over_distinct_over_time_preserves_both_aggregates() { + // One series with window samples [1, 2] produces one distinct-count + // result (value 2). The outer count counts that one series, yielding 1. + for (query, reduction) in [ + ( + "count(distinct_over_time(unique_users[5m]))", + Reduction::by(vec![]), + ), + ( + "count by (job) (distinct_over_time(unique_users[5m]))", + Reduction::by(vec![2]), + ), + ] { + let tree = lower(query); + let QueryExpr::Aggregate { + measures, + reduction: actual, + child, + .. + } = &tree + else { + panic!("expected outer Count: {tree:?}"); + }; + assert!( + matches!(measures.as_slice(), [AggIntent::Count { .. }]), + "{query}: {tree:?}" + ); + assert_eq!(actual, &reduction, "{query}"); + let QueryExpr::Aggregate { + measures, + reduction, + child, + .. + } = child.as_ref() + else { + panic!("expected inner per-series Cardinality: {tree:?}"); + }; + assert!( + matches!(measures.as_slice(), [AggIntent::Cardinality { .. }]), + "{query}: {tree:?}" + ); + assert_eq!(reduction, &Reduction::PerEntity, "{query}"); + assert!( + matches!(child.as_ref(), QueryExpr::TimeRange { range, .. } if range.as_secs() == 300) + ); + } +} + // ── count / cardinality ─────────────────────────────────────────────────────── // Both selector fast paths and recursive vector expressions count rows, not values. diff --git a/crates/frontend-sql/Cargo.toml b/crates/frontend-sql/Cargo.toml index 57062f5d..66179346 100644 --- a/crates/frontend-sql/Cargo.toml +++ b/crates/frontend-sql/Cargo.toml @@ -16,6 +16,7 @@ datafusion = "43" serde_json = "1" [dev-dependencies] +asap-aware-mapping = { path = "../asap-aware-mapping" } tokio = { version = "1", features = ["rt", "macros", "rt-multi-thread"] } # bgp_jan2024_workload corpus is sourced verbatim as YAML (ASAPQuery PR #561) # rather than transcribed into the flat .sql shape the other corpora use. diff --git a/crates/frontend-sql/tests/maintained_population.rs b/crates/frontend-sql/tests/maintained_population.rs new file mode 100644 index 00000000..6ad7e987 --- /dev/null +++ b/crates/frontend-sql/tests/maintained_population.rs @@ -0,0 +1,182 @@ +//! SQL and PromQL use the same shared-state rule without sharing membership semantics. +use asap_aware_mapping::maintained_population::MaintainedPopulationStrategy; +use asap_frontend_sql::{lower_sql, SqlCatalog}; +use asap_types::{ + post_asap::{ + compile_executable_dag, + maintained_population::{MaintainedPopulation, PopulationInput}, + share_common_summary_subtrees, SummaryExpr, ValueOperation, + }, + pre_asap::{Column, DataType, QueryExpr, Schema}, + types::AccuracyTarget, +}; +use std::rc::Rc; + +async fn aggregate(q: &str) -> Rc { + let catalog = SqlCatalog::new().with_table( + "samples", + Schema::new(vec![ + Column::new("latency", DataType::Float64, false), + Column::new("job", DataType::Utf8, false), + ]), + ); + let root = lower_sql(q, &catalog, AccuracyTarget::Exact).await.unwrap(); + Rc::new(root) +} + +fn population( + mut node: &asap_types::post_asap::SummaryNode, +) -> ( + &Rc, + &MaintainedPopulation, +) { + while let SummaryExpr::ValueOperation { + child, + operation: ValueOperation::Project { .. }, + .. + } = &node.expr + { + node = child; + } + let SummaryExpr::ValueOperation { child, .. } = &node.expr else { + panic!("readout") + }; + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainPopulation { population }, + .. + } = &child.expr + else { + panic!("state") + }; + (child, population) +} + +// Quantile parameters are readout identity, while source, value column and grouping are state identity. +#[tokio::test] +async fn sql_quantiles_share_rows_without_promql_lookback() { + let roots = vec![ + aggregate("SELECT median(latency) FROM samples").await, + aggregate("SELECT approx_percentile_cont(latency, 0.99) FROM samples").await, + ]; + let rule = MaintainedPopulationStrategy::new(&roots); + let plans = share_common_summary_subtrees( + roots + .iter() + .enumerate() + .map(|(i, r)| (i, rule.candidate(r).expect("table population"))) + .collect(), + ); + for (_, plan) in &plans { + compile_executable_dag(plan).unwrap(); + } + let (a, spec) = population(&plans[0].1); + let (b, _) = population(&plans[1].1); + assert!(Rc::ptr_eq(a, b)); + assert!(matches!( + spec.input, + PopulationInput::Rows { + value_column: 0, + .. + } + )); +} + +// Different GROUP BY populations must not be merged just because they read the same table. +#[tokio::test] +async fn sql_grouping_separates_populations() { + let roots = vec![ + aggregate("SELECT median(latency) FROM samples").await, + aggregate("SELECT job, median(latency) FROM samples GROUP BY job").await, + ]; + let rule = MaintainedPopulationStrategy::new(&roots); + let a = rule.candidate(&roots[0]).unwrap(); + let b = rule.candidate(&roots[1]).unwrap(); + assert_ne!(population(&a).1.input, population(&b).1.input); +} + +// Input predicates and value expressions remain part of sharing identity. +#[tokio::test] +async fn sql_filters_separate_populations() { + let roots = vec![ + aggregate("SELECT median(latency) FROM samples WHERE job = 'api'").await, + aggregate("SELECT median(latency) FROM samples WHERE job = 'db'").await, + ]; + let rule = MaintainedPopulationStrategy::new(&roots); + let a = rule.candidate(&roots[0]).expect("filtered table input"); + let b = rule.candidate(&roots[1]).expect("filtered table input"); + assert_ne!(population(&a).1.input, population(&b).1.input); +} + +// All four scalar readouts can share the same non-null numeric SQL population. +#[tokio::test] +async fn sql_scalar_readouts_share_membership() { + let mut roots = Vec::new(); + for function in [ + "median(latency)", + "sum(latency)", + "avg(latency)", + "count(*)", + ] { + roots.push(aggregate(&format!("SELECT {function} FROM samples")).await); + } + let rule = MaintainedPopulationStrategy::new(&roots); + let plans = share_common_summary_subtrees( + roots + .iter() + .enumerate() + .map(|(i, r)| (i, rule.candidate(r).expect("scalar population"))) + .collect(), + ); + for (_, plan) in &plans { + compile_executable_dag(plan).unwrap(); + assert!(Rc::ptr_eq(population(&plans[0].1).0, population(plan).0)); + } +} + +// A readout cannot reinterpret a label column as its numeric population. +#[tokio::test] +async fn malformed_table_population_fails_validation() { + let root = aggregate("SELECT median(latency) FROM samples").await; + let rule = MaintainedPopulationStrategy::new(std::slice::from_ref(&root)); + let mut candidate = rule.candidate(&root).unwrap(); + let SummaryExpr::ValueOperation { child, .. } = &mut Rc::make_mut(&mut candidate).expr else { + unreachable!() + }; + let SummaryExpr::ValueOperation { child, .. } = &mut Rc::make_mut(child).expr else { + unreachable!() + }; + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainPopulation { population }, + .. + } = &mut Rc::make_mut(child).expr + else { + unreachable!() + }; + let PopulationInput::Rows { value_column, .. } = &mut population.input else { + unreachable!() + }; + *value_column = 1; + assert!(compile_executable_dag(&candidate).is_err()); +} + +// SQL ORDER BY value DESC LIMIT k uses the same maximum-k state contract. +#[tokio::test] +async fn sql_topk_limits_share_maximum_k() { + let roots = vec![ + aggregate("SELECT * FROM samples ORDER BY latency DESC LIMIT 1").await, + aggregate("SELECT * FROM samples ORDER BY latency DESC LIMIT 5").await, + ]; + let rule = MaintainedPopulationStrategy::new(&roots); + let plans = share_common_summary_subtrees( + roots + .iter() + .enumerate() + .map(|(i, r)| (i, rule.candidate(r).expect("SQL topk"))) + .collect(), + ); + for (_, plan) in &plans { + compile_executable_dag(plan).unwrap(); + assert_eq!(population(plan).1.max_k, 5); + assert!(Rc::ptr_eq(population(&plans[0].1).0, population(plan).0)); + } +} diff --git a/crates/integration-tests/tests/exact_composition.rs b/crates/integration-tests/tests/exact_composition.rs index fe32e568..de138d48 100644 --- a/crates/integration-tests/tests/exact_composition.rs +++ b/crates/integration-tests/tests/exact_composition.rs @@ -253,7 +253,7 @@ fn every_exact_accumulator_is_finalized_before_an_outer_sketch() { AggIntent::Min { col: None }, Rc::new(metric_scan(&["zone"])), ), - ExactKind::MinMax, + ExactKind::Min, ), ( agg( @@ -261,7 +261,7 @@ fn every_exact_accumulator_is_finalized_before_an_outer_sketch() { AggIntent::Max { col: None }, Rc::new(metric_scan(&["zone"])), ), - ExactKind::MinMax, + ExactKind::Max, ), ( per_entity( @@ -616,8 +616,8 @@ fn readout_under_maintenance_is_rejected_at_construction() { expr: SummaryExpr::SummaryAgg { child: post, family: SummaryFamilyType::ExactAggregate( - ExactKind::MinMax, - asap_types::post_asap::ExactParams::MinMax, + ExactKind::Max, + asap_types::post_asap::ExactParams::Max, ), input: SummaryUpdate::column(asap_types::pre_asap::ColumnRef::SampleValue), reduction: Reduction::by(vec![]), diff --git a/crates/integration-tests/tests/promql_numeric_regressions.rs b/crates/integration-tests/tests/promql_numeric_regressions.rs new file mode 100644 index 00000000..41ea2396 --- /dev/null +++ b/crates/integration-tests/tests/promql_numeric_regressions.rs @@ -0,0 +1,278 @@ +//! Numeric regression fixtures: actual PromQL lowering plus numeric update/readout checks. +//! The count/sum interpreter below verifies planner update semantics, not a deployed backend. +use asap_aware_mapping::{Replacement, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG}; +use asap_frontend_promql::lower_promql; +use asap_types::post_asap::{ + compile_executable_dag, ExactKind, SummaryExpr, SummaryFamilyType, SummaryInputExpr, + SummaryNode, SummaryUpdate, +}; +use asap_types::pre_asap::{ColumnRef, Reduction}; +use asap_types::types::AccuracyTarget; +use std::rc::Rc; + +fn plan(query: &str, accuracy: AccuracyTarget) -> Rc { + let pre = Rc::new(lower_promql(query, accuracy).unwrap()); + SketchAlgorithmStrategy::default_cost_model() + .replacements(&TargetSubDAG::new(&pre)) + .into_iter() + .find_map(|r| match r.replacement { + Replacement::Summary(n) => Some(n), + _ => None, + }) + .unwrap_or_else(|| asap_aware_mapping::replacement::keep_pre_asap(&pre).unwrap()) +} +fn aggregate(node: &SummaryNode) -> (&SummaryFamilyType, &SummaryUpdate, &Reduction) { + match &node.expr { + SummaryExpr::SummaryAgg { + family, + input, + reduction, + .. + } => (family, input, reduction), + SummaryExpr::SummaryEstimate { summary_input, .. } => aggregate(summary_input), + SummaryExpr::ValueOperation { child, .. } => aggregate(child), + other => panic!("not a maintained accumulator: {other:?}"), + } +} +fn contribution(family: &SummaryFamilyType, update: &SummaryUpdate, value: f64) -> f64 { + if matches!( + family, + SummaryFamilyType::ExactAggregate(ExactKind::Count, _) + ) { + return 1.; + } + match update.weight { + SummaryInputExpr::Constant(v) => v, + SummaryInputExpr::Column(ColumnRef::SampleValue) => value, + ref other => panic!("unexpected update {other:?}"), + } +} + +/// Target count depends on series multiplicity, never distinct values or health. +#[test] +fn count_up_counts_targets_even_when_values_repeat_or_change_sign() { + let node = plan("count(up)", AccuracyTarget::Exact); + let (family, update, _) = aggregate(&node); + assert!(matches!( + family, + SummaryFamilyType::ExactAggregate(ExactKind::Count, _) + )); + for values in [[1., 1., 1.], [1., 1., 0.], [0., 0., 0.], [-1., -1., -1.]] { + assert_eq!( + values + .into_iter() + .map(|v| contribution(family, update, v)) + .sum::(), + 3. + ); + } + compile_executable_dag(&node).unwrap(); +} + +/// Ten samples give count ten, whereas sum retains the signed sample values. +#[test] +fn window_counts_and_sums_distinguish_one_zero_three_and_negative_values() { + for (query, kind, is_count) in [ + ("count_over_time(up[5m])", ExactKind::Count, true), + ("sum_over_time(up[5m])", ExactKind::Sum, false), + ] { + let node = plan(query, AccuracyTarget::Exact); + let (family, update, reduction) = aggregate(&node); + assert!(matches!(family, SummaryFamilyType::ExactAggregate(k, _) if *k == kind)); + assert_eq!(*reduction, Reduction::PerEntity); + for value in [1., 0., 3., -3.] { + let got: f64 = (0..10).map(|_| contribution(family, update, value)).sum(); + assert_eq!(got, if is_count { 10. } else { value * 10. }); + } + compile_executable_dag(&node).unwrap(); + } +} + +/// Exact summary candidates must exist, rather than merely retaining the original query. +#[test] +fn sum_rate_and_increase_have_real_exact_accumulator_nodes() { + for (query, kind) in [ + ("sum by(job)(up)", ExactKind::Sum), + ("rate(requests_total[5m])", ExactKind::Rate), + ("increase(requests_total[5m])", ExactKind::Increase), + ] { + let node = plan(query, AccuracyTarget::Exact); + let (family, _, _) = aggregate(&node); + assert!(matches!(family, SummaryFamilyType::ExactAggregate(k, _) if *k == kind)); + assert!(node.guarantee.as_ref().unwrap().is_exact()); + compile_executable_dag(&node).unwrap(); + } +} + +/// A finite, nonzero approximate denominator does not prove a valid relative quantile bound. +#[test] +fn checked_ratio_must_not_certify_cross_zero_interpolation() { + let node = plan( + "quantile_over_time(0.5, data[5m]) / quantile_over_time(0.9, data[5m])", + AccuracyTarget::Epsilon(0.01), + ); + assert!( + matches!(node.expr, SummaryExpr::KeepPreAsap(_)), + "unproved ratio must retain native execution" + ); + compile_executable_dag(&node).unwrap(); + // Keep the actual signed-sketch counterexample: division guards alone pass + // even though the quantile interpolation does not preserve relative error. + let alpha = (0.01 - 8.0 * f64::EPSILON) / 2.01; + let estimate = |q| { + let mut sketch = asap_sketchlib::DdSketch::new(alpha); + for value in [-1., 1.011] { + sketch.try_update(value).unwrap(); + } + sketch.quantile_interpolated(q).unwrap() + }; + let numerator = estimate(0.5); + let denominator = estimate(0.9); + let got = numerator / denominator; + assert!( + numerator.is_finite() && denominator.is_finite() && denominator != 0. && got.is_normal() + ); + let exact_quantile = |q: f64| -(1. - q) + 1.011 * q; + let want = exact_quantile(0.5) / exact_quantile(0.9); + let error = (got - want).abs() / want.abs(); + assert!( + error > 0.01, + "fixture must expose cancellation beyond the requested budget" + ); +} + +/// A new conditional average rewrite must not invalidate an otherwise usable outer sketch. +#[test] +fn quantile_over_temporal_average_keeps_a_legal_candidate() { + for query in [ + "quantile(0.9, avg_over_time(a[5m]))", + "quantile(0.9, avg_over_time(a[5m]) + avg_over_time(b[5m]))", + ] { + let node = plan(query, AccuracyTarget::Epsilon(0.01)); + assert!( + !matches!(node.expr, SummaryExpr::KeepPreAsap(_)), + "outer sketch candidate must survive: {query}" + ); + let SummaryExpr::SummaryEstimate { summary_input, .. } = &node.expr else { + panic!("outer sketch readout") + }; + let SummaryExpr::SummaryAgg { child, .. } = &summary_input.expr else { + panic!("outer sketch state") + }; + assert!( + matches!(child.expr, SummaryExpr::KeepPreAsap(_)), + "guarded expression must retain native maintenance input" + ); + compile_executable_dag(&node).unwrap(); + } +} + +struct OneKeyTopKEvidence; +impl asap_aware_mapping::accuracy::AccuracyEvidenceProvider for OneKeyTopKEvidence { + fn propagation_stats( + &self, + op: &asap_types::post_asap::CompositionOperator, + _family: &SummaryFamilyType, + _query: Option<&asap_types::post_asap::SketchQuery>, + ) -> asap_aware_mapping::accuracy::PropagationStats { + // Single-key fixture: no excluded keys; bounds cover every value below. + if matches!( + op, + asap_types::post_asap::CompositionOperator::TopKSelection + ) { + asap_aware_mapping::accuracy::PropagationStats { + topk_selected_lower_bound: Some(-1000.), + topk_excluded_upper_bound: Some(-1001.), + topk_interval_failure_probability: Some(0.001), + ..Default::default() + } + } else { + Default::default() + } + } +} + +#[test] +fn sketch_counts_use_unit_weights_and_signed_sums_keep_value_weights() { + use asap_aware_mapping::accuracy::{DefaultAccuracyModel, EqualSplitAllocator}; + use asap_aware_mapping::cost_model::DefaultCostModel; + use asap_types::post_asap::{NonNegativeWeightProof, SketchAlgorithm, WeightDomain}; + let strategy = SketchAlgorithmStrategy::with_models_and_evidence( + &DefaultCostModel, + &DefaultAccuracyModel, + &EqualSplitAllocator, + &OneKeyTopKEvidence, + ); + for is_count in [true, false] { + let query = if is_count { + "topk(1, count_over_time(up[5m]))" + } else { + "topk(1, sum_over_time(up[5m]))" + }; + let pre = Rc::new(lower_promql(query, AccuracyTarget::Epsilon(0.01)).unwrap()); + let candidates = strategy.replacements(&TargetSubDAG::new(&pre)); + let wanted = if is_count { + SketchAlgorithm::CmsWithHeap + } else { + SketchAlgorithm::CountSketchWithHeap + }; + let node = candidates + .iter() + .find_map(|c| { + let Replacement::Summary(node) = &c.replacement else { + return None; + }; + let (family, _, _) = aggregate(node); + matches!(family, SummaryFamilyType::Sketch(kind, _) if kind.algorithm() == &wanted) + .then_some(node) + }) + .expect("weighted sketch candidate"); + let (family, update, _) = aggregate(node); + if is_count { + assert_eq!(update.weight, SummaryInputExpr::Constant(1.)); + assert_eq!( + update.weight_domain, + WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount + } + ); + } else { + assert_eq!( + update.weight, + SummaryInputExpr::Column(ColumnRef::SampleValue) + ); + for c in &candidates { + if let Replacement::Summary(n) = &c.replacement { + assert!( + !matches!(aggregate(n).0, SummaryFamilyType::Sketch(kind, _) if kind.algorithm() == &SketchAlgorithm::CmsWithHeap) + ); + } + } + } + for value in [1., 0., 3., -3.] { + let mut cms = + asap_sketchlib::message_pack_format::portable::countminsketch::new_sketchlib_cms( + 5, 128, + ); + let mut cs = + asap_sketchlib::message_pack_format::portable::countsketch::CountSketch::new( + 5, 128, + ); + for _ in 0..10 { + let weight = contribution(family, update, value); + if is_count { + cms.insert_many(&asap_sketchlib::DataInput::String("a".into()), weight); + } else { + cs.update("a", weight); + } + } + let got = if is_count { + cms.estimate(&asap_sketchlib::DataInput::String("a".into())) + } else { + cs.estimate("a") + }; + assert_eq!(got, if is_count { 10. } else { 10. * value }); + } + compile_executable_dag(node).unwrap(); + } +} diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index 9cb3e00c..30b85e6e 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -415,7 +415,7 @@ fn promql_binary_arithmetic_preserves_both_scalar_operand_orders() { #[test] fn promql_binary_arithmetic_falls_back_as_a_whole_for_unsupported_arm() { - let root = lower_and_realize("rate(a[1m]) + avg_over_time(b[1m])"); + let root = lower_and_realize("rate(a[1m]) + stddev_over_time(b[1m])"); assert!(matches!(root.expr, SummaryExpr::KeepPreAsap(_))); } diff --git a/crates/types/src/post_asap/cse.rs b/crates/types/src/post_asap/cse.rs index 92e29913..a8077f74 100644 --- a/crates/types/src/post_asap/cse.rs +++ b/crates/types/src/post_asap/cse.rs @@ -417,6 +417,8 @@ mod tests { lhs: Rc::clone(¤t), rhs: current, operator: super::super::BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, kind: crate::pre_asap::BinaryOpKind::Arithmetic( crate::pre_asap::ArithmeticOpKind::Add, ), diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs index fba174b0..2dcd15b3 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -153,6 +153,8 @@ impl ExecutionDataStateEdge { /// it expects, and so tests can assert the *reason* a plan was rejected. #[derive(Debug, Clone, PartialEq, Eq, Error)] pub enum ExecutionDataStateError { + #[error("invalid maintained-population maintenance/readout contract")] + InvalidMaintainedPopulation, /// A query-time value (`SummaryEstimate` / read-time `ValueOperation` output) /// placed beneath a maintained summary — the one shape issue #171's /// data_state split exists to make unrepresentable. @@ -196,6 +198,8 @@ pub enum ExecutionDataStateError { MaintenanceRowsAtRoot, #[error("unsupported maintenance binary schema or operator")] InvalidMaintenanceBinary, + #[error("checked division requires one valid guard on a read-time division operator")] + InvalidCheckedDivision, /// An `ExactOperation` whose input columns are not all `Plain` at its /// declared data_state. #[error("exact operator consumes non-plain column {column:?} ({dtype})")] @@ -330,6 +334,18 @@ fn visit( timing, operator, } => { + if (operator.checked_relative_division && operator.checked_finite_division) + || (operator.checked_relative_division || operator.checked_finite_division) + && (*timing != ExecutionTiming::ReadTime + || !matches!( + operator.kind, + crate::pre_asap::BinaryOpKind::Arithmetic( + crate::pre_asap::ArithmeticOpKind::Div + ) + )) + { + return Err(ExecutionDataStateError::InvalidCheckedDivision); + } if *timing == ExecutionTiming::MaintenanceTime { use crate::pre_asap::{BinaryOpKind, DataType}; if operator.vector_match.is_some() @@ -462,6 +478,20 @@ fn visit( operation, timing, } => { + let valid_population = match operation { + ValueOperation::MaintainPopulation { population } => { + *timing == ExecutionTiming::MaintenanceTime + && matches!(&child.expr, SummaryExpr::KeepPreAsap(input) if population.matches_input(input)) + } + ValueOperation::ReadPopulation { readout } => { + *timing == ExecutionTiming::ReadTime + && matches!(&child.expr, SummaryExpr::ValueOperation { operation: ValueOperation::MaintainPopulation { population }, timing: ExecutionTiming::MaintenanceTime, .. } if population.supports(readout)) + } + _ => true, + }; + if !valid_population { + return Err(ExecutionDataStateError::InvalidMaintainedPopulation); + } let required = match timing { ExecutionTiming::MaintenanceTime => ExecutionDataState::MAINTENANCE_ROWS, ExecutionTiming::ReadTime => ExecutionDataState::READ_ROWS, @@ -471,7 +501,17 @@ fn visit( || matches!(operation, ValueOperation::FinalizeExactAccumulator)) && s == ExecutionDataState::MAINTENANCE_SUMMARY && is_exact_accumulator_state(&child.schema).is_ok(); - if s != required && !exact_readout { + let population_readout = matches!(operation, ValueOperation::ReadPopulation { .. }) + && *timing == ExecutionTiming::ReadTime + && matches!( + &child.expr, + SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainPopulation { .. }, + timing: ExecutionTiming::MaintenanceTime, + .. + } + ); + if s != required && !exact_readout && !population_readout { return Err(ExecutionDataStateError::IllegalChildDataState { edge: ExecutionDataStateEdge::ValueOperationChild.describe(), child: s, diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index f0110717..c5c878f9 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -25,6 +25,15 @@ pub enum ExactOperation { #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] #[non_exhaustive] pub enum ValueOperation { + /// Maintain the full declared population, including membership changes, + /// so removing a TopK member can promote another. + MaintainPopulation { + population: super::maintained_population::MaintainedPopulation, + }, + /// Read an aggregate or TopK prefix from the maintained population. + ReadPopulation { + readout: super::maintained_population::PopulationReadout, + }, Exact(ExactOperation), /// Read an exact accumulator's state as its finalized scalar value. /// @@ -249,6 +258,16 @@ pub enum SummaryExpr { /// All semantics owned by a post-ASAP binary operator. #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] pub struct BinaryOperator { + /// Execute division only for finite operands, a nonzero divisor, and a + /// normal finite result; otherwise use exact execution. Required by the + /// relative-value division certificate, including floating-point range. + #[serde(default)] + pub checked_relative_division: bool, + /// Conditional exact rewrites (such as temporal average from sum/count) + /// require finite operands and quotient. Zero/subnormal results are valid; + /// overflow must fall back to the original query rather than emit infinity. + #[serde(default)] + pub checked_finite_division: bool, pub kind: BinaryOpKind, /// `None` is the only currently supported vector/vector matching mode. /// The field is retained so execution never has to recover semantics by diff --git a/crates/types/src/post_asap/guarantee.rs b/crates/types/src/post_asap/guarantee.rs index 25eb4d8e..cc171d1b 100644 --- a/crates/types/src/post_asap/guarantee.rs +++ b/crates/types/src/post_asap/guarantee.rs @@ -212,6 +212,9 @@ impl ProbabilityExpr { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "op", rename_all = "snake_case")] pub enum CompositionOperator { + /// Relative division with runtime finite/nonzero/range checks. For operand + /// bounds a,b the ratio bound is (a+b)/(1-b), with b < 1. + CheckedRelativeDivision, /// An approximate summary built over its inputs' (approximate) values /// — the sketch-over-sketch case. Its own `local` guarantee composes /// with the inputs' under a same-metric rule. diff --git a/crates/types/src/post_asap/maintained_population.rs b/crates/types/src/post_asap/maintained_population.rs new file mode 100644 index 00000000..fc953d53 --- /dev/null +++ b/crates/types/src/post_asap/maintained_population.rs @@ -0,0 +1,145 @@ +//! Language-independent maintained populations and their readouts. +//! Resource limits, ingestion placement and data structures belong to the executor. +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CurrentSeriesInput { + pub metric: String, + pub matchers: Vec, + pub grouping: Vec, + pub without: bool, + pub lookback_ms: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub struct CurrentSeriesMatcher { + pub label: String, + pub value: String, + pub operation: CurrentSeriesMatch, +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub enum CurrentSeriesMatch { + Equal, + NotEqual, + Regex, + NotRegex, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum PopulationReadout { + Quantile { q: f64 }, + TopK { k: usize }, + Sum, + Count, + Average, +} + +impl CurrentSeriesInput { + /// Verify the named contract against the canonical maintenance input. + pub fn matches_input(&self, input: &crate::pre_asap::QueryExpr) -> bool { + use crate::pre_asap::{CompareOpKind, DataType, QueryExpr, ScalarValue, Source}; + let QueryExpr::Scan { + source: Source::TimeSeries { metric }, + predicates, + schema, + } = input + else { + return false; + }; + if self.metric.is_empty() + || *metric != self.metric + || schema.closed + || schema.time_index.is_none() + || self.lookback_ms != 300_000 + { + return false; + } + if self.grouping.iter().any(|label| { + !schema + .columns + .iter() + .any(|c| c.name == *label && c.dtype == DataType::Utf8) + }) { + return false; + } + let mut matchers = Vec::new(); + for predicate in predicates { + let QueryExpr::Compare { left, op, right } = predicate.0.as_ref() else { + return false; + }; + let (QueryExpr::Column(col), QueryExpr::Literal(ScalarValue::Utf8(value))) = + (left.as_ref(), right.as_ref()) + else { + return false; + }; + let Some(column) = schema.columns.get(*col) else { + return false; + }; + if column.dtype != DataType::Utf8 { + return false; + } + let operation = match op { + CompareOpKind::Eq => CurrentSeriesMatch::Equal, + CompareOpKind::Ne => CurrentSeriesMatch::NotEqual, + CompareOpKind::Regex => CurrentSeriesMatch::Regex, + CompareOpKind::NotRegex => CurrentSeriesMatch::NotRegex, + _ => return false, + }; + matchers.push(CurrentSeriesMatcher { + label: column.name.clone(), + value: value.clone(), + operation, + }); + } + matchers.sort(); + matchers.dedup(); + self.matchers == matchers && self.grouping.windows(2).all(|w| w[0] < w[1]) + } +} + +/// Membership is part of state identity. Table rows must never acquire implicit +/// latest-per-series selection, stale markers, or a PromQL lookback. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub enum PopulationInput { + CurrentSeries(CurrentSeriesInput), + Rows { + input: std::rc::Rc, + value_column: usize, + grouping: crate::pre_asap::GroupKeys, + }, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct MaintainedPopulation { + pub input: PopulationInput, + pub max_k: usize, + pub quantiles: bool, +} + +impl MaintainedPopulation { + pub fn matches_input(&self, input: &crate::pre_asap::QueryExpr) -> bool { + match &self.input { + PopulationInput::CurrentSeries(spec) => spec.matches_input(input), + PopulationInput::Rows { + input: expected, + value_column, + grouping, + } => { + use crate::pre_asap::{DataType, QueryExpr, Source}; + expected.as_ref() == input + && matches!(input, QueryExpr::Scan { source: Source::Table { .. }, schema, .. } + if schema.closed && schema.columns.get(*value_column).is_some_and(|c| c.dtype == DataType::Float64 && !c.nullable) + && !grouping.is_without() && grouping.keys().iter().all(|k| *k < schema.columns.len())) + } + } + } + + pub fn supports(&self, readout: &PopulationReadout) -> bool { + match readout { + PopulationReadout::Quantile { q } => self.quantiles && q.is_finite(), + PopulationReadout::TopK { k } => *k <= self.max_k, + PopulationReadout::Sum | PopulationReadout::Count | PopulationReadout::Average => true, + } + } +} diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 52e9f690..5b7ba43b 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -32,6 +32,7 @@ pub mod executable_dag; pub mod execution_data_state; pub mod expr; pub mod guarantee; +pub mod maintained_population; pub mod query_time; pub mod schema; pub mod sketch; diff --git a/crates/types/src/post_asap/schema.rs b/crates/types/src/post_asap/schema.rs index a61abcef..ffef2a05 100644 --- a/crates/types/src/post_asap/schema.rs +++ b/crates/types/src/post_asap/schema.rs @@ -24,7 +24,7 @@ pub enum SummaryFamilyType { /// pre-ASAP `DataType` (`Int64`/`Float64`/`Utf8`/`Bool`/`Timestamp`), /// passed through unchanged from a pre-ASAP edge. Plain(DataType), - /// Exact, mergeable accumulator state (`Sum`/`Count`/`MinMax`/`Rate`/ + /// Exact, mergeable accumulator state (`Sum`/`Count`/`Min`/`Max`/`Rate`/ /// `Increase`) — the partial state *is* the value; no readout needed. ExactAggregate(ExactKind, ExactParams), /// Approximate sketch state (KLL/CMS/HLL/…), read out via a diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs index b6f43826..9d870a6c 100644 --- a/crates/types/src/post_asap/sketch.rs +++ b/crates/types/src/post_asap/sketch.rs @@ -13,8 +13,10 @@ pub enum ExactKind { Sum, /// Exact count accumulator (mergeable by addition). Count, - /// Exact min/max accumulator (mergeable by comparison). - MinMax, + /// Exact minimum accumulator (mergeable by comparison). + Min, + /// Exact maximum accumulator (mergeable by comparison). + Max, /// Exact increase accumulator (counter-reset-aware delta). Increase, /// Rate accumulator (increase / time window duration). @@ -31,7 +33,8 @@ pub enum ExactKind { pub enum ExactParams { Sum, Count, - MinMax, + Min, + Max, Increase, Rate, IRate, diff --git a/docs/design_docs/asap-aware-mapping/README.md b/docs/design_docs/asap-aware-mapping/README.md index ab8a1925..534561e7 100644 --- a/docs/design_docs/asap-aware-mapping/README.md +++ b/docs/design_docs/asap-aware-mapping/README.md @@ -41,6 +41,7 @@ budgets; deployment belongs to a later stage. - **Replacement Sub-DAG**: A candidate post-ASAP sub-DAG to replace a target sub-DAG. For example, a quantile aggregation may have KLL, DDSketch, and exact aggregation as alternatives. - **ReplacementStrategy**: A rule to recognize a target Sub-DAG and produces one or more valid replacement Sub-DAGs. - **Candidate Plan**: A complete post-ASAP plan formed by choosing compatible ReplacementStrategies across the plan. +- **Maintained population**: A multiset of qualifying records retained across evaluations and updated as members enter, change, leave or expire; multiple readouts can share this state. - **Cost Model**: A model used to compare valid candidate plans according to criteria such as storage, update cost, query latency, and accuracy. The distinction between **ReplacementStrategy** and **Candidate Plan** is important. A ReplacementStrategy is a local choice at one decision point, while a candidate plan is a complete plan that combines choices across all relevant decision points. @@ -88,6 +89,8 @@ The design is split into focused documents: combines, checks, costs, and ranks alternatives across a workload. - [Optimizations](optimizations.md) describes summary selection, parameterization, subpopulation and time organization, roll-ups, sharing, semantic rewrites, and hybrid execution. +- [Shared maintained population rule](maintained-populations.md) defines population membership, + SQL/PromQL input contracts, sharing preconditions, the replacement DAG, and deployment obligations. - [Summary properties](summary_properties.md) lists the capabilities used to determine whether summaries and optimizations can be composed safely. - [End-to-end accuracy guarantees](end-to-end-accuracy-guarantees.md) specifies the typed diff --git a/docs/design_docs/asap-aware-mapping/maintained-populations.md b/docs/design_docs/asap-aware-mapping/maintained-populations.md new file mode 100644 index 00000000..7b496ae7 --- /dev/null +++ b/docs/design_docs/asap-aware-mapping/maintained-populations.md @@ -0,0 +1,178 @@ +# Shared maintained population rule + +## Definition and motivation + +A **population** is the multiset of records that an aggregation is defined over, +after applying its source selection, predicates, membership semantics and grouping. +A **maintained population** is that multiset represented by state which is kept +across evaluations and updated when members enter, change, leave or expire. +A **readout** computes a result from the maintained state at an admitted evaluation. + +For group `g` and evaluation `t`, write this multiset as `P_g(t)`. The contract is +that `ReadPopulation(f, t)` returns `f(P_g(t))`; it must not read a partial or stale +population outside the deployment's admitted coverage/freshness contract. The +population describes **which records count**. The physical data structure describes +**how those records are retained and read**. + +For example, suppose two live PromQL series currently have values `7` and `7`. +Their population contains two members: `count(a)` is `2`, not `1`. If the first +series changes to `9`, the population becomes `{9, 7}`, not `{7, 7, 9}`. Its previous +value is replaced. This distinction requires an explicit rule and state contract: +an append-only quantile sketch cannot by itself implement current-series updates. + +The current rule retains an exact population. It does not prescribe a particular +tree, heap or sketch implementation, and it does not imply a deletable DDSketch. + +## Membership semantics + +| Input contract | Members | Membership changes | +| --- | --- | --- | +| `CurrentSeries` | Latest live sample for each matching series at `t`, partitioned by the declared labels | A newer sample replaces that series' member; a stale marker removes it; lookback expiry removes it | +| `Rows` | Every row of the declared table input, preserving duplicate multiplicity and applying its predicates/grouping | Inserts add members, updates replace affected members, deletes remove the corresponding occurrences; a complete snapshot can atomically replace the multiset | + +The canonical PromQL contract uses a five-minute lookback. For Prometheus 3.5, +the valid sample interval is `(t - 5m, t]`: a sample exactly at the lower boundary +is expired. This is a membership requirement, not a configurable sketch window. +SQL table rows do not inherit this lookback, series identity, or stale-marker behavior. +For example, two historical rows belonging to one device still count as two SQL +rows unless the SQL plan explicitly selects the latest row per device. + +`CurrentSeriesInput` carries the metric, label matchers, grouping and lookback. +`Rows` carries the canonical input (including predicates/schema), value-column +index and grouping. Source identity, predicates, membership semantics, value +column and grouping determine whether consumers refer to the same population. + +## Rule: share one population across compatible readouts + +**Implementation:** `MaintainedPopulationStrategy`, an opt-in `ReplacementStrategy` +in [maintained_population.rs](../../../crates/asap-aware-mapping/src/maintained_population.rs). + +**Target sub-DAGs:** + +- A single-measure `Aggregate(Reduce(grouping), input)` with Quantile, Sum, Count + or Average intent and no HAVING clause. +- A descending single-column `Sort(input)` followed by `Limit(k, offset=0)`. +- SQL projections above these targets are preserved in the replacement. + +The supported input is a canonical direct scan with one of the membership +contracts above. Current-series scans must have the canonical open time-series +schema and supported label predicates. Table scans require a closed schema and +a non-null Float64 value column. Arbitrary relational inputs, nullable value +columns and multi-measure aggregates need additional rules. + +**Replacement sub-DAG:** + +```text +KeepPreAsap(input) + -> MaintainPopulation { input, max_k, quantiles } [maintenance] + -> ReadPopulation { Quantile(q1) } [read] + -> ReadPopulation { Quantile(q2) } [read] + -> ReadPopulation { TopK(k1) } [read] + -> ReadPopulation { TopK(k2) } [read] + -> ReadPopulation { Sum | Count | Average } [read] +``` + +The rule examines compatible workload roots, sets `max_k` to the largest requested +k and enables quantile readout if any consumer needs it. It emits a candidate for +each root; canonical summary CSE interns their identical maintenance producers. +The readout rank `q` and requested prefix `k` do not identify different input +populations. The union of readout requirements does affect the shared producer's +configuration, retained memory and cost. + +**Concrete transformation:** + +```promql +quantile(0.5, a) +quantile(0.99, a) +topk(1, a) +topk(5, a) +``` + +These queries can use one `CurrentSeries` producer with `max_k=5` and quantile +readout enabled. The full population remains available: deleting a TopK member +must allow a previously lower-ranked member to be promoted. Retaining only the +largest five values would not preserve that behavior. + +The same rule can represent these SQL consumers using a `Rows` producer: + +```sql +SELECT median(latency) FROM samples; +SELECT approx_percentile_cont(latency, 0.99) FROM samples; +SELECT * FROM samples ORDER BY latency DESC LIMIT 1; +SELECT * FROM samples ORDER BY latency DESC LIMIT 5; +``` + +By contrast, `a{job="api"}` and `a{job="db"}`, different value columns, and +`by(job)` versus `by(region)` identify different populations and are not shared +by this rule. SQL rows and PromQL current-series members never share state merely +because their source names or numeric values happen to agree. + +## Validation, selection and execution responsibilities + +Planner validates the declared input, maintenance/read phases and readout +compatibility. Its intended guarantee is exact membership and exact readout; +a physical implementation still must preserve the language's numeric and empty-input +semantics. In particular, SQL global COUNT over an empty population returns a row +with zero, while PromQL COUNT over an empty vector returns an empty vector. + +The rule proposes a candidate; it does not select it unconditionally. A compiler +must lower the typed DAG only if its executor supports that membership contract. +Installation requires complete cost evidence for population construction, updates, +retention, readouts, retirement and any required raw-data work. Shared state is +not automatically cheaper than independent or native execution. + +The executor owns record identity, input completeness, replacement/retraction, +coverage, freshness, atomic publication and resource limits. Missing coverage, +unsupported semantics or exhausted resources must not produce a partial result +advertised as exact. The backend's current-series implementation rejects evaluations +older than retained state and falls back while coverage is insufficient. + +At the PR #404/#700 implementation boundary, current-series populations are deployable; +SQL `Rows` candidates are representable but require a table-update/deletion executor. +Existing SQL window-summary compilation is separate. The SQL executor work is being +implemented separately; this design does not treat it as already shipped. + +## Relation to sketch rules and other optimizations + +This rule adds an exact maintained-state alternative. It is distinct from choosing +a sketch family or merging temporal panes, and can coexist with those alternatives +in the same workload. Temporal sketch rules continue to emit +`SummaryAgg -> SummaryEstimate` DAGs. + +For example, `distinct_over_time(a[5m])`, `l2_over_time(a[5m])` and +`entropy_over_time(a[5m])` can read one UnivMon frequency summary when input, +partitioning, window and sketch parameters match. Here L2 is +`sqrt(sum_v count(v)^2)`, and entropy is computed from the same value frequencies. +Each readout still needs its own accuracy evidence: sharing an entropy certificate +does not establish a cardinality or L2 bound. This is the same separation of +population/state identity from readout identity, implemented by the existing sketch +rules rather than by converting UnivMon into an exact `MaintainPopulation` node. + +Quantile division requires evidence that the individual readouts satisfy their +relative-error bounds. Finite operands, a nonzero divisor and a normal quotient +alone do not provide that evidence: interpolation between negative and positive +samples can cancel. The sketch strategy therefore retains native division when +an operand is approximate and no input-domain proof is available. + +The sum/count rewrite of temporal average uses a read-time finite-division guard. +When an outer maintained sketch consumes that average, the planner retains the +native average expression as its maintenance input. It cannot move a read-time +fallback guard into the update path, where a failed update could already have +contaminated the outer state. This also applies to arithmetic containing a guarded +average; the outer sketch candidate remains available. + +## Acceptance evidence + +- PromQL quantiles, TopK limits and scalar readouts share only compatible populations. +- SQL frontend tests cover shared quantiles/scalar readouts/maximum k, separation + by grouping and filters, preservation of projections, and invalid value-column rejection. +- Backend admission rejects a table-row producer when only a current-series executor + is available. +- Process tests compare current-series replacements and expiry with Prometheus 3.5. +- The UnivMon process test installs one compatible materialization for all three + readouts and checks that missing entropy evidence does not disable the L2 path. + +These tests establish the covered semantic and sharing behavior, not measured +end-to-end speedups or universal floating-point equivalence. The review regressions +for the exact lookback boundary and temporal-average overflow are separate checks; +passing the ordinary workload examples alone does not establish those edge cases. diff --git a/docs/design_docs/asap-aware-mapping/optimizations.md b/docs/design_docs/asap-aware-mapping/optimizations.md index 4491ed8c..51c217ff 100644 --- a/docs/design_docs/asap-aware-mapping/optimizations.md +++ b/docs/design_docs/asap-aware-mapping/optimizations.md @@ -14,6 +14,24 @@ ASAP-aware mapping should support several largely orthogonal dimensions of optim Some of these are described below with examples. +## Shared maintained population rule + +A maintained population is the multiset of qualifying input records represented +by state retained across query evaluations. Membership updates and aggregate +readouts are separate operations. This lets different quantiles, TopK limits and +scalar aggregates share one producer when their input semantics agree. + +`MaintainedPopulationStrategy` recognizes supported Aggregate or Sort/Limit +sub-DAGs and emits `MaintainPopulation -> ReadPopulation` candidates. For example, +`quantile(0.5, a)`, `quantile(0.99, a)`, `topk(1, a)` and `topk(5, a)` can share one +current-series population and a maximum-k cache of five. SQL table-row consumers +use the same rule with a different membership contract; they do not become +latest-series queries. + +See [the rule specification](maintained-populations.md) for the definition, +matching conditions, sharing identity, exactness requirements, SQL examples, +compiler capability checks and relation to UnivMon/sketch readouts. + ## Using a subpopulation sketch Queries often compute the same statistic over many subpopulations: diff --git a/docs/design_docs/asap-aware-mapping/physical-plan-integration.md b/docs/design_docs/asap-aware-mapping/physical-plan-integration.md index 330c8d4b..827336e8 100644 --- a/docs/design_docs/asap-aware-mapping/physical-plan-integration.md +++ b/docs/design_docs/asap-aware-mapping/physical-plan-integration.md @@ -404,3 +404,19 @@ replace those failures with zero cost or structural node counting. See [Analytical resource cost](analytical-resource-cost.md) for the resource formulas, evidence validation, comparison-scope rules, and calibration model. + +## Conditional temporal-average lowering + +`avg_over_time(a[5m])` can expose independently maintained sum and count +components, but their division is conditional. Two finite samples of `1e308` +have a finite average even though their sum overflows. Planner therefore emits +a read-time `BinaryOperator` with `checked_finite_division=true` and never exports +this temporal transformation as an unconditional pre-ASAP rewrite. + +The backend lowers the guard to `FiniteDiv`: operands and quotient must be finite, +and the divisor must be nonzero. Failure executes the original average query. +Zero and subnormal averages remain valid accelerated results. This guard is +distinct from `checked_relative_division`, whose relative-error certificate also +requires a normal result; setting both guards or attaching a guard to a non-division +operator is invalid. Compilers must preserve this typed condition rather than +recovering average semantics from query text.