From ea721e889b79e9ca22741a4d0370e9929bcf5b89 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 13:43:02 -0600 Subject: [PATCH 01/12] Add Planner rules for shared current-series populations --- Cargo.lock | 1 + crates/asap-aware-mapping/Cargo.toml | 3 + .../asap-aware-mapping/src/current_series.rs | 352 ++++++++++++++++++ crates/asap-aware-mapping/src/lib.rs | 2 + crates/types/src/post_asap/current_series.rs | 104 ++++++ .../src/post_asap/execution_data_state.rs | 28 +- crates/types/src/post_asap/expr.rs | 9 + crates/types/src/post_asap/mod.rs | 1 + .../current-series-populations.md | 32 ++ 9 files changed, 531 insertions(+), 1 deletion(-) create mode 100644 crates/asap-aware-mapping/src/current_series.rs create mode 100644 crates/types/src/post_asap/current_series.rs create mode 100644 docs/design_docs/asap-aware-mapping/current-series-populations.md diff --git a/Cargo.lock b/Cargo.lock index c008c0f5..d9e299c2 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", "serde", "serde_json", diff --git a/crates/asap-aware-mapping/Cargo.toml b/crates/asap-aware-mapping/Cargo.toml index 352ed1cf..7f158edc 100644 --- a/crates/asap-aware-mapping/Cargo.toml +++ b/crates/asap-aware-mapping/Cargo.toml @@ -12,3 +12,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/current_series.rs b/crates/asap-aware-mapping/src/current_series.rs new file mode 100644 index 00000000..f85b21cb --- /dev/null +++ b/crates/asap-aware-mapping/src/current_series.rs @@ -0,0 +1,352 @@ +//! Exact current-series population candidates over canonical PromQL IR. +use crate::replacement::{ + Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG, +}; +use asap_types::post_asap::{ + current_series::*, 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 recognize( + root: &QueryExpr, +) -> Option<(CurrentSeriesPopulation, CurrentSeriesReadout, Rc)> { + let (source, grouping, readout) = match root { + QueryExpr::Aggregate { + child, + reduction: Reduction::Reduce(grouping), + measures, + having: None, + .. + } => { + let [AggIntent::Quantile { q, col, .. }] = measures.as_slice() else { + return None; + }; + if !q.is_finite() { + return None; + } + let schema = child.output_schema().ok()?; + if col.is_some_and(|c| schema.columns.get(c).is_none_or(|c| c.name != "value")) { + return None; + } + (child, grouping, CurrentSeriesReadout::Quantile { q: *q }) + } + 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 || child.output_schema().ok()?.columns.get(*col)?.name != "value" { + return None; + } + (child, partition_by, CurrentSeriesReadout::TopK { k: *n }) + } + _ => return None, + }; + let QueryExpr::Scan { + source: Source::TimeSeries { metric }, + predicates, + schema, + } = source.as_ref() + else { + 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(( + CurrentSeriesPopulation { + 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 +/// current-series inputs and price the maintenance/readout boundary. +/// The population is exact; max_k bounds the shared readout cache, not its members. +pub struct CurrentSeriesStrategy { + roots: Vec>, +} +impl CurrentSeriesStrategy { + pub fn new(roots: &[Rc]) -> Self { + Self { + roots: roots.to_vec(), + } + } + pub fn candidate(&self, root: &Rc) -> Option> { + 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 { + CurrentSeriesReadout::Quantile { .. } => population.quantiles = true, + CurrentSeriesReadout::TopK { k } => { + population.max_k = population.max_k.max(k) + } + } + } + } + } + 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::MaintainCurrentSeries { population }, + timing: ExecutionTiming::MaintenanceTime, + }, + schema: input_schema, + guarantee: Some(ResultGuarantee::exact( + "latest value per series with stale retraction and lookback expiry", + )), + }); + Some(Rc::new(SummaryNode { + expr: SummaryExpr::ValueOperation { + child: maintained, + operation: ValueOperation::ReadCurrentSeries { readout }, + timing: ExecutionTiming::ReadTime, + }, + schema: plain(root.output_schema().ok()?), + guarantee: Some(ResultGuarantee::exact("exact current-population readout")), + })) + } +} +impl ReplacementStrategy for CurrentSeriesStrategy { + 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: "CurrentSeriesStrategy", replacement: Replacement::Summary(node), provenance: ReplacementProvenance::SummaryImplementation, rationale: "share an exact retractable current-series population across quantiles and TopK limits".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(), + ) + } + + // 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 = CurrentSeriesStrategy::new(&roots); + let space = crate::search_workload_with( + roots + .iter() + .enumerate() + .map(|(i, r)| (i, Rc::clone(r))) + .collect(), + &[Box::new(CurrentSeriesStrategy::new(&roots))], + ); + assert!(space + .groups() + .flat_map(|g| &g.candidates) + .any(|c| c.strategy == "CurrentSeriesStrategy")); + 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::ReadCurrentSeries { .. }, + .. + } = &plan.expr + else { + panic!("missing typed readout") + }; + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainCurrentSeries { 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 = CurrentSeriesStrategy::new(&roots); + let (p, _, _) = recognize(&roots[0]).unwrap(); + assert!(p.grouping.is_empty()); + let candidate = strategy.candidate(&roots[0]).unwrap(); + let SummaryExpr::ValueOperation { child, .. } = &candidate.expr else { + unreachable!() + }; + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainCurrentSeries { 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(); + 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 = CurrentSeriesStrategy::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::ReadCurrentSeries { + readout: CurrentSeriesReadout::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::ReadCurrentSeries { + readout: CurrentSeriesReadout::TopK { k: 5 }, + }; + let producer = Rc::make_mut(child); + let SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainCurrentSeries { population }, + .. + } = &mut producer.expr + else { + unreachable!() + }; + population.metric = "b".into(); + assert!(compile_executable_dag(&Rc::new(bad)).is_err()); + } +} diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index da96958a..7f03f18d 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 current_series; diff --git a/crates/types/src/post_asap/current_series.rs b/crates/types/src/post_asap/current_series.rs new file mode 100644 index 00000000..2df25ebe --- /dev/null +++ b/crates/types/src/post_asap/current_series.rs @@ -0,0 +1,104 @@ +//! Semantic contract for a retractable population of current PromQL series values. +//! Resource limits, ingestion placement and data structures belong to the executor. +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CurrentSeriesPopulation { + pub metric: String, + pub matchers: Vec, + pub grouping: Vec, + pub without: bool, + pub lookback_ms: u64, + pub max_k: usize, + pub quantiles: bool, +} + +#[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 CurrentSeriesReadout { + Quantile { q: f64 }, + TopK { k: usize }, +} + +impl CurrentSeriesPopulation { + /// 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]) + } + pub fn supports(&self, readout: &CurrentSeriesReadout) -> bool { + match readout { + CurrentSeriesReadout::Quantile { q } => self.quantiles && q.is_finite(), + CurrentSeriesReadout::TopK { k } => *k <= self.max_k, + } + } +} diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs index fba174b0..53c8112e 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 current-series maintenance/readout contract")] + InvalidCurrentSeries, /// 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. @@ -462,6 +464,20 @@ fn visit( operation, timing, } => { + let valid_current = match operation { + ValueOperation::MaintainCurrentSeries { population } => { + *timing == ExecutionTiming::MaintenanceTime + && matches!(&child.expr, SummaryExpr::KeepPreAsap(input) if population.matches_input(input)) + } + ValueOperation::ReadCurrentSeries { readout } => { + *timing == ExecutionTiming::ReadTime + && matches!(&child.expr, SummaryExpr::ValueOperation { operation: ValueOperation::MaintainCurrentSeries { population }, timing: ExecutionTiming::MaintenanceTime, .. } if population.supports(readout)) + } + _ => true, + }; + if !valid_current { + return Err(ExecutionDataStateError::InvalidCurrentSeries); + } let required = match timing { ExecutionTiming::MaintenanceTime => ExecutionDataState::MAINTENANCE_ROWS, ExecutionTiming::ReadTime => ExecutionDataState::READ_ROWS, @@ -471,7 +487,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 current_readout = matches!(operation, ValueOperation::ReadCurrentSeries { .. }) + && *timing == ExecutionTiming::ReadTime + && matches!( + &child.expr, + SummaryExpr::ValueOperation { + operation: ValueOperation::MaintainCurrentSeries { .. }, + timing: ExecutionTiming::MaintenanceTime, + .. + } + ); + if s != required && !exact_readout && !current_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..aa54a0ac 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 { + /// Replace each series' current value, retract stale/expired values, and + /// retain the full population so removing a TopK member can promote another. + MaintainCurrentSeries { + population: super::current_series::CurrentSeriesPopulation, + }, + /// Read one quantile or TopK prefix from the maintained current population. + ReadCurrentSeries { + readout: super::current_series::CurrentSeriesReadout, + }, Exact(ExactOperation), /// Read an exact accumulator's state as its finalized scalar value. /// diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 52e9f690..f6fea4f2 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -28,6 +28,7 @@ //! — see `asap_aware_mapping::grouping`'s module docs for why. pub mod cse; +pub mod current_series; pub mod executable_dag; pub mod execution_data_state; pub mod expr; diff --git a/docs/design_docs/asap-aware-mapping/current-series-populations.md b/docs/design_docs/asap-aware-mapping/current-series-populations.md new file mode 100644 index 00000000..16b7e7a1 --- /dev/null +++ b/docs/design_docs/asap-aware-mapping/current-series-populations.md @@ -0,0 +1,32 @@ +# Current-series population candidates + +`CurrentSeriesStrategy` is an opt-in `ReplacementStrategy` for deployments that +can continuously maintain complete PromQL current-series inputs. It matches +canonical cross-series quantile aggregates and descending value Sort/Limit over +a direct open time-series Scan. Temporal ranges, shifted selectors, nested inputs +and bottom-k do not match. + +The selected post-ASAP DAG is: + +```text +KeepPreAsap(Scan) [maintenance rows] + -> ValueOperation::MaintainCurrentSeries(population) [maintenance rows] + -> ValueOperation::ReadCurrentSeries(quantile q | top-k k) [read rows] +``` + +The population owns source, label predicates, grouping and five-minute selector +lookback semantics. Updates replace each series' latest value; stale markers and +expiry retract it. It retains the full population, not only the largest k values. +Compatible workload consumers use the largest requested k and one quantile +population; canonical CSE shares their maintenance producer. Different sources, +matchers or groups do not share. The readout remains exact. + +IR validation checks the maintenance input against the population contract, +phase placement and the readout's compatibility with its producer. The only new +maintenance-row/read-row bridge is this explicit producer/readout pair. + +The strategy does not parse query text, choose a deployment, assign byte budgets +or claim a performance benefit. Backends lower these typed operations into their +state implementation, bind resource/coverage limits and price build, update, +residency, retirement and readout costs. Other deployments must keep it disabled +until they support that contract. Existing default strategy selection is unchanged. From 15e6cf8ff7763ad56eb1589a0d26476a00526c79 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 15:12:02 -0600 Subject: [PATCH 02/12] Add typed minimum and current-series aggregate candidates --- .../asap-aware-mapping/src/current_series.rs | 41 +++++++-- .../asap-aware-mapping/src/function_rules.rs | 23 ++++- crates/asap-aware-mapping/src/replacement.rs | 2 +- crates/asap-aware-mapping/src/rewrite.rs | 89 ++++++++++++++++--- crates/frontend-promql/src/promql.rs | 30 ++++--- .../tests/promql_conformance.rs | 24 +++-- .../frontend-promql/tests/promql_lowering.rs | 12 +-- crates/types/src/post_asap/current_series.rs | 6 ++ crates/types/src/post_asap/sketch.rs | 4 + 9 files changed, 176 insertions(+), 55 deletions(-) diff --git a/crates/asap-aware-mapping/src/current_series.rs b/crates/asap-aware-mapping/src/current_series.rs index f85b21cb..157859b1 100644 --- a/crates/asap-aware-mapping/src/current_series.rs +++ b/crates/asap-aware-mapping/src/current_series.rs @@ -37,17 +37,23 @@ fn recognize( having: None, .. } => { - let [AggIntent::Quantile { q, col, .. }] = measures.as_slice() else { + let [intent] = measures.as_slice() else { return None; }; - if !q.is_finite() { - return None; - } + let (col, readout) = match intent { + AggIntent::Quantile { q, col, .. } if q.is_finite() => { + (*col, CurrentSeriesReadout::Quantile { q: *q }) + } + AggIntent::Sum { col } => (*col, CurrentSeriesReadout::Sum), + AggIntent::Count { .. } => (None, CurrentSeriesReadout::Count), + AggIntent::Avg { col } => (*col, CurrentSeriesReadout::Average), + _ => return None, + }; let schema = child.output_schema().ok()?; if col.is_some_and(|c| schema.columns.get(c).is_none_or(|c| c.name != "value")) { return None; } - (child, grouping, CurrentSeriesReadout::Quantile { q: *q }) + (child, grouping, readout) } QueryExpr::Limit { n, @@ -162,6 +168,9 @@ impl CurrentSeriesStrategy { CurrentSeriesReadout::TopK { k } => { population.max_k = population.max_k.max(k) } + CurrentSeriesReadout::Sum + | CurrentSeriesReadout::Count + | CurrentSeriesReadout::Average => {} } } } @@ -214,6 +223,28 @@ mod tests { ) } + // 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 = CurrentSeriesStrategy::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() { diff --git a/crates/asap-aware-mapping/src/function_rules.rs b/crates/asap-aware-mapping/src/function_rules.rs index 82ad4e20..e5c5f7ba 100644 --- a/crates/asap-aware-mapping/src/function_rules.rs +++ b/crates/asap-aware-mapping/src/function_rules.rs @@ -15,7 +15,11 @@ 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::Min, ExactParams::Min)), + ), + AggIntent::Max { .. } => ( CompositionOperator::ExactExtremum, Some((ExactKind::MinMax, ExactParams::MinMax)), ), @@ -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/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index fccc188d..50f2423a 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -5626,7 +5626,7 @@ mod tests { ), // exact mergeable accumulators (A::Sum { col: None }, Acc(E::Sum)), - (A::Min { col: None }, Acc(E::MinMax)), + (A::Min { col: None }, Acc(E::Min)), (A::Max { col: None }, Acc(E::MinMax)), (A::Rate, Acc(E::Rate)), (A::IRate, Acc(E::IRate)), diff --git a/crates/asap-aware-mapping/src/rewrite.rs b/crates/asap-aware-mapping/src/rewrite.rs index 459f2e8a..3d8eebcd 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 averages use two independent per-entity accumulators and direct +//! division, preserving open series labels and the single-measure invariant. //! -//! - **`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,7 +127,54 @@ 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). +fn temporal_average_rewrite(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> { + if let Some(rewritten) = temporal_average_rewrite(root) { + return Some(rewritten); + } let (group_count, col) = avg_rewrite_target(root)?; let QueryExpr::Aggregate { reduction, @@ -316,6 +355,7 @@ pub use SemanticEquivalentRewriteStrategy as AvgToSumOverCountStrategy; impl ReplacementStrategy for SemanticEquivalentRewriteStrategy { fn matches(&self, target: &TargetSubDAG<'_>) -> bool { avg_rewrite_target(target.root).is_some() + || temporal_average_rewrite(target.root).is_some() || composed_aggregate_rewrite(target.root).is_some() } @@ -378,6 +418,31 @@ mod tests { } } + // Temporal averages expose two single-measure children without closing labels. + #[test] + fn temporal_average_rewrite_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(), + ); + let rewrites = SemanticEquivalentRewriteStrategy.replacements(&TargetSubDAG::new(&root)); + let rewritten = rewrites + .iter() + .find_map(|r| match &r.replacement { + Replacement::Rewrite(q) => Some(q), + _ => None, + }) + .expect("temporal average rewrite"); + assert_eq!( + root.output_schema().unwrap(), + rewritten.output_schema().unwrap() + ); + assert!(matches!(rewritten.as_ref(), QueryExpr::BinaryOp { .. })); + } + // ── matches ────────────────────────────────────────────────────────── #[test] diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index 6310f129..4a5861fa 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -45,7 +45,7 @@ //! | `info(v, [selector])` | `PromqlInfoEnrich{selector}` — label-enrichment join against the info metric(s); join keys resolved during post-ASAP binding (issue #84) | //! | `group` / `offset` / `@` / `info` | **rejected** — distinct semantics with no intent-algebra representation yet (`info` label-join → #84) | //! | `OUTER by (dims) (…)` | `Aggregate.reduction = Reduce(by = dims)` (generic `topk by`/`bottomk` grouping → `Sort.partition_by`) | -//! | `count by (d) (…)` | `Aggregate{[Cardinality], …}` | +//! | `count by (d) (…)` | `Aggregate{[Count], …}` | //! | `group(v)` / `count_values("l", v)` | `Aggregate{[Group]}` (constant 1) / `Aggregate{[CountValues{l}]}` (group-by-value + count, new label `l`) — issue #49 | //! | `limitk(k, v)` / `limit_ratio(r, v)` | `PromqlSeriesSample{LimitK(k) \| LimitRatio(r)}` — series-sampling selection, whole series kept unchanged (issue #86) | //! | `topk(k, count_over_time(…))` / `topk(k, sum_over_time(…))` | `Aggregate{[TopK{k}]}` (heavy-hitter intent) over the explicit inner `Aggregate{[Count/Sum]}` | @@ -592,8 +592,7 @@ fn build_over_subtree(outer: Outer, keys: Vec, child: Unresolved) -> Outer::Plain(intent) => outer_aggregate(keys, outer_intent(&intent), child), Outer::Count => outer_aggregate( keys, - AggIntent::Cardinality { - col: None, + AggIntent::Count { accuracy: current_accuracy(), }, child, @@ -1448,11 +1447,23 @@ fn build(inner: Inner, keys: Vec, outer: Outer) -> Result } }), Outer::Count => Ok(match &inner.func { - None => windowed_aggregate(inner, keys, cardinality()), + None => windowed_aggregate( + inner, + keys, + AggIntent::Count { + accuracy: current_accuracy(), + }, + ), Some(f) => { let inner_i = inner_intent(f); let inner_agg = windowed_aggregate(inner, vec![], inner_i); - outer_aggregate(keys, cardinality(), inner_agg) + outer_aggregate( + keys, + AggIntent::Count { + accuracy: current_accuracy(), + }, + inner_agg, + ) } }), Outer::CountValues { label } => Ok(match &inner.func { @@ -1648,15 +1659,6 @@ fn filtered_source(metric: String, matchers: Vec, shift: TimeShift) } } -/// `count(v)` / `count by (…) (v)` — SQL `COUNT(DISTINCT col)`'s PromQL -/// counterpart, over the (always implicit) sample value. -fn cardinality() -> AggIntent { - AggIntent::Cardinality { - col: None, - accuracy: current_accuracy(), - } -} - fn inner_intent(f: &InnerFunc) -> AggIntent { match f { InnerFunc::FrequencyL2 => AggIntent::FrequencyL2 { diff --git a/crates/frontend-promql/tests/promql_conformance.rs b/crates/frontend-promql/tests/promql_conformance.rs index 1d6830f5..d346b077 100644 --- a/crates/frontend-promql/tests/promql_conformance.rs +++ b/crates/frontend-promql/tests/promql_conformance.rs @@ -328,10 +328,10 @@ fn sum_by_groups_via_positional_aggregate() { } #[test] -fn count_is_cardinality() { +fn count_counts_series() { assert!(has(&ok("count(up)"), |i| matches!( i, - AggIntent::Cardinality { .. } + AggIntent::Count { .. } ))); } @@ -718,22 +718,21 @@ fn double_unary_negation_nests() { } #[test] -fn count_maps_to_cardinality_and_inherits_accuracy() { - // SEMANTICS (review #2): PromQL `count by (...)` counts distinct series → the - // `Cardinality` intent. The workload's AccuracyTarget threads onto it: +fn count_maps_to_count_and_inherits_accuracy() { + // PromQL counts vector elements, including series with identical values. + // The workload accuracy target is preserved on the Count intent: // `Exact` stays exact (no silent HLL substitution); an approximate target is // carried through for post-ASAP binding to honor. This pins the - // intentional count→Cardinality mapping and its accuracy gating. + // count mapping and its accuracy gating. let exact = lower_promql("count by (job) (up)", AccuracyTarget::Exact).unwrap(); assert!( has(&exact, |i| matches!( i, - AggIntent::Cardinality { - col: None, + AggIntent::Count { accuracy: AccuracyTarget::Exact } )), - "count→Cardinality must stay Exact under AccuracyTarget::Exact, got {:?}", + "count→Count must stay Exact under AccuracyTarget::Exact, got {:?}", intents(&exact) ); @@ -741,12 +740,11 @@ fn count_maps_to_cardinality_and_inherits_accuracy() { assert!( has(&approx, |i| matches!( i, - AggIntent::Cardinality { - col: None, + AggIntent::Count { accuracy: AccuracyTarget::Epsilon(e) } if (*e - 0.01).abs() < 1e-9 )), - "count→Cardinality must carry the approximate target, got {:?}", + "count→Count must carry the approximate target, got {:?}", intents(&approx) ); } @@ -2072,7 +2070,7 @@ fn limitk_by_carries_the_grouping_and_composes_in_a_set_op() { // the PromqlSeriesSample must be preserved under the set op (it must lower, not reject). assert!(has( &ok("count(limitk(2, http_requests) and http_requests)"), - |i| matches!(i, AggIntent::Cardinality { .. }) + |i| matches!(i, AggIntent::Count { .. }) )); } diff --git a/crates/frontend-promql/tests/promql_lowering.rs b/crates/frontend-promql/tests/promql_lowering.rs index 5cfa32cc..128783f9 100644 --- a/crates/frontend-promql/tests/promql_lowering.rs +++ b/crates/frontend-promql/tests/promql_lowering.rs @@ -373,10 +373,7 @@ fn count_over_rate_keeps_both_levels() { else { panic!("expected outer Aggregate{{Cardinality}}, got {qe:?}"); }; - assert!(matches!( - measures.as_slice(), - [AggIntent::Cardinality { .. }] - )); + assert!(matches!(measures.as_slice(), [AggIntent::Count { .. }])); assert!(matches!( child.as_ref(), QueryExpr::Aggregate { measures, .. } if matches!(measures.as_slice(), [AggIntent::Rate]) @@ -399,7 +396,7 @@ fn count_over_time_is_count_intent() { } #[test] -fn outer_count_is_cardinality() { +fn outer_count_counts_vector_elements() { // `count by (symbol) (count_over_time(...))`: inner per-series sample count // over the window (label-preserving), outer cross-series cardinality grouped // on a positional `Aggregate.by`. Leaf = [ts, value, symbol] → symbol = col 2. @@ -414,10 +411,7 @@ fn outer_count_is_cardinality() { panic!("expected outer Aggregate grouped by symbol, got {qe:?}"); }; assert_eq!(reduction, &Reduction::by(vec![2])); - assert!(matches!( - measures.as_slice(), - [AggIntent::Cardinality { .. }] - )); + assert!(matches!(measures.as_slice(), [AggIntent::Count { .. }])); // Inner: Aggregate{Count} over TimeRange (per-series count_over_time). let QueryExpr::Aggregate { measures, child, .. diff --git a/crates/types/src/post_asap/current_series.rs b/crates/types/src/post_asap/current_series.rs index 2df25ebe..6820d36f 100644 --- a/crates/types/src/post_asap/current_series.rs +++ b/crates/types/src/post_asap/current_series.rs @@ -32,6 +32,9 @@ pub enum CurrentSeriesMatch { pub enum CurrentSeriesReadout { Quantile { q: f64 }, TopK { k: usize }, + Sum, + Count, + Average, } impl CurrentSeriesPopulation { @@ -99,6 +102,9 @@ impl CurrentSeriesPopulation { match readout { CurrentSeriesReadout::Quantile { q } => self.quantiles && q.is_finite(), CurrentSeriesReadout::TopK { k } => *k <= self.max_k, + CurrentSeriesReadout::Sum + | CurrentSeriesReadout::Count + | CurrentSeriesReadout::Average => true, } } } diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs index b6f43826..6a14e126 100644 --- a/crates/types/src/post_asap/sketch.rs +++ b/crates/types/src/post_asap/sketch.rs @@ -15,6 +15,8 @@ pub enum ExactKind { Count, /// Exact min/max accumulator (mergeable by comparison). MinMax, + /// Exact minimum, distinct from the legacy maximum accumulator. + Min, /// Exact increase accumulator (counter-reset-aware delta). Increase, /// Rate accumulator (increase / time window duration). @@ -32,6 +34,8 @@ pub enum ExactParams { Sum, Count, MinMax, + /// Exact minimum, distinct from the legacy maximum accumulator. + Min, Increase, Rate, IRate, From e5e81b85da02c25f14710185e8feba4f55c5d94f Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 15:21:03 -0600 Subject: [PATCH 03/12] Size guarded quantile division candidates for expression accuracy --- crates/asap-aware-mapping/src/accuracy.rs | 56 +++++++ crates/asap-aware-mapping/src/replacement.rs | 141 ++++++++++++++++++ crates/asap-aware-mapping/src/rewrite.rs | 2 +- .../src/summary_maintenance_cost/model.rs | 1 + crates/types/src/post_asap/cse.rs | 1 + .../src/post_asap/execution_data_state.rs | 13 ++ crates/types/src/post_asap/expr.rs | 5 + crates/types/src/post_asap/guarantee.rs | 3 + 8 files changed, 221 insertions(+), 1 deletion(-) diff --git a/crates/asap-aware-mapping/src/accuracy.rs b/crates/asap-aware-mapping/src/accuracy.rs index 45a03e44..a132eb39 100644 --- a/crates/asap-aware-mapping/src/accuracy.rs +++ b/crates/asap-aware-mapping/src/accuracy.rs @@ -716,6 +716,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()) @@ -954,6 +991,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/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 50f2423a..efdab7d4 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -1668,6 +1668,9 @@ pub(crate) fn realize_child_with( models: Models<'_>, end_to_end_target: Option<&AccuracyTarget>, ) -> Result, ImplementError> { + if let Some(rewritten) = crate::rewrite::temporal_average_rewrite(root) { + return realize_child_with(&rewritten, models, end_to_end_target); + } if let Some(composed) = realize_binary(root, models, end_to_end_target)? { return Ok(composed); } @@ -1704,6 +1707,118 @@ pub(crate) fn realize_child_with( } } +// The logical rule sizes DDSketch operands against the final expression budget. +// It never converts a rank certificate into a value certificate. +fn relative_division_candidate( + root: &Rc, + models: Models<'_>, + target: Option<&AccuracyTarget>, +) -> Result>, ImplementError> { + let QueryExpr::BinaryOp { + op: BinaryOpKind::Arithmetic(asap_types::pre_asap::ArithmeticOpKind::Div), + lhs, + rhs, + vector_match: None, + } = root.as_ref() + else { + return Ok(None); + }; + let target = target + .or_else(|| bindable_intent(lhs).and_then(accuracy_target)) + .or_else(|| bindable_intent(rhs).and_then(accuracy_target)); + let Some(target) = target else { + return Ok(None); + }; + let epsilon = match target { + AccuracyTarget::Exact => return Ok(None), + AccuracyTarget::Epsilon(e) => *e, + AccuracyTarget::EpsilonDelta { epsilon, .. } => *epsilon, + }; + if !epsilon.is_finite() || epsilon <= 1e-12 { + return Ok(None); + } + let alpha = (epsilon - 8.0 * f64::EPSILON) / (2.0 + epsilon); + let local_target = AccuracyTarget::Epsilon(alpha); + let operand = |expr: &Rc, + layer: usize| + -> Result>, ImplementError> { + if let Some(intent @ AggIntent::Quantile { .. }) = bindable_intent(expr) { + if matches!(accuracy_target(intent), Some(AccuracyTarget::Exact)) { + return Ok(None); + } + let intent = override_accuracy(intent, &local_target); + for implementation in implementations_for_with(&intent, models.cost) { + if !matches!(&implementation, Implementation::Sketch(kind) if kind.algorithm() == &SketchAlgorithm::DDSketch) + { + continue; + } + if let Ok(node) = construct_summary_with( + expr, + &intent, + implementation, + models, + None, + Some(GuaranteeSource::BudgetAllocation { + allocator: "RelativeDivisionAllocator".into(), + layer, + layer_count: 2, + local_target: local_target.clone(), + end_to_end_target: target.clone(), + }), + ) { + return Ok(Some(node)); + } + } + return Ok(None); + } + let node = realize_child_with(expr, models, None)?; + Ok(node + .guarantee + .as_ref() + .is_some_and(ResultGuarantee::is_exact) + .then_some(node)) + }; + let (Some(left), Some(right)) = (operand(lhs, 0)?, operand(rhs, 1)?) else { + return Ok(None); + }; + let left = finalize_exact_accumulator(left, lhs)?; + let right = finalize_exact_accumulator(right, rhs)?; + let Some(inputs) = [left.guarantee.clone(), right.guarantee.clone()] + .into_iter() + .collect::>>() + else { + return Ok(None); + }; + if inputs.iter().all(ResultGuarantee::is_exact) { + return Ok(None); + } + let Ok(guarantee) = models.accuracy.propagate( + &CompositionOperator::CheckedRelativeDivision, + &inputs, + None, + &Default::default(), + ) else { + return Ok(None); + }; + if !models.accuracy.satisfies(&guarantee, target) { + return Ok(None); + } + Ok(Some(Rc::new(SummaryNode { + expr: SummaryExpr::BinaryOp { + timing: ExecutionTiming::ReadTime, + lhs: left, + rhs: right, + operator: asap_types::post_asap::BinaryOperator { + checked_relative_division: true, + kind: BinaryOpKind::Arithmetic(asap_types::pre_asap::ArithmeticOpKind::Div), + vector_match: None, + }, + }, + schema: lift(&root.output_schema()?), + guarantee: Some(guarantee), + }))) +} + /// Preserve an exact arithmetic root while allowing each vector operand to /// select its own summary implementation. If either vector arm cannot be /// accelerated, return `None` so the caller keeps the whole query exact; @@ -1726,6 +1841,9 @@ fn realize_binary( return Ok(None); } + if let Some(candidate) = relative_division_candidate(root, models, end_to_end_target)? { + return Ok(Some(candidate)); + } let lhs_scalar = is_promql_scalar(lhs); let rhs_scalar = is_promql_scalar(rhs); if lhs_scalar && rhs_scalar { @@ -1800,6 +1918,7 @@ fn realize_binary( lhs: lhs_node, rhs: rhs_node, operator: asap_types::post_asap::BinaryOperator { + checked_relative_division: false, kind: op.clone(), vector_match: vector_match.clone(), }, @@ -5512,6 +5631,28 @@ mod tests { })) } + // A ratio needs a value-error certificate for the expression, not two rank bounds. + #[test] + fn quantile_ratio_has_a_sized_relative_value_candidate() { + 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); + let node = realize_binary(&root, models, Some(&target)) + .unwrap() + .expect("ratio candidate"); + let guarantee = node.guarantee.as_ref().expect("ratio guarantee"); + assert_eq!(guarantee.metric, ErrorMetric::RelativeValue); + assert!(DefaultAccuracyModel.satisfies(guarantee, &target)); + } + } + #[test] fn relational_join_predicate_requires_and_normalizes_cross_input_columns() { let forward = normalize_cross_input_equi_predicate(&equi_pred(1, 3), 2, 4) diff --git a/crates/asap-aware-mapping/src/rewrite.rs b/crates/asap-aware-mapping/src/rewrite.rs index 3d8eebcd..ab75d0c9 100644 --- a/crates/asap-aware-mapping/src/rewrite.rs +++ b/crates/asap-aware-mapping/src/rewrite.rs @@ -127,7 +127,7 @@ 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). -fn temporal_average_rewrite(root: &Rc) -> Option> { +pub(crate) fn temporal_average_rewrite(root: &Rc) -> Option> { let QueryExpr::Aggregate { reduction: Reduction::PerEntity, measures, 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..eabf23c7 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,7 @@ mod tests { lhs: Rc::clone(&operand), rhs: operand, operator: asap_types::post_asap::BinaryOperator { + checked_relative_division: false, kind: asap_types::pre_asap::BinaryOpKind::Arithmetic( asap_types::pre_asap::ArithmeticOpKind::Add, ), diff --git a/crates/types/src/post_asap/cse.rs b/crates/types/src/post_asap/cse.rs index 92e29913..1688a81e 100644 --- a/crates/types/src/post_asap/cse.rs +++ b/crates/types/src/post_asap/cse.rs @@ -417,6 +417,7 @@ mod tests { lhs: Rc::clone(¤t), rhs: current, operator: super::super::BinaryOperator { + checked_relative_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 53c8112e..025db139 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -198,6 +198,8 @@ pub enum ExecutionDataStateError { MaintenanceRowsAtRoot, #[error("unsupported maintenance binary schema or operator")] InvalidMaintenanceBinary, + #[error("checked relative division requires 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})")] @@ -332,6 +334,17 @@ fn visit( timing, operator, } => { + if operator.checked_relative_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() diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index aa54a0ac..d4fe80e2 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -258,6 +258,11 @@ 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, 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 770cd295..46b8a0c1 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. From 0e661f50a6febcc189f685f2f346487562571163 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 15:26:49 -0600 Subject: [PATCH 04/12] Expose the realized temporal average rewrite for physical costing --- crates/asap-aware-mapping/src/replacement.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index efdab7d4..66377ffb 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -1366,6 +1366,17 @@ 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 intent_override.is_none() { + if let Some(rewritten) = crate::rewrite::temporal_average_rewrite(root) { + if let Ok(node) = realize_child_with(&rewritten, self.models, None) { + proposals.candidates.push(ReplacementSubDAG { + replacement: Replacement::Summary(node), strategy: "SemanticEquivalentRewriteStrategy", + provenance: ReplacementProvenance::SummaryImplementation, + rationale: "realize the temporal average rewrite as independently maintained sum and count states".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 { From 03f25fef6467e8cd55a250a7b0950a7bc34f40d0 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 15:44:20 -0600 Subject: [PATCH 05/12] Realize exact Top-K over maintained temporal values --- crates/asap-aware-mapping/src/replacement.rs | 90 ++++++++++++++++++++ 1 file changed, 90 insertions(+) diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 66377ffb..75fc4ecf 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -1366,6 +1366,15 @@ 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 Some(rewritten) = crate::rewrite::temporal_average_rewrite(root) { if let Ok(node) = realize_child_with(&rewritten, self.models, None) { @@ -1674,6 +1683,62 @@ 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 { .. }]) { + 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)) +} + pub(crate) fn realize_child_with( root: &Rc, models: Models<'_>, @@ -5643,6 +5708,31 @@ mod tests { } // A ratio needs a value-error certificate for the expression, not two rank bounds. + // 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(); + } + } + #[test] fn quantile_ratio_has_a_sized_relative_value_candidate() { let target = AccuracyTarget::EpsilonDelta { From a0f43e48563958b2ea6a0aea73b2a31578f41157 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 15:49:57 -0600 Subject: [PATCH 06/12] Keep approximate temporal Top-K candidate selection unchanged --- crates/asap-aware-mapping/src/replacement.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 75fc4ecf..08157d7b 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -1697,7 +1697,13 @@ fn exact_topk_over_temporal_values( else { return Ok(None); }; - if !matches!(measures.as_slice(), [AggIntent::TopK { .. }]) { + if !matches!( + measures.as_slice(), + [AggIntent::TopK { + accuracy: AccuracyTarget::Exact, + .. + }] + ) { return Ok(None); } let QueryExpr::Aggregate { From 8a943abcefc009e4265fcea4ac3235245c342d52 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 12 Sep 2026 16:00:13 -0600 Subject: [PATCH 07/12] Update integration contracts for row counts minimum and temporal average --- crates/integration-tests/tests/aggregate.rs | 7 +++---- crates/integration-tests/tests/exact_composition.rs | 2 +- crates/integration-tests/tests/promql_to_post_asap.rs | 2 +- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/crates/integration-tests/tests/aggregate.rs b/crates/integration-tests/tests/aggregate.rs index 1c9ffb8b..1a1e0ca2 100644 --- a/crates/integration-tests/tests/aggregate.rs +++ b/crates/integration-tests/tests/aggregate.rs @@ -65,15 +65,14 @@ fn q06_sum_by_job() { ); } -// #7 — PromQL `count` is cross-series cardinality, not per-sample Count +// #7 — PromQL `count` counts current vector rows, including repeated sample values. #[test] -fn q07_count_is_cardinality() { +fn q07_count_counts_series_rows() { assert_eq!( lower("count(http_requests_total)"), agg( vec![], - AggIntent::Cardinality { - col: None, + AggIntent::Count { accuracy: AccuracyTarget::Exact }, scan("http_requests_total", &[]), diff --git a/crates/integration-tests/tests/exact_composition.rs b/crates/integration-tests/tests/exact_composition.rs index fe32e568..86b5ad1f 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( diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index 0668a4d3..3945a2b6 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -414,7 +414,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(_))); } From 304c2a9f12a37db864d4dfe324e8a584a13da30b Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 07:36:44 -0600 Subject: [PATCH 08/12] Generalize maintained population rules across SQL and PromQL --- Cargo.lock | 1 + crates/asap-aware-mapping/src/lib.rs | 2 +- ...ent_series.rs => maintained_population.rs} | 178 +++++++++++++----- crates/frontend-sql/Cargo.toml | 1 + .../tests/maintained_population.rs | 160 ++++++++++++++++ .../src/post_asap/execution_data_state.rs | 22 +-- crates/types/src/post_asap/expr.rs | 14 +- ...ent_series.rs => maintained_population.rs} | 59 ++++-- crates/types/src/post_asap/mod.rs | 2 +- .../current-series-populations.md | 62 +++--- 10 files changed, 393 insertions(+), 108 deletions(-) rename crates/asap-aware-mapping/src/{current_series.rs => maintained_population.rs} (65%) create mode 100644 crates/frontend-sql/tests/maintained_population.rs rename crates/types/src/post_asap/{current_series.rs => maintained_population.rs} (62%) diff --git a/Cargo.lock b/Cargo.lock index d9e299c2..02a0ffdc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -350,6 +350,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/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index 7f03f18d..b3262c28 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -255,4 +255,4 @@ pub use summary_maintenance_lifecycle::{ }; pub use topk_reuse::TopKLimitReuseStrategy; -pub mod current_series; +pub mod maintained_population; diff --git a/crates/asap-aware-mapping/src/current_series.rs b/crates/asap-aware-mapping/src/maintained_population.rs similarity index 65% rename from crates/asap-aware-mapping/src/current_series.rs rename to crates/asap-aware-mapping/src/maintained_population.rs index 157859b1..ba9de4f6 100644 --- a/crates/asap-aware-mapping/src/current_series.rs +++ b/crates/asap-aware-mapping/src/maintained_population.rs @@ -1,9 +1,9 @@ -//! Exact current-series population candidates over canonical PromQL IR. +//! Shared maintained-population candidates over canonical relational IR. use crate::replacement::{ Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG, }; use asap_types::post_asap::{ - current_series::*, ExecutionTiming, ResultGuarantee, SummaryExpr, SummaryFamilyType, + maintained_population::*, ExecutionTiming, ResultGuarantee, SummaryExpr, SummaryFamilyType, SummaryField, SummaryNode, SummarySchema, ValueOperation, }; use asap_types::pre_asap::{ @@ -26,10 +26,16 @@ fn plain(schema: Schema) -> SummarySchema { } } -fn recognize( - root: &QueryExpr, -) -> Option<(CurrentSeriesPopulation, CurrentSeriesReadout, Rc)> { - let (source, grouping, readout) = match root { +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), @@ -42,18 +48,18 @@ fn recognize( }; let (col, readout) = match intent { AggIntent::Quantile { q, col, .. } if q.is_finite() => { - (*col, CurrentSeriesReadout::Quantile { q: *q }) + (*col, PopulationReadout::Quantile { q: *q }) } - AggIntent::Sum { col } => (*col, CurrentSeriesReadout::Sum), - AggIntent::Count { .. } => (None, CurrentSeriesReadout::Count), - AggIntent::Avg { col } => (*col, CurrentSeriesReadout::Average), + 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_or(|c| c.name != "value")) { + if col.is_some_and(|c| schema.columns.get(c).is_none()) { return None; } - (child, grouping, readout) + (child, grouping, readout, col) } QueryExpr::Limit { n, @@ -74,13 +80,44 @@ fn recognize( let QueryExpr::Column(col) = &key.expr else { return None; }; - if key.ascending || child.output_schema().ok()?.columns.get(*col)?.name != "value" { + if key.ascending { return None; } - (child, partition_by, CurrentSeriesReadout::TopK { k: *n }) + ( + 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, @@ -89,6 +126,9 @@ fn recognize( 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; @@ -130,12 +170,14 @@ fn recognize( labels.sort(); labels.dedup(); Some(( - CurrentSeriesPopulation { - metric: metric.clone(), - matchers, - grouping: labels, - without: grouping.is_without(), - lookback_ms: 300_000, + MaintainedPopulation { + input: PopulationInput::CurrentSeries(CurrentSeriesInput { + metric: metric.clone(), + matchers, + grouping: labels, + without: grouping.is_without(), + lookback_ms: 300_000, + }), max_k: 0, quantiles: false, }, @@ -146,31 +188,49 @@ fn recognize( /// Workload-aware rule: compatible readouts share one retractable population. /// Deployments opt in by registering this strategy when they can maintain complete -/// current-series inputs and price the maintenance/readout boundary. +/// population updates and price the maintenance/readout boundary. /// The population is exact; max_k bounds the shared readout cache, not its members. -pub struct CurrentSeriesStrategy { +pub struct MaintainedPopulationStrategy { roots: Vec>, } -impl CurrentSeriesStrategy { +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 { - CurrentSeriesReadout::Quantile { .. } => population.quantiles = true, - CurrentSeriesReadout::TopK { k } => { - population.max_k = population.max_k.max(k) - } - CurrentSeriesReadout::Sum - | CurrentSeriesReadout::Count - | CurrentSeriesReadout::Average => {} + PopulationReadout::Quantile { .. } => population.quantiles = true, + PopulationReadout::TopK { k } => population.max_k = population.max_k.max(k), + PopulationReadout::Sum + | PopulationReadout::Count + | PopulationReadout::Average => {} } } } @@ -184,18 +244,18 @@ impl CurrentSeriesStrategy { let maintained = Rc::new(SummaryNode { expr: SummaryExpr::ValueOperation { child: scan, - operation: ValueOperation::MaintainCurrentSeries { population }, + operation: ValueOperation::MaintainPopulation { population }, timing: ExecutionTiming::MaintenanceTime, }, schema: input_schema, guarantee: Some(ResultGuarantee::exact( - "latest value per series with stale retraction and lookback expiry", + "exact members under the declared population semantics", )), }); Some(Rc::new(SummaryNode { expr: SummaryExpr::ValueOperation { child: maintained, - operation: ValueOperation::ReadCurrentSeries { readout }, + operation: ValueOperation::ReadPopulation { readout }, timing: ExecutionTiming::ReadTime, }, schema: plain(root.output_schema().ok()?), @@ -203,12 +263,22 @@ impl CurrentSeriesStrategy { })) } } -impl ReplacementStrategy for CurrentSeriesStrategy { +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: "CurrentSeriesStrategy", replacement: Replacement::Summary(node), provenance: ReplacementProvenance::SummaryImplementation, rationale: "share an exact retractable current-series population across quantiles and TopK limits".into() }).into_iter().collect() + 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() } } @@ -236,7 +306,7 @@ mod tests { ] .map(lower) .into(); - let rule = CurrentSeriesStrategy::new(&roots); + let rule = MaintainedPopulationStrategy::new(&roots); for root in roots { let candidate = rule .candidate(&root) @@ -256,19 +326,19 @@ mod tests { ] .map(lower) .into(); - let strategy = CurrentSeriesStrategy::new(&roots); + let strategy = MaintainedPopulationStrategy::new(&roots); let space = crate::search_workload_with( roots .iter() .enumerate() .map(|(i, r)| (i, Rc::clone(r))) .collect(), - &[Box::new(CurrentSeriesStrategy::new(&roots))], + &[Box::new(MaintainedPopulationStrategy::new(&roots))], ); assert!(space .groups() .flat_map(|g| &g.candidates) - .any(|c| c.strategy == "CurrentSeriesStrategy")); + .any(|c| c.strategy == "MaintainedPopulationStrategy")); let plans = share_common_summary_subtrees( roots .iter() @@ -281,14 +351,14 @@ mod tests { compile_executable_dag(plan).unwrap(); let SummaryExpr::ValueOperation { child, - operation: ValueOperation::ReadCurrentSeries { .. }, + operation: ValueOperation::ReadPopulation { .. }, .. } = &plan.expr else { panic!("missing typed readout") }; let SummaryExpr::ValueOperation { - operation: ValueOperation::MaintainCurrentSeries { population }, + operation: ValueOperation::MaintainPopulation { population }, .. } = &child.expr else { @@ -312,15 +382,15 @@ mod tests { ] .map(lower) .into(); - let strategy = CurrentSeriesStrategy::new(&roots); + let strategy = MaintainedPopulationStrategy::new(&roots); let (p, _, _) = recognize(&roots[0]).unwrap(); - assert!(p.grouping.is_empty()); + 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::MaintainCurrentSeries { population }, + operation: ValueOperation::MaintainPopulation { population }, .. } = &child.expr else { @@ -342,6 +412,9 @@ mod tests { } 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); @@ -350,14 +423,14 @@ mod tests { #[test] fn malformed_population_dags_fail_closed() { let root = lower("topk(5,a)"); - let strategy = CurrentSeriesStrategy::new(std::slice::from_ref(&root)); + 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::ReadCurrentSeries { - readout: CurrentSeriesReadout::TopK { k: 6 }, + *operation = ValueOperation::ReadPopulation { + readout: PopulationReadout::TopK { k: 6 }, }; assert!(compile_executable_dag(&Rc::new(bad.clone())).is_err()); let SummaryExpr::ValueOperation { @@ -366,18 +439,21 @@ mod tests { else { unreachable!() }; - *operation = ValueOperation::ReadCurrentSeries { - readout: CurrentSeriesReadout::TopK { k: 5 }, + *operation = ValueOperation::ReadPopulation { + readout: PopulationReadout::TopK { k: 5 }, }; let producer = Rc::make_mut(child); let SummaryExpr::ValueOperation { - operation: ValueOperation::MaintainCurrentSeries { population }, + operation: ValueOperation::MaintainPopulation { population }, .. } = &mut producer.expr else { unreachable!() }; - population.metric = "b".into(); + 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/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..9dc1bb4d --- /dev/null +++ b/crates/frontend-sql/tests/maintained_population.rs @@ -0,0 +1,160 @@ +//! 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()); +} diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs index 025db139..f5690d60 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -153,8 +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 current-series maintenance/readout contract")] - InvalidCurrentSeries, + #[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. @@ -477,19 +477,19 @@ fn visit( operation, timing, } => { - let valid_current = match operation { - ValueOperation::MaintainCurrentSeries { population } => { + let valid_population = match operation { + ValueOperation::MaintainPopulation { population } => { *timing == ExecutionTiming::MaintenanceTime && matches!(&child.expr, SummaryExpr::KeepPreAsap(input) if population.matches_input(input)) } - ValueOperation::ReadCurrentSeries { readout } => { + ValueOperation::ReadPopulation { readout } => { *timing == ExecutionTiming::ReadTime - && matches!(&child.expr, SummaryExpr::ValueOperation { operation: ValueOperation::MaintainCurrentSeries { population }, timing: ExecutionTiming::MaintenanceTime, .. } if population.supports(readout)) + && matches!(&child.expr, SummaryExpr::ValueOperation { operation: ValueOperation::MaintainPopulation { population }, timing: ExecutionTiming::MaintenanceTime, .. } if population.supports(readout)) } _ => true, }; - if !valid_current { - return Err(ExecutionDataStateError::InvalidCurrentSeries); + if !valid_population { + return Err(ExecutionDataStateError::InvalidMaintainedPopulation); } let required = match timing { ExecutionTiming::MaintenanceTime => ExecutionDataState::MAINTENANCE_ROWS, @@ -500,17 +500,17 @@ fn visit( || matches!(operation, ValueOperation::FinalizeExactAccumulator)) && s == ExecutionDataState::MAINTENANCE_SUMMARY && is_exact_accumulator_state(&child.schema).is_ok(); - let current_readout = matches!(operation, ValueOperation::ReadCurrentSeries { .. }) + let population_readout = matches!(operation, ValueOperation::ReadPopulation { .. }) && *timing == ExecutionTiming::ReadTime && matches!( &child.expr, SummaryExpr::ValueOperation { - operation: ValueOperation::MaintainCurrentSeries { .. }, + operation: ValueOperation::MaintainPopulation { .. }, timing: ExecutionTiming::MaintenanceTime, .. } ); - if s != required && !exact_readout && !current_readout { + 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 d4fe80e2..655bc2ca 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -25,14 +25,14 @@ pub enum ExactOperation { #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] #[non_exhaustive] pub enum ValueOperation { - /// Replace each series' current value, retract stale/expired values, and - /// retain the full population so removing a TopK member can promote another. - MaintainCurrentSeries { - population: super::current_series::CurrentSeriesPopulation, + /// Maintain the full declared population, including membership changes, + /// so removing a TopK member can promote another. + MaintainPopulation { + population: super::maintained_population::MaintainedPopulation, }, - /// Read one quantile or TopK prefix from the maintained current population. - ReadCurrentSeries { - readout: super::current_series::CurrentSeriesReadout, + /// 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. diff --git a/crates/types/src/post_asap/current_series.rs b/crates/types/src/post_asap/maintained_population.rs similarity index 62% rename from crates/types/src/post_asap/current_series.rs rename to crates/types/src/post_asap/maintained_population.rs index 6820d36f..fc953d53 100644 --- a/crates/types/src/post_asap/current_series.rs +++ b/crates/types/src/post_asap/maintained_population.rs @@ -1,16 +1,14 @@ -//! Semantic contract for a retractable population of current PromQL series values. +//! 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 CurrentSeriesPopulation { +pub struct CurrentSeriesInput { pub metric: String, pub matchers: Vec, pub grouping: Vec, pub without: bool, pub lookback_ms: u64, - pub max_k: usize, - pub quantiles: bool, } #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] @@ -29,7 +27,7 @@ pub enum CurrentSeriesMatch { } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub enum CurrentSeriesReadout { +pub enum PopulationReadout { Quantile { q: f64 }, TopK { k: usize }, Sum, @@ -37,7 +35,7 @@ pub enum CurrentSeriesReadout { Average, } -impl CurrentSeriesPopulation { +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}; @@ -98,13 +96,50 @@ impl CurrentSeriesPopulation { matchers.dedup(); self.matchers == matchers && self.grouping.windows(2).all(|w| w[0] < w[1]) } - pub fn supports(&self, readout: &CurrentSeriesReadout) -> bool { +} + +/// 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 { - CurrentSeriesReadout::Quantile { q } => self.quantiles && q.is_finite(), - CurrentSeriesReadout::TopK { k } => *k <= self.max_k, - CurrentSeriesReadout::Sum - | CurrentSeriesReadout::Count - | CurrentSeriesReadout::Average => true, + 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 f6fea4f2..5b7ba43b 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -28,11 +28,11 @@ //! — see `asap_aware_mapping::grouping`'s module docs for why. pub mod cse; -pub mod current_series; 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/docs/design_docs/asap-aware-mapping/current-series-populations.md b/docs/design_docs/asap-aware-mapping/current-series-populations.md index 16b7e7a1..5f8e0df8 100644 --- a/docs/design_docs/asap-aware-mapping/current-series-populations.md +++ b/docs/design_docs/asap-aware-mapping/current-series-populations.md @@ -1,32 +1,44 @@ -# Current-series population candidates +# Maintained population candidates -`CurrentSeriesStrategy` is an opt-in `ReplacementStrategy` for deployments that -can continuously maintain complete PromQL current-series inputs. It matches -canonical cross-series quantile aggregates and descending value Sort/Limit over -a direct open time-series Scan. Temporal ranges, shifted selectors, nested inputs -and bottom-k do not match. - -The selected post-ASAP DAG is: +`MaintainedPopulationStrategy` is an opt-in rule over canonical relational IR. +It separates membership semantics from shared aggregate readouts: ```text -KeepPreAsap(Scan) [maintenance rows] - -> ValueOperation::MaintainCurrentSeries(population) [maintenance rows] - -> ValueOperation::ReadCurrentSeries(quantile q | top-k k) [read rows] +KeepPreAsap(input) + -> MaintainPopulation { input, max_k, quantiles } [maintenance] + -> ReadPopulation { Quantile(q) | TopK(k) | Sum | Count | Average } [read] + -> optional SQL projection ``` -The population owns source, label predicates, grouping and five-minute selector -lookback semantics. Updates replace each series' latest value; stale markers and -expiry retract it. It retains the full population, not only the largest k values. -Compatible workload consumers use the largest requested k and one quantile -population; canonical CSE shares their maintenance producer. Different sources, -matchers or groups do not share. The readout remains exact. +`PopulationInput::CurrentSeries` selects each series' latest live value with +label matchers, grouping, stale retraction and the canonical five-minute lookback. +`PopulationInput::Rows` retains the full input multiset, including duplicate rows; +it carries the table scan (including predicates), value column and grouping. +Table populations do not inherit latest-value selection or lookback expiry. +The initial table rule supports non-null Float64 value columns. It preserves SQL +projections and aliases; unsupported types, nullable value columns, multi-measure +aggregates and arbitrary relational inputs need further rules. + +Compatible consumers share the maintenance producer through canonical CSE. +Quantile ranks are readout parameters; TopK requests retain the largest requested +k. The full population remains available so deletion of a TopK member can promote +another. Source, predicates, value column, grouping and membership semantics are +part of state identity. The maintained population and its readouts are exact; +this operator does not imply a deletable DDSketch. + +IR validation checks the maintenance input, execution phases and producer/readout +compatibility. A compiler must explicitly support the selected membership model: +remote-write current-series execution is not an implementation of table-row state. +The backend currently deploys the current-series variant; table-row deployment +requires a complete row-update/deletion executor. Existing SQL summary compilation +continues separately. -IR validation checks the maintenance input against the population contract, -phase placement and the readout's compatibility with its producer. The only new -maintenance-row/read-row bridge is this explicit producer/readout pair. +Temporal sketch rules remain SummaryAgg/SummaryEstimate DAGs. UnivMon can expose +cardinality, frequency L2 and entropy readouts over the same unit-frequency input. +Sharing requires matching input, partitioning, windows and sketch parameters; +accuracy admission remains readout-specific. Entropy evidence cannot certify L2 +or cardinality. State sharing alone does not establish an error bound or cost win. -The strategy does not parse query text, choose a deployment, assign byte budgets -or claim a performance benefit. Backends lower these typed operations into their -state implementation, bind resource/coverage limits and price build, update, -residency, retirement and readout costs. Other deployments must keep it disabled -until they support that contract. Existing default strategy selection is unchanged. +The rule does not parse query text, choose a deployment, assign memory limits or +claim performance gains. Compilers lower supported typed DAGs and price complete +maintenance and readout costs before installation. From 7f84cbbd8773d8dedc69890d62b18a3d4e479779 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 07:38:23 -0600 Subject: [PATCH 09/12] Verify SQL TopK consumers share maximum-k population --- .../tests/maintained_population.rs | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/crates/frontend-sql/tests/maintained_population.rs b/crates/frontend-sql/tests/maintained_population.rs index 9dc1bb4d..6ad7e987 100644 --- a/crates/frontend-sql/tests/maintained_population.rs +++ b/crates/frontend-sql/tests/maintained_population.rs @@ -158,3 +158,25 @@ async fn malformed_table_population_fails_validation() { *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)); + } +} From 04b8e9dc36322b0b7951ad6e25ce84b17d891618 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 08:27:03 -0600 Subject: [PATCH 10/12] fix: guard temporal average division against intermediate overflow --- crates/asap-aware-mapping/src/replacement.rs | 66 ++++++++++++++++--- crates/asap-aware-mapping/src/rewrite.rs | 27 ++++---- .../src/summary_maintenance_cost/model.rs | 1 + crates/types/src/post_asap/cse.rs | 1 + .../src/post_asap/execution_data_state.rs | 19 +++--- crates/types/src/post_asap/expr.rs | 5 ++ 6 files changed, 84 insertions(+), 35 deletions(-) diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 08157d7b..deef4095 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -1376,14 +1376,13 @@ impl<'a> SketchAlgorithmStrategy<'a> { }); } if intent_override.is_none() { - if let Some(rewritten) = crate::rewrite::temporal_average_rewrite(root) { - if let Ok(node) = realize_child_with(&rewritten, self.models, None) { - proposals.candidates.push(ReplacementSubDAG { - replacement: Replacement::Summary(node), strategy: "SemanticEquivalentRewriteStrategy", - provenance: ReplacementProvenance::SummaryImplementation, - rationale: "realize the temporal average rewrite as independently maintained sum and count states".into(), - }); - } + 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) { @@ -1745,13 +1744,30 @@ fn exact_topk_over_temporal_values( 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(rewritten) = crate::rewrite::temporal_average_rewrite(root) { - return realize_child_with(&rewritten, models, end_to_end_target); + 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); @@ -1892,6 +1908,7 @@ fn relative_division_candidate( rhs: right, operator: asap_types::post_asap::BinaryOperator { checked_relative_division: true, + checked_finite_division: false, kind: BinaryOpKind::Arithmetic(asap_types::pre_asap::ArithmeticOpKind::Div), vector_match: None, }, @@ -2001,6 +2018,7 @@ fn realize_binary( 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(), }, @@ -5713,6 +5731,34 @@ 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" + ); + } + // A ratio needs a value-error certificate for the expression, not two rank bounds. // Exact Top-K consumes the Planner's maintained temporal values. #[test] diff --git a/crates/asap-aware-mapping/src/rewrite.rs b/crates/asap-aware-mapping/src/rewrite.rs index ab75d0c9..1c481d86 100644 --- a/crates/asap-aware-mapping/src/rewrite.rs +++ b/crates/asap-aware-mapping/src/rewrite.rs @@ -29,8 +29,8 @@ //! ## Scope //! //! Ordinary `by(...)` averages use a schema-preserving projection. Temporal -//! Float64 averages use two independent per-entity accumulators and direct -//! division, preserving open series labels and the single-measure invariant. +//! Float64 temporal averages require a typed finite-division guard; their +//! sum/count components are never exported as an unconditional logical rewrite. //! //! - **`without(...)` grouping** leaves an `Aggregate`'s own output schema //! *open* (`closed: false`, see `without_output_schema`), while the @@ -127,7 +127,9 @@ 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). -pub(crate) fn temporal_average_rewrite(root: &Rc) -> Option> { +// 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, @@ -172,9 +174,6 @@ pub(crate) fn temporal_average_rewrite(root: &Rc) -> Option) -> Option> { - if let Some(rewritten) = temporal_average_rewrite(root) { - return Some(rewritten); - } let (group_count, col) = avg_rewrite_target(root)?; let QueryExpr::Aggregate { reduction, @@ -355,7 +354,6 @@ pub use SemanticEquivalentRewriteStrategy as AvgToSumOverCountStrategy; impl ReplacementStrategy for SemanticEquivalentRewriteStrategy { fn matches(&self, target: &TargetSubDAG<'_>) -> bool { avg_rewrite_target(target.root).is_some() - || temporal_average_rewrite(target.root).is_some() || composed_aggregate_rewrite(target.root).is_some() } @@ -420,7 +418,7 @@ mod tests { // Temporal averages expose two single-measure children without closing labels. #[test] - fn temporal_average_rewrite_preserves_schema_and_exposes_sum_count() { + 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])", @@ -428,14 +426,11 @@ mod tests { ) .unwrap(), ); - let rewrites = SemanticEquivalentRewriteStrategy.replacements(&TargetSubDAG::new(&root)); - let rewritten = rewrites - .iter() - .find_map(|r| match &r.replacement { - Replacement::Rewrite(q) => Some(q), - _ => None, - }) - .expect("temporal average rewrite"); + 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() 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 eabf23c7..9cfc370d 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs @@ -3005,6 +3005,7 @@ mod tests { 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/types/src/post_asap/cse.rs b/crates/types/src/post_asap/cse.rs index 1688a81e..a8077f74 100644 --- a/crates/types/src/post_asap/cse.rs +++ b/crates/types/src/post_asap/cse.rs @@ -418,6 +418,7 @@ mod tests { 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 f5690d60..2dcd15b3 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -198,7 +198,7 @@ pub enum ExecutionDataStateError { MaintenanceRowsAtRoot, #[error("unsupported maintenance binary schema or operator")] InvalidMaintenanceBinary, - #[error("checked relative division requires a read-time division operator")] + #[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. @@ -334,14 +334,15 @@ fn visit( timing, operator, } => { - if operator.checked_relative_division - && (*timing != ExecutionTiming::ReadTime - || !matches!( - operator.kind, - crate::pre_asap::BinaryOpKind::Arithmetic( - crate::pre_asap::ArithmeticOpKind::Div - ) - )) + 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); } diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index 655bc2ca..c5c878f9 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -263,6 +263,11 @@ pub struct BinaryOperator { /// 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 From b8b5d705215200361b6a2c3b389d513d554077ca Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 08:27:06 -0600 Subject: [PATCH 11/12] docs: define maintained populations and specify the shared-state rule --- docs/design_docs/asap-aware-mapping/README.md | 3 + .../current-series-populations.md | 46 +---- .../maintained-populations.md | 165 ++++++++++++++++++ .../asap-aware-mapping/optimizations.md | 18 ++ .../physical-plan-integration.md | 16 ++ 5 files changed, 205 insertions(+), 43 deletions(-) create mode 100644 docs/design_docs/asap-aware-mapping/maintained-populations.md 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/current-series-populations.md b/docs/design_docs/asap-aware-mapping/current-series-populations.md index 5f8e0df8..c9432421 100644 --- a/docs/design_docs/asap-aware-mapping/current-series-populations.md +++ b/docs/design_docs/asap-aware-mapping/current-series-populations.md @@ -1,44 +1,4 @@ -# Maintained population candidates +# Maintained populations -`MaintainedPopulationStrategy` is an opt-in rule over canonical relational IR. -It separates membership semantics from shared aggregate readouts: - -```text -KeepPreAsap(input) - -> MaintainPopulation { input, max_k, quantiles } [maintenance] - -> ReadPopulation { Quantile(q) | TopK(k) | Sum | Count | Average } [read] - -> optional SQL projection -``` - -`PopulationInput::CurrentSeries` selects each series' latest live value with -label matchers, grouping, stale retraction and the canonical five-minute lookback. -`PopulationInput::Rows` retains the full input multiset, including duplicate rows; -it carries the table scan (including predicates), value column and grouping. -Table populations do not inherit latest-value selection or lookback expiry. -The initial table rule supports non-null Float64 value columns. It preserves SQL -projections and aliases; unsupported types, nullable value columns, multi-measure -aggregates and arbitrary relational inputs need further rules. - -Compatible consumers share the maintenance producer through canonical CSE. -Quantile ranks are readout parameters; TopK requests retain the largest requested -k. The full population remains available so deletion of a TopK member can promote -another. Source, predicates, value column, grouping and membership semantics are -part of state identity. The maintained population and its readouts are exact; -this operator does not imply a deletable DDSketch. - -IR validation checks the maintenance input, execution phases and producer/readout -compatibility. A compiler must explicitly support the selected membership model: -remote-write current-series execution is not an implementation of table-row state. -The backend currently deploys the current-series variant; table-row deployment -requires a complete row-update/deletion executor. Existing SQL summary compilation -continues separately. - -Temporal sketch rules remain SummaryAgg/SummaryEstimate DAGs. UnivMon can expose -cardinality, frequency L2 and entropy readouts over the same unit-frequency input. -Sharing requires matching input, partitioning, windows and sketch parameters; -accuracy admission remains readout-specific. Entropy evidence cannot certify L2 -or cardinality. State sharing alone does not establish an error bound or cost win. - -The rule does not parse query text, choose a deployment, assign memory limits or -claim performance gains. Compilers lower supported typed DAGs and price complete -maintenance and readout costs before installation. +The definition and complete rule specification have moved to +[Shared maintained population rule](maintained-populations.md). 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..74469111 --- /dev/null +++ b/docs/design_docs/asap-aware-mapping/maintained-populations.md @@ -0,0 +1,165 @@ +# 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. + +## 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. From 84287fe274fc432ae872a3fae0292820a7b879eb Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 09:09:31 -0600 Subject: [PATCH 12/12] docs: remove obsolete current-series population entry --- .../asap-aware-mapping/current-series-populations.md | 4 ---- 1 file changed, 4 deletions(-) delete mode 100644 docs/design_docs/asap-aware-mapping/current-series-populations.md diff --git a/docs/design_docs/asap-aware-mapping/current-series-populations.md b/docs/design_docs/asap-aware-mapping/current-series-populations.md deleted file mode 100644 index c9432421..00000000 --- a/docs/design_docs/asap-aware-mapping/current-series-populations.md +++ /dev/null @@ -1,4 +0,0 @@ -# Maintained populations - -The definition and complete rule specification have moved to -[Shared maintained population rule](maintained-populations.md).