From 61fdb613dccc8e27ad469561fce0803227df884a Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 21:47:04 -0600 Subject: [PATCH 1/3] feat(promql): bind canonical queries to native exact kernels --- Cargo.lock | 10 +- Cargo.toml | 8 +- control_plane/src/physical/mod.rs | 2 + control_plane/src/physical/promql_exact.rs | 382 ++++++++++++ .../query_engines/canonical/exact_promql.rs | 579 ++++++++++++++++++ data_plane/src/query_engines/canonical/mod.rs | 2 + 6 files changed, 974 insertions(+), 9 deletions(-) create mode 100644 control_plane/src/physical/promql_exact.rs create mode 100644 data_plane/src/query_engines/canonical/exact_promql.rs diff --git a/Cargo.lock b/Cargo.lock index e173aa68..7e1a22ba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -364,7 +364,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f27b16a747e5d7fcd70a5510075c0cd062f0dcea#f27b16a747e5d7fcd70a5510075c0cd062f0dcea" dependencies = [ "asap-types", "serde", @@ -375,7 +375,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f27b16a747e5d7fcd70a5510075c0cd062f0dcea#f27b16a747e5d7fcd70a5510075c0cd062f0dcea" dependencies = [ "asap-types", "promql-parser 0.10.0 (git+https://github.com/ProjectASAP/promql-parser?rev=9fede7eecca923c9882fe256484d00d37f8706cb)", @@ -384,7 +384,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f27b16a747e5d7fcd70a5510075c0cd062f0dcea#f27b16a747e5d7fcd70a5510075c0cd062f0dcea" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -407,12 +407,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f27b16a747e5d7fcd70a5510075c0cd062f0dcea#f27b16a747e5d7fcd70a5510075c0cd062f0dcea" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f27b16a747e5d7fcd70a5510075c0cd062f0dcea#f27b16a747e5d7fcd70a5510075c0cd062f0dcea" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index ae7e90f8..0165c059 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,10 +20,10 @@ asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", branch [workspace.dependencies] # Keep Planner frontends, selection, and IR on the same immutable revision (current-series Planner PR). # Alias upstream asap-types because this workspace also defines asap_types. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" } -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" } -asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f27b16a747e5d7fcd70a5510075c0cd062f0dcea" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f27b16a747e5d7fcd70a5510075c0cd062f0dcea" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f27b16a747e5d7fcd70a5510075c0cd062f0dcea" } +asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f27b16a747e5d7fcd70a5510075c0cd062f0dcea" } # Shared external deps (used by 2+ crates) serde = { version = "1.0", features = ["derive"] } diff --git a/control_plane/src/physical/mod.rs b/control_plane/src/physical/mod.rs index a31de4e5..8ae2ca9b 100644 --- a/control_plane/src/physical/mod.rs +++ b/control_plane/src/physical/mod.rs @@ -16,3 +16,5 @@ pub mod workload_cost; pub mod publication; pub(crate) mod maintained_population; + +pub mod promql_exact; diff --git a/control_plane/src/physical/promql_exact.rs b/control_plane/src/physical/promql_exact.rs new file mode 100644 index 00000000..92ae2e56 --- /dev/null +++ b/control_plane/src/physical/promql_exact.rs @@ -0,0 +1,382 @@ +//! Executable exact kernels for the PromQL float-sample surface. +//! +//! Summary binding cannot implement ordered-window reducers or label-producing +//! operators with the existing five accumulator families. This plan binds each +//! operator to a backend kernel and explicitly requires raw timestamped samples. +use std::rc::Rc; + +use planner_types::pre_asap::{ + AggIntent, CompareOpKind, GroupKeys, QueryExpr, Reduction, SampleKind, ScalarValue, Source, + VectorMatchKind, +}; +use planner_types::types::AccuracyTarget; +use promql_parser::label::Matcher; +use promql_parser::parser::token; + +macro_rules! kernels { + ($name:ident { $($variant:ident => $text:literal),+ $(,)? }) => { + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub enum $name { $($variant),+ } + impl std::str::FromStr for $name { + type Err = anyhow::Error; + fn from_str(name: &str) -> anyhow::Result { + match name { $($text => Ok(Self::$variant),)+ _ => anyhow::bail!("no exact kernel for {name}") } + } + } + } +} +kernels!(AggregateKernel { + Sum => "sum", Avg => "avg", Count => "count", Min => "min", Max => "max", + Group => "group", Stddev => "stddev", Stdvar => "stdvar", TopK => "topk", + BottomK => "bottomk", CountValues => "count_values", Quantile => "quantile", + LimitK => "limitk", LimitRatio => "limit_ratio", +}); +kernels!(RangeKernel { + Avg => "avg_over_time", Min => "min_over_time", Max => "max_over_time", + Sum => "sum_over_time", Count => "count_over_time", Quantile => "quantile_over_time", + Stddev => "stddev_over_time", Stdvar => "stdvar_over_time", Last => "last_over_time", + Present => "present_over_time", Absent => "absent_over_time", Changes => "changes", + Delta => "delta", Deriv => "deriv", IDelta => "idelta", Increase => "increase", + IRate => "irate", PredictLinear => "predict_linear", Rate => "rate", Resets => "resets", + Smoothing => "double_exponential_smoothing", Mad => "mad_over_time", + TsMin => "ts_of_min_over_time", TsMax => "ts_of_max_over_time", TsLast => "ts_of_last_over_time", +}); +kernels!(BinaryKernel { + Add => "+", Sub => "-", Mul => "*", Div => "/", Mod => "%", Pow => "^", + And => "and", Or => "or", Unless => "unless", +}); + +#[derive(Debug, Clone)] +pub struct Grouping { + pub labels: Vec, + pub without: bool, +} + +#[derive(Debug, Clone)] +pub struct ExactSelector { + pub metric: String, + pub matchers: Vec, +} + +#[derive(Debug, Clone)] +pub enum ExactExpr { + Scalar(f64), + Select { + selector: ExactSelector, + range_seconds: Option, + }, + Aggregate { + kernel: AggregateKernel, + parameter: Option, + label: Option, + grouping: Grouping, + input: Box, + }, + Range { + kernel: RangeKernel, + parameters: Vec, + input: Box, + }, + Binary { + kernel: BinaryKernel, + lhs: Box, + rhs: Box, + }, +} + +#[derive(Debug, Clone)] +pub struct ExactPromqlPlan { + /// The canonical frontend result is retained for review and semantic regression checks. + pub canonical: Rc, + root: ExactExpr, +} + +impl ExactPromqlPlan { + pub fn bind(query: &str) -> anyhow::Result { + let canonical = Rc::new(crate::query_parser::parse_query_expr_canonical( + query, + AccuracyTarget::Exact, + )?); + Self::from_canonical(canonical) + } + + /// Bind the planner's canonical tree; execution never reparses the query text. + pub fn from_canonical(canonical: Rc) -> anyhow::Result { + let root = bind_expr(&canonical)?; + anyhow::ensure!( + !matches!( + &root, + ExactExpr::Scalar(_) + | ExactExpr::Select { + range_seconds: Some(_), + .. + } + ), + "exact endpoint requires an instant-vector result" + ); + Ok(Self { canonical, root }) + } + + pub fn root(&self) -> &ExactExpr { + &self.root + } +} + +fn grouping(keys: &GroupKeys, child: &QueryExpr) -> anyhow::Result { + let schema = child.output_schema()?; + let labels = keys + .keys() + .iter() + .map(|index| { + let column = schema + .columns + .get(*index) + .ok_or_else(|| anyhow::anyhow!("invalid grouping column {index}"))?; + anyhow::ensure!( + column.name != "ts" && column.name != "value", + "grouping requires label columns" + ); + Ok(column.name.clone()) + }) + .collect::>>()?; + Ok(Grouping { + labels, + without: keys.is_without(), + }) +} + +fn bind_expr(expr: &QueryExpr) -> anyhow::Result { + match expr { + QueryExpr::PromqlScalarBridge(child) => bind_expr(child), + QueryExpr::Literal(ScalarValue::Float64(value)) => Ok(ExactExpr::Scalar(*value)), + QueryExpr::Literal(ScalarValue::Int64(value)) => Ok(ExactExpr::Scalar(*value as f64)), + QueryExpr::Scan { + source: Source::TimeSeries { metric }, + predicates, + schema, + } => { + let mut matchers = Vec::new(); + for predicate in predicates { + let QueryExpr::Compare { left, op, right } = predicate.0.as_ref() else { + anyhow::bail!("unsupported exact scan predicate") + }; + let (QueryExpr::Column(index), QueryExpr::Literal(ScalarValue::Utf8(value))) = + (left.as_ref(), right.as_ref()) + else { + anyhow::bail!("exact scan requires literal label matchers") + }; + let column = schema + .columns + .get(*index) + .ok_or_else(|| anyhow::anyhow!("invalid matcher column"))?; + let token = match op { + CompareOpKind::Eq => token::T_EQL, + CompareOpKind::Ne => token::T_NEQ, + CompareOpKind::Regex => token::T_EQL_REGEX, + CompareOpKind::NotRegex => token::T_NEQ_REGEX, + _ => anyhow::bail!("unsupported label comparison"), + }; + matchers.push( + Matcher::new_matcher(token, column.name.clone(), value.clone()) + .map_err(anyhow::Error::msg)?, + ); + } + Ok(ExactExpr::Select { + selector: ExactSelector { + metric: metric.clone(), + matchers, + }, + range_seconds: None, + }) + } + QueryExpr::TimeRange { range, child } => { + let mut input = bind_expr(child)?; + let ExactExpr::Select { range_seconds, .. } = &mut input else { + anyhow::bail!("exact range currently requires a raw selector") + }; + anyhow::ensure!( + range_seconds.is_none(), + "nested raw ranges are not supported" + ); + *range_seconds = Some(range.as_secs_f64()); + Ok(input) + } + QueryExpr::Aggregate { + reduction, + measures, + having: None, + child, + .. + } if measures.len() == 1 => { + let intent = &measures[0]; + let input = Box::new(bind_expr(child)?); + match reduction { + Reduction::PerEntity => { + anyhow::ensure!( + matches!( + input.as_ref(), + ExactExpr::Select { + range_seconds: Some(_), + .. + } + ), + "exact rollup needs a raw range input" + ); + let (kernel, parameters) = range_kernel(intent)?; + Ok(ExactExpr::Range { + kernel, + parameters, + input, + }) + } + Reduction::Reduce(keys) => { + let (kernel, parameter, label) = aggregate_kernel(intent)?; + Ok(ExactExpr::Aggregate { + kernel, + parameter, + label, + grouping: grouping(keys, child)?, + input, + }) + } + } + } + QueryExpr::Limit { + n, + offset: 0, + child, + } => { + let QueryExpr::Sort { + keys, + partition_by, + child: input, + } = child.as_ref() + else { + anyhow::bail!("limit requires a bound value sort") + }; + anyhow::ensure!(keys.len() == 1, "exact topk requires one value sort key"); + let QueryExpr::Column(index) = keys[0].expr else { + anyhow::bail!("topk must sort sample values") + }; + anyhow::ensure!( + input + .output_schema()? + .columns + .get(index) + .is_some_and(|c| c.name == "value"), + "topk must sort sample values" + ); + Ok(ExactExpr::Aggregate { + kernel: if keys[0].ascending { + AggregateKernel::BottomK + } else { + AggregateKernel::TopK + }, + parameter: Some(*n as f64), + label: None, + grouping: grouping(partition_by, input)?, + input: Box::new(bind_expr(input)?), + }) + } + QueryExpr::PromqlSeriesSample { by, kind, child } => { + let (kernel, parameter) = match kind { + SampleKind::LimitK(k) => (AggregateKernel::LimitK, *k as f64), + SampleKind::LimitRatio(r) => (AggregateKernel::LimitRatio, *r), + }; + Ok(ExactExpr::Aggregate { + kernel, + parameter: Some(parameter), + label: None, + grouping: grouping(by, child)?, + input: Box::new(bind_expr(child)?), + }) + } + QueryExpr::BinaryOp { + op, + lhs, + rhs, + vector_match, + } => { + if let Some(m) = vector_match { + anyhow::ensure!( + m.kind == VectorMatchKind::Ignoring + && m.labels.is_empty() + && m.grouping.is_none(), + "exact kernel does not implement explicit vector matching" + ); + } + Ok(ExactExpr::Binary { + kernel: op.to_string().to_lowercase().parse()?, + lhs: Box::new(bind_expr(lhs)?), + rhs: Box::new(bind_expr(rhs)?), + }) + } + _ => anyhow::bail!("no executable exact kernel for canonical node {expr:?}"), + } +} + +fn aggregate_kernel( + intent: &AggIntent, +) -> anyhow::Result<(AggregateKernel, Option, Option)> { + use AggregateKernel as K; + let (kernel, parameter, label) = match intent { + AggIntent::Sum { col: None } => (K::Sum, None, None), + AggIntent::Avg { col: None } => (K::Avg, None, None), + AggIntent::Count { .. } => (K::Count, None, None), + AggIntent::Min { col: None } => (K::Min, None, None), + AggIntent::Max { col: None } => (K::Max, None, None), + AggIntent::Group => (K::Group, None, None), + AggIntent::StdDev { + population: true, + col: None, + } => (K::Stddev, None, None), + AggIntent::Variance { + population: true, + col: None, + } => (K::Stdvar, None, None), + AggIntent::Quantile { q, col: None, .. } => (K::Quantile, Some(*q), None), + AggIntent::CountValues { label } => (K::CountValues, None, Some(label.clone())), + _ => anyhow::bail!("no exact aggregation kernel for {intent:?}"), + }; + Ok((kernel, parameter, label)) +} + +fn range_kernel(intent: &AggIntent) -> anyhow::Result<(RangeKernel, Vec)> { + use RangeKernel as K; + Ok(match intent { + AggIntent::Sum { col: None } => (K::Sum, vec![]), + AggIntent::Avg { col: None } => (K::Avg, vec![]), + AggIntent::Count { .. } => (K::Count, vec![]), + AggIntent::Min { col: None } => (K::Min, vec![]), + AggIntent::Max { col: None } => (K::Max, vec![]), + AggIntent::StdDev { + population: true, + col: None, + } => (K::Stddev, vec![]), + AggIntent::Variance { + population: true, + col: None, + } => (K::Stdvar, vec![]), + AggIntent::Quantile { q, col: None, .. } => (K::Quantile, vec![*q]), + AggIntent::LastOverTime => (K::Last, vec![]), + AggIntent::PresentOverTime => (K::Present, vec![]), + AggIntent::AbsentOverTime => (K::Absent, vec![]), + AggIntent::Changes => (K::Changes, vec![]), + AggIntent::Delta => (K::Delta, vec![]), + AggIntent::Deriv => (K::Deriv, vec![]), + AggIntent::IDelta => (K::IDelta, vec![]), + AggIntent::Increase => (K::Increase, vec![]), + AggIntent::IRate => (K::IRate, vec![]), + AggIntent::Rate => (K::Rate, vec![]), + AggIntent::Resets => (K::Resets, vec![]), + AggIntent::PredictLinear { seconds } => (K::PredictLinear, vec![*seconds]), + AggIntent::DoubleExpSmoothing { smoothing, trend } => { + (K::Smoothing, vec![*smoothing, *trend]) + } + AggIntent::MadOverTime => (K::Mad, vec![]), + AggIntent::TsOfMinOverTime => (K::TsMin, vec![]), + AggIntent::TsOfMaxOverTime => (K::TsMax, vec![]), + AggIntent::TsOfLastOverTime => (K::TsLast, vec![]), + _ => anyhow::bail!("no exact range kernel for {intent:?}"), + }) +} diff --git a/data_plane/src/query_engines/canonical/exact_promql.rs b/data_plane/src/query_engines/canonical/exact_promql.rs new file mode 100644 index 00000000..4e81f446 --- /dev/null +++ b/data_plane/src/query_engines/canonical/exact_promql.rs @@ -0,0 +1,579 @@ +//! Execute the control plane's bound exact PromQL kernels over raw float samples. +//! No query is forwarded and no sketch value is substituted for a raw observation. +use std::collections::{BTreeMap, BTreeSet}; + +use control_plane::physical::promql_exact::{ + AggregateKernel as A, BinaryKernel as B, ExactExpr, ExactPromqlPlan, Grouping, RangeKernel as R, +}; +use serde::{Deserialize, Serialize}; + +pub type Labels = BTreeMap; + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct RawSeries { + pub labels: Labels, + /// Unix seconds, strictly increasing; missing samples are absent from this list. + pub samples: Vec<(f64, f64)>, +} + +#[derive(Debug, Clone)] +pub struct ExactSample { + pub labels: Labels, + pub value: f64, +} + +enum Value { + Scalar(f64), + Vector(Vec), + Matrix { + series: Vec, + seconds: f64, + }, +} +impl Value { + fn vector(self) -> anyhow::Result> { + match self { + Self::Vector(v) => Ok(v), + _ => anyhow::bail!("expected an instant vector"), + } + } +} + +/// Evaluate only the bound program. Raw data must already be a consistent snapshot. +pub fn execute( + plan: &ExactPromqlPlan, + data: &[RawSeries], + evaluation: f64, + lookback: f64, +) -> anyhow::Result> { + anyhow::ensure!( + evaluation.is_finite() && lookback.is_finite() && lookback > 0.0, + "invalid evaluation time/lookback" + ); + let mut seen = BTreeSet::new(); + for series in data { + anyhow::ensure!(seen.insert(&series.labels), "duplicate input label set"); + anyhow::ensure!( + series.samples.iter().all(|(t, _)| t.is_finite()) + && series.samples.windows(2).all(|p| p[0].0 < p[1].0), + "raw samples must have unique increasing finite timestamps" + ); + } + let out = eval(plan.root(), data, evaluation, lookback)?.vector()?; + let mut seen = BTreeSet::new(); + anyhow::ensure!( + out.iter().all(|s| seen.insert(&s.labels)), + "duplicate output label set" + ); + Ok(out) +} + +fn eval(expr: &ExactExpr, data: &[RawSeries], time: f64, lookback: f64) -> anyhow::Result { + match expr { + ExactExpr::Scalar(value) => Ok(Value::Scalar(*value)), + ExactExpr::Select { + selector, + range_seconds, + } => { + let mut selected = Vec::new(); + for series in data { + if series.labels.get("__name__") != Some(&selector.metric) { + continue; + } + if !selector.matchers.iter().all(|m| { + m.is_match(series.labels.get(&m.name).map(String::as_str).unwrap_or("")) + }) { + continue; + } + let seconds = range_seconds.unwrap_or(lookback); + let mut samples: Vec<_> = series + .samples + .iter() + .copied() + .filter(|(t, _)| *t > time - seconds && *t <= time) + .collect(); + if range_seconds.is_none() && !samples.is_empty() { + samples = vec![*samples.last().unwrap()]; + } + if !samples.is_empty() { + selected.push(RawSeries { + labels: series.labels.clone(), + samples, + }); + } + } + selected.sort_by(|a, b| a.labels.cmp(&b.labels)); + Ok(match range_seconds { + Some(seconds) => Value::Matrix { + series: selected, + seconds: *seconds, + }, + None => Value::Vector( + selected + .into_iter() + .map(|s| ExactSample { + labels: s.labels, + value: s.samples[0].1, + }) + .collect(), + ), + }) + } + ExactExpr::Aggregate { + kernel, + parameter, + label, + grouping, + input, + } => { + let rows = eval(input, data, time, lookback)?.vector()?; + Ok(Value::Vector(aggregate( + *kernel, + *parameter, + label.as_deref(), + grouping, + rows, + )?)) + } + ExactExpr::Range { + kernel, + parameters, + input, + } => { + let Value::Matrix { series, seconds } = eval(input, data, time, lookback)? else { + anyhow::bail!("range kernel needs raw range samples") + }; + if *kernel == R::Absent { + let labels = absent_labels(input); + return Ok(Value::Vector(if series.is_empty() { + vec![ExactSample { labels, value: 1.0 }] + } else { + vec![] + })); + } + let mut out = Vec::new(); + for series in series { + if let Some(value) = rollup(*kernel, parameters, &series.samples, time, seconds)? { + let mut labels = series.labels; + if *kernel != R::Last { + labels.remove("__name__"); + } + out.push(ExactSample { labels, value }); + } + } + Ok(Value::Vector(out)) + } + ExactExpr::Binary { kernel, lhs, rhs } => binary( + *kernel, + eval(lhs, data, time, lookback)?, + eval(rhs, data, time, lookback)?, + ), + } +} + +fn absent_labels(input: &ExactExpr) -> Labels { + let mut labels = Labels::new(); + if let ExactExpr::Select { selector, .. } = input { + // Derive a label only when it has a single equality matcher. + let mut counts = BTreeMap::new(); + for m in &selector.matchers { + *counts.entry(&m.name).or_insert(0) += 1; + } + for m in &selector.matchers { + if m.name != "__name__" + && m.op.to_string() == "=" + && counts[&m.name] == 1 + && !m.value.is_empty() + { + labels.insert(m.name.clone(), m.value.clone()); + } + } + } + labels +} + +fn group_key(labels: &Labels, grouping: &Grouping) -> Labels { + labels + .iter() + .filter(|(key, _)| { + let included = grouping.labels.contains(key); + if grouping.without { + key.as_str() != "__name__" && !included + } else { + included + } + }) + .map(|(k, v)| (k.clone(), v.clone())) + .collect() +} + +fn aggregate( + kernel: A, + parameter: Option, + label: Option<&str>, + grouping: &Grouping, + rows: Vec, +) -> anyhow::Result> { + let mut groups: BTreeMap> = BTreeMap::new(); + for mut row in rows { + let mut key = group_key(&row.labels, grouping); + if kernel == A::CountValues { + let label = label.ok_or_else(|| anyhow::anyhow!("count_values requires a label"))?; + // The output value label participates in grouping even with no 'by'. + row.labels.insert(label.into(), float_label(row.value)); + key.insert(label.into(), float_label(row.value)); + } + groups.entry(key).or_default().push(row); + } + let mut out = Vec::new(); + for (labels, mut rows) in groups { + match kernel { + A::TopK | A::BottomK | A::LimitK => { + let k = parameter.unwrap_or(0.0); + anyhow::ensure!(k.is_finite() && k >= 0.0, "invalid selection count"); + if kernel != A::LimitK { + rows.sort_by(|a, b| { + if a.value.is_nan() { + return if b.value.is_nan() { + std::cmp::Ordering::Equal + } else { + std::cmp::Ordering::Greater + }; + } + if b.value.is_nan() { + return std::cmp::Ordering::Less; + } + if kernel == A::TopK { + b.value.total_cmp(&a.value) + } else { + a.value.total_cmp(&b.value) + } + }); + } + out.extend(rows.into_iter().take(k as usize)); + } + A::LimitRatio => { + let ratio = parameter.unwrap_or(0.0).clamp(-1.0, 1.0); + anyhow::ensure!(ratio.is_finite(), "invalid sampling ratio"); + out.extend(rows.into_iter().filter(|row| { + let mut bytes = Vec::new(); + for (key, value) in &row.labels { + bytes.extend(key.as_bytes()); + bytes.push(255); + bytes.extend(value.as_bytes()); + bytes.push(255); + } + let offset = xxhash_rust::xxh64::xxh64(&bytes, 0) as f64 / u64::MAX as f64; + if ratio < 0.0 { + offset >= 1.0 + ratio + } else { + offset < ratio + } + })); + } + _ => { + let values: Vec<_> = rows.iter().map(|s| s.value).collect(); + let value = match kernel { + A::Sum => values.iter().sum(), + A::Avg => mean(&values), + A::Count | A::CountValues => values.len() as f64, + A::Min => minimum(&values), + A::Max => maximum(&values), + A::Group => 1.0, + A::Stdvar => variance(&values), + A::Stddev => variance(&values).sqrt(), + A::Quantile => quantile(&values, parameter.unwrap_or(f64::NAN)), + _ => unreachable!(), + }; + out.push(ExactSample { labels, value }); + } + } + } + Ok(out) +} + +fn float_label(value: f64) -> String { + if value.is_nan() { + "NaN".into() + } else if value == f64::INFINITY { + "+Inf".into() + } else if value == f64::NEG_INFINITY { + "-Inf".into() + } else { + value.to_string() + } +} +fn minimum(values: &[f64]) -> f64 { + values.iter().copied().reduce(f64::min).unwrap_or(f64::NAN) +} +fn maximum(values: &[f64]) -> f64 { + values.iter().copied().reduce(f64::max).unwrap_or(f64::NAN) +} +fn mean(values: &[f64]) -> f64 { + values.iter().sum::() / values.len() as f64 +} +fn variance(values: &[f64]) -> f64 { + let mut mean = 0.0; + let mut m2 = 0.0; + for (i, value) in values.iter().enumerate() { + let d = value - mean; + mean += d / (i + 1) as f64; + m2 += d * (value - mean); + } + m2 / values.len() as f64 +} +fn quantile(values: &[f64], phi: f64) -> f64 { + if phi.is_nan() || values.is_empty() { + return f64::NAN; + } + if phi < 0.0 { + return f64::NEG_INFINITY; + } + if phi > 1.0 { + return f64::INFINITY; + } + let mut sorted = values.to_vec(); + sorted.sort_by(|a, b| match (a.is_nan(), b.is_nan()) { + (true, false) => std::cmp::Ordering::Less, + (false, true) => std::cmp::Ordering::Greater, + _ => a.total_cmp(b), + }); + let rank = phi * (sorted.len() - 1) as f64; + let lower = rank.floor() as usize; + let upper = rank.ceil() as usize; + let weight = rank - lower as f64; + sorted[lower] * (1.0 - weight) + sorted[upper] * weight +} + +fn rollup( + kernel: R, + parameters: &[f64], + samples: &[(f64, f64)], + time: f64, + seconds: f64, +) -> anyhow::Result> { + if samples.is_empty() { + return Ok(None); + } + let values: Vec<_> = samples.iter().map(|s| s.1).collect(); + let first = samples[0]; + let last = *samples.last().unwrap(); + let needs_two = matches!( + kernel, + R::Delta + | R::Deriv + | R::IDelta + | R::Increase + | R::IRate + | R::PredictLinear + | R::Rate + | R::Smoothing + ); + if needs_two && samples.len() < 2 { + return Ok(None); + } + Ok(Some(match kernel { + R::Avg => mean(&values), + R::Min => minimum(&values), + R::Max => maximum(&values), + R::Sum => values.iter().sum(), + R::Count => values.len() as f64, + R::Quantile => quantile(&values, parameters[0]), + R::Stddev => variance(&values).sqrt(), + R::Stdvar => variance(&values), + R::Last => last.1, + R::Present => 1.0, + R::Changes => values + .windows(2) + .filter(|p| p[0] != p[1] && !(p[0].is_nan() && p[1].is_nan())) + .count() as f64, + R::Resets => values.windows(2).filter(|p| p[1] < p[0]).count() as f64, + R::IDelta => last.1 - samples[samples.len() - 2].1, + R::IRate => { + let previous = samples[samples.len() - 2]; + let delta = if last.1 < previous.1 { + last.1 + } else { + last.1 - previous.1 + }; + delta / (last.0 - previous.0) + } + R::Rate | R::Increase | R::Delta => { + let counter = kernel != R::Delta; + let mut difference = last.1 - first.1; + if counter { + for pair in values.windows(2) { + if pair[1] < pair[0] { + difference += pair[0]; + } + } + } + let observed = last.0 - first.0; + let interval = observed / (samples.len() - 1) as f64; + let mut before = first.0 - (time - seconds); + let mut after = time - last.0; + if before >= interval * 1.1 { + before = interval / 2.0; + } + if after >= interval * 1.1 { + after = interval / 2.0; + } + if counter && difference > 0.0 && first.1 >= 0.0 { + before = before.min(observed * first.1 / difference); + } + difference * ((observed + before + after) / observed) + / if kernel == R::Rate { seconds } else { 1.0 } + } + R::Deriv | R::PredictLinear => { + // Center timestamps near the window to avoid losing precision on Unix time. + let xs: Vec<_> = samples.iter().map(|s| s.0 - time).collect(); + let mx = mean(&xs); + let my = mean(&values); + let slope = xs + .iter() + .zip(&values) + .map(|(x, y)| (x - mx) * (y - my)) + .sum::() + / xs.iter().map(|x| (x - mx).powi(2)).sum::(); + if kernel == R::Deriv { + slope + } else { + my + slope * (parameters[0] - mx) + } + } + R::Smoothing => { + let (sf, tf) = (parameters[0], parameters[1]); + anyhow::ensure!( + sf > 0.0 && sf < 1.0 && tf > 0.0 && tf < 1.0, + "invalid smoothing/trend factor" + ); + let mut level = first.1; + let mut previous = 0.0; + let mut trend = values[1] - values[0]; + for (i, value) in values.iter().enumerate().skip(1) { + if i > 1 { + trend = tf * (level - previous) + (1.0 - tf) * trend; + } + previous = level; + level = sf * value + (1.0 - sf) * (level + trend); + } + level + } + R::Mad => { + let median = quantile(&values, 0.5); + let deviations: Vec<_> = values.iter().map(|v| (v - median).abs()).collect(); + quantile(&deviations, 0.5) + } + R::TsLast => last.0, + R::TsMin | R::TsMax => { + let mut selected = first; + for sample in samples.iter().copied().skip(1) { + if selected.1.is_nan() + || (kernel == R::TsMin && sample.1 <= selected.1) + || (kernel == R::TsMax && sample.1 >= selected.1) + { + selected = sample; + } + } + selected.0 + } + R::Absent => unreachable!(), + })) +} + +fn arithmetic(kernel: B, a: f64, b: f64) -> anyhow::Result { + Ok(match kernel { + B::Add => a + b, + B::Sub => a - b, + B::Mul => a * b, + B::Div => a / b, + B::Mod => a % b, + B::Pow => a.powf(b), + _ => anyhow::bail!("set operator needs two vectors"), + }) +} +fn matching_key(labels: &Labels) -> Labels { + let mut key = labels.clone(); + key.remove("__name__"); + key +} +fn binary(kernel: B, lhs: Value, rhs: Value) -> anyhow::Result { + let vector = match (lhs, rhs) { + (Value::Scalar(a), Value::Scalar(b)) => { + return Ok(Value::Scalar(arithmetic(kernel, a, b)?)) + } + (Value::Vector(rows), Value::Scalar(scalar)) + | (Value::Scalar(scalar), Value::Vector(rows)) + if matches!(kernel, B::Add | B::Mul) => + { + rows.into_iter() + .map(|mut row| { + row.value = arithmetic(kernel, row.value, scalar)?; + row.labels.remove("__name__"); + Ok(row) + }) + .collect::>>()? + } + (Value::Vector(rows), Value::Scalar(scalar)) => rows + .into_iter() + .map(|mut row| { + row.value = arithmetic(kernel, row.value, scalar)?; + row.labels.remove("__name__"); + Ok(row) + }) + .collect::>>()?, + (Value::Scalar(scalar), Value::Vector(rows)) => rows + .into_iter() + .map(|mut row| { + row.value = arithmetic(kernel, scalar, row.value)?; + row.labels.remove("__name__"); + Ok(row) + }) + .collect::>>()?, + (Value::Vector(left), Value::Vector(right)) => { + let left_keys: BTreeSet<_> = left.iter().map(|s| matching_key(&s.labels)).collect(); + let right_keys: BTreeSet<_> = right.iter().map(|s| matching_key(&s.labels)).collect(); + match kernel { + B::Or => left + .into_iter() + .chain( + right + .into_iter() + .filter(|r| !left_keys.contains(&matching_key(&r.labels))), + ) + .collect(), + B::And => left + .into_iter() + .filter(|s| right_keys.contains(&matching_key(&s.labels))) + .collect(), + B::Unless => left + .into_iter() + .filter(|s| !right_keys.contains(&matching_key(&s.labels))) + .collect(), + _ => { + anyhow::ensure!( + left_keys.len() == left.len() && right_keys.len() == right.len(), + "non-unique vector match" + ); + let right: BTreeMap<_, _> = right + .into_iter() + .map(|s| (matching_key(&s.labels), s.value)) + .collect(); + let mut out = Vec::new(); + for row in left { + let key = matching_key(&row.labels); + if let Some(value) = right.get(&key) { + out.push(ExactSample { + labels: key, + value: arithmetic(kernel, row.value, *value)?, + }); + } + } + out + } + } + } + _ => anyhow::bail!("binary operator cannot consume a range vector"), + }; + Ok(Value::Vector(vector)) +} diff --git a/data_plane/src/query_engines/canonical/mod.rs b/data_plane/src/query_engines/canonical/mod.rs index c170e5d0..67fba694 100644 --- a/data_plane/src/query_engines/canonical/mod.rs +++ b/data_plane/src/query_engines/canonical/mod.rs @@ -20,3 +20,5 @@ pub mod result { pub mod sds_resolver { pub use asap_types::sds::{DataDescriptorId, SummaryDescriptorId}; } + +pub mod exact_promql; From 269143dea9c414d0f7cb6e94b14071a38aba329d Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 21:50:57 -0600 Subject: [PATCH 2/3] test(promql): cover every aggregation and range function with exact smoke cases --- control_plane/examples/promql_smoke.rs | 117 ++ data_plane/examples/promql_exact_smoke.rs | 164 ++ data_plane/tests/promql_exact_execution.rs | 155 ++ tools/promql-smoke/README.md | 157 ++ tools/promql-smoke/RESULTS.md | 57 + tools/promql-smoke/cases.json | 1849 ++++++++++++++++++++ tools/promql-smoke/catalog.json | 174 ++ tools/promql-smoke/run.py | 273 +++ tools/promql-smoke/test_runner.py | 90 + 9 files changed, 3036 insertions(+) create mode 100644 control_plane/examples/promql_smoke.rs create mode 100644 data_plane/examples/promql_exact_smoke.rs create mode 100644 data_plane/tests/promql_exact_execution.rs create mode 100644 tools/promql-smoke/README.md create mode 100644 tools/promql-smoke/RESULTS.md create mode 100644 tools/promql-smoke/cases.json create mode 100644 tools/promql-smoke/catalog.json create mode 100644 tools/promql-smoke/run.py create mode 100644 tools/promql-smoke/test_runner.py diff --git a/control_plane/examples/promql_smoke.rs b/control_plane/examples/promql_smoke.rs new file mode 100644 index 00000000..2976f4ff --- /dev/null +++ b/control_plane/examples/promql_smoke.rs @@ -0,0 +1,117 @@ +//! Run the smoke corpus through the backend's pinned PromQL planner bridge. +//! Binding is a capability diagnostic, not proof of execution or value correctness. +use control_plane::physical::post_asap::{bind_query_expr, PhysicalExpr, PostAsapPlan}; +use control_plane::physical::promql_exact::ExactPromqlPlan; +use control_plane::query_parser::parse_query_expr_canonical; +use planner_types::post_asap::SummaryExpr; +use planner_types::types::AccuracyTarget; +use serde_json::{json, Value}; +use std::collections::BTreeMap; + +fn inspect(query: &Value) -> Value { + let expr = query["expr"].as_str().expect("query expr must be a string"); + let mut row = json!({ + "id": query["id"], "category": query["category"], + "function": query["function"], "experimental": query["experimental"], + "expr": expr, + }); + match parse_query_expr_canonical(expr, AccuracyTarget::Exact) { + Err(error) => { + row["status"] = json!("PARSE_REJECTED"); + row["error"] = json!(error.to_string()); + } + Ok(tree) => { + row["canonical"] = json!(format!("{tree:#?}")); + match bind_query_expr(&tree, AccuracyTarget::Exact) { + Ok(plan) => { + let logical_only = matches!( + &plan, + PhysicalExpr::Committed(PostAsapPlan::Summary(node)) + if matches!(&node.expr, SummaryExpr::KeepPreAsap(_)) + ); + row["status"] = json!(if logical_only { + "LOGICAL_ONLY" + } else { + "BOUND" + }); + row["plan"] = json!(format!("{plan:#?}")); + } + Err(error) => { + row["status"] = json!("BIND_REJECTED"); + row["error"] = json!(error.to_string()); + } + } + } + } + row["summary_status"] = row["status"].take(); + row["summary_plan"] = row["plan"].take(); + row["summary_error"] = row["error"].take(); + match ExactPromqlPlan::bind(expr) { + Ok(plan) => { + row["status"] = json!("BOUND_EXACT"); + row["executor"] = json!("data_plane::query_engines::canonical::exact_promql"); + row["requires"] = json!("raw timestamped float samples"); + row["plan"] = json!(format!("{:#?}", plan.root())); + } + Err(error) => { + row["status"] = json!("EXACT_BIND_REJECTED"); + row["error"] = json!(error.to_string()); + } + } + row +} + +fn main() -> Result<(), Box> { + let args: Vec<_> = std::env::args().skip(1).collect(); + let json_output = args.iter().any(|arg| arg == "--json"); + let require_bound = args.iter().any(|arg| arg == "--require-bound"); + for arg in &args { + if arg.starts_with("--") && arg != "--json" && arg != "--require-bound" { + return Err(format!("unknown option: {arg}").into()); + } + } + let path = args + .iter() + .find(|arg| !arg.starts_with("--")) + .map(String::as_str) + .unwrap_or("tools/promql-smoke/cases.json"); + let cases: Value = serde_json::from_slice(&std::fs::read(path)?)?; + let mut results = Vec::new(); + let mut counts = BTreeMap::::new(); + for query in cases["queries"].as_array().ok_or("missing queries")? { + // A panic is a failed case; continue to expose the other unsupported queries. + let row = std::panic::catch_unwind(|| inspect(query)).unwrap_or_else( + |_| json!({"id": query["id"], "expr": query["expr"], "status": "PANIC"}), + ); + let status = row["status"].as_str().expect("case status"); + *counts.entry(status.to_owned()).or_default() += 1; + if !json_output { + println!( + "{status} {}: {}{}", + query["id"].as_str().unwrap_or("?"), + query["expr"].as_str().unwrap_or("?"), + row["error"] + .as_str() + .map(|e| format!(" — {e}")) + .unwrap_or_default(), + ); + } + results.push(row); + } + let has_unbound = results.iter().any(|row| row["status"] != "BOUND_EXACT"); + if json_output { + println!( + "{}", + serde_json::to_string_pretty(&json!({ + "accuracy": "Exact", "scope": "Canonical tree to native exact kernels; run promql_exact_execution for values", + "summary": counts, "queries": results, + }))? + ); + } else { + println!("Summary: {counts:?}; binding does not prove execution or correct values."); + } + if require_bound && has_unbound { + std::process::exit(1); + } + Ok(()) +} diff --git a/data_plane/examples/promql_exact_smoke.rs b/data_plane/examples/promql_exact_smoke.rs new file mode 100644 index 00000000..0ec221db --- /dev/null +++ b/data_plane/examples/promql_exact_smoke.rs @@ -0,0 +1,164 @@ +//! Serve the small raw fixture using real canonical binding and native exact execution. +//! This is a local test server; it does not alter the production store or routing profile. +use axum::{ + extract::{Query, State}, + http::StatusCode, + routing::get, + Json, Router, +}; +use control_plane::physical::promql_exact::ExactPromqlPlan; +use data_plane::query_engines::canonical::exact_promql::{execute, Labels, RawSeries}; +use serde_json::{json, Value}; +use std::{ + collections::{BTreeMap, HashMap}, + sync::Arc, +}; + +struct Snapshot { + data: Vec, + time: f64, +} +type Reply = (StatusCode, Json); +fn error(message: impl ToString) -> Reply { + ( + StatusCode::BAD_REQUEST, + Json(json!({"status":"error","errorType":"bad_data","error":message.to_string()})), + ) +} +fn timestamp( + params: &HashMap, + key: &str, + default: Option, +) -> anyhow::Result { + let value = match params.get(key) { + Some(value) => value.parse::()?, + None => default.ok_or_else(|| anyhow::anyhow!("missing {key}"))?, + }; + anyhow::ensure!(value.is_finite(), "nonfinite {key}"); + Ok(value) +} +fn sample_value(value: f64) -> String { + if value == f64::INFINITY { + "+Inf".into() + } else if value == f64::NEG_INFINITY { + "-Inf".into() + } else { + value.to_string() + } +} +fn result(data: Value) -> Reply { + ( + StatusCode::OK, + Json( + json!({"status":"success","data":data,"infos":["data_source: asap_exact","accuracy: exact","plan: bound canonical kernels"]}), + ), + ) +} +async fn instant( + State(snapshot): State>, + Query(params): Query>, +) -> Reply { + let run = (|| -> anyhow::Result { + let text = params + .get("query") + .ok_or_else(|| anyhow::anyhow!("missing query"))?; + let plan = ExactPromqlPlan::bind(text)?; + let time = timestamp(¶ms, "time", Some(snapshot.time))?; + let rows = execute(&plan, &snapshot.data, time, 300.0)?; + Ok( + json!({"resultType":"vector","result":rows.into_iter().map(|s|json!({"metric":s.labels,"value":[time,sample_value(s.value)]})).collect::>()}), + ) + })(); + match run { + Ok(data) => result(data), + Err(e) => error(e), + } +} +async fn range( + State(snapshot): State>, + Query(params): Query>, +) -> Reply { + let run = (|| -> anyhow::Result { + let text = params + .get("query") + .ok_or_else(|| anyhow::anyhow!("missing query"))?; + let plan = ExactPromqlPlan::bind(text)?; + let start = timestamp(¶ms, "start", None)?; + let end = timestamp(¶ms, "end", None)?; + let step = timestamp(¶ms, "step", None)?; + anyhow::ensure!(end >= start && step > 0.0, "invalid range bounds or step"); + let steps = ((end - start) / step).floor(); + anyhow::ensure!(steps < 11000.0, "too many evaluation steps"); + let count = steps as usize + 1; + let mut rows: BTreeMap> = BTreeMap::new(); + for i in 0..count { + let time = start + i as f64 * step; + for sample in execute(&plan, &snapshot.data, time, 300.0)? { + rows.entry(sample.labels) + .or_default() + .push(json!([time, sample_value(sample.value)])); + } + } + Ok( + json!({"resultType":"matrix","result":rows.into_iter().map(|(labels,values)|json!({"metric":labels,"values":values})).collect::>()}), + ) + })(); + match run { + Ok(data) => result(data), + Err(e) => error(e), + } +} +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let mut args = std::env::args().skip(1); + let fixture = args + .next() + .unwrap_or_else(|| "tools/promql-smoke/cases.json".into()); + let listen = args.next().unwrap_or_else(|| "127.0.0.1:18081".into()); + anyhow::ensure!( + args.next().is_none(), + "usage: promql_exact_smoke [cases.json] [listen-address]" + ); + let cases: Value = serde_json::from_slice(&std::fs::read(fixture)?)?; + let start = cases["start"] + .as_f64() + .ok_or_else(|| anyhow::anyhow!("missing start"))?; + let interval = cases["interval"] + .as_f64() + .ok_or_else(|| anyhow::anyhow!("missing interval"))?; + let time = start + + cases["eval_offset"] + .as_f64() + .ok_or_else(|| anyhow::anyhow!("missing eval_offset"))?; + let mut data = Vec::new(); + for series in cases["series"] + .as_array() + .ok_or_else(|| anyhow::anyhow!("missing series"))? + { + let labels = serde_json::from_value(series["labels"].clone())?; + let samples = series["values"] + .as_array() + .ok_or_else(|| anyhow::anyhow!("missing values"))? + .iter() + .enumerate() + .filter_map(|(i, v)| v.as_f64().map(|v| (start + i as f64 * interval, v))) + .collect(); + data.push(RawSeries { labels, samples }); + } + let app = Router::new() + .route("/api/v1/query", get(instant)) + .route("/api/v1/query_range", get(range)) + .route("/api/v1/health", get(|| async { "ok" })) + .with_state(Arc::new(Snapshot { data, time })); + let listener = tokio::net::TcpListener::bind(&listen).await?; + eprintln!( + "Native exact PromQL smoke server at http://{}", + listener.local_addr()? + ); + axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = tokio::signal::ctrl_c().await; + }) + .await?; + Ok(()) +} diff --git a/data_plane/tests/promql_exact_execution.rs b/data_plane/tests/promql_exact_execution.rs new file mode 100644 index 00000000..41a89b08 --- /dev/null +++ b/data_plane/tests/promql_exact_execution.rs @@ -0,0 +1,155 @@ +//! The same hand-checked fixture used by official promtool must execute locally. +use control_plane::physical::promql_exact::ExactPromqlPlan; +use data_plane::query_engines::canonical::exact_promql::{execute, Labels, RawSeries}; +use serde_json::Value; + +fn number(value: &Value) -> f64 { + value + .as_f64() + .unwrap_or_else(|| value.as_str().unwrap().parse().unwrap()) +} + +/// Every smoke query must bind to a real kernel and produce the official labels and values. +#[test] +fn all_smoke_queries_bind_and_execute_exactly() { + let cases: Value = + serde_json::from_str(include_str!("../../tools/promql-smoke/cases.json")).unwrap(); + let start = cases["start"].as_f64().unwrap(); + let interval = cases["interval"].as_f64().unwrap(); + let evaluation = start + cases["eval_offset"].as_f64().unwrap(); + let data: Vec<_> = cases["series"] + .as_array() + .unwrap() + .iter() + .map(|series| RawSeries { + labels: serde_json::from_value(series["labels"].clone()).unwrap(), + samples: series["values"] + .as_array() + .unwrap() + .iter() + .enumerate() + .filter_map(|(i, v)| v.as_f64().map(|value| (start + i as f64 * interval, value))) + .collect(), + }) + .collect(); + let mut failures = Vec::new(); + for query in cases["queries"].as_array().unwrap() { + let id = query["id"].as_str().unwrap(); + let expr = query["expr"].as_str().unwrap(); + let result = (|| -> anyhow::Result<()> { + let plan = ExactPromqlPlan::bind(expr)?; + let actual = execute(&plan, &data, evaluation, 300.0)?; + let expected = query["expected"].as_array().unwrap(); + anyhow::ensure!( + actual.len() == expected.len(), + "series count {} != {}", + actual.len(), + expected.len() + ); + for (index, sample) in expected.iter().enumerate() { + let labels: Labels = serde_json::from_value(sample["labels"].clone())?; + let value = number(&sample["value"]) + + if query["value_is_timestamp"] == true { + start + } else { + 0.0 + }; + let got = actual + .iter() + .find(|s| s.labels == labels) + .ok_or_else(|| anyhow::anyhow!("missing labels {labels:?}; got {actual:?}"))?; + let equal = if value.is_nan() { + got.value.is_nan() + } else if value.is_infinite() { + got.value == value + } else { + (got.value - value).abs() <= 1e-12 + 1e-12 * value.abs() + }; + anyhow::ensure!(equal, "{labels:?}: {} != {value}", got.value); + if query["ordered"] == true { + anyhow::ensure!(actual[index].labels == labels, "wrong series order"); + } + } + Ok(()) + })(); + match result { + Ok(()) => println!("PASS {id}"), + Err(error) => failures.push(format!("{id}: {expr}: {error}")), + } + } + assert!( + failures.is_empty(), + "{} queries failed:\n{}", + failures.len(), + failures.join("\n") + ); +} + +/// Unsupported shape modifiers must be rejected during binding, never ignored by execution. +#[test] +fn unsupported_exact_shapes_are_not_bound() { + for query in [ + "sum(smoke_gauge offset 1m)", + "sum(smoke_gauge @ 100)", + "sum_over_time(smoke_gauge[5m:1m])", + "smoke_gauge + on(job) smoke_gauge", + ] { + assert!(ExactPromqlPlan::bind(query).is_err(), "must reject {query}"); + } +} + +/// Duplicate timestamps and duplicate label identities cannot enter the evaluator silently. +#[test] +fn invalid_raw_snapshots_are_rejected() { + let plan = ExactPromqlPlan::bind("sum(smoke_gauge)").unwrap(); + let series = RawSeries { + labels: [("__name__".into(), "smoke_gauge".into())] + .into_iter() + .collect(), + samples: vec![(1.0, 2.0), (1.0, 3.0)], + }; + assert!(execute(&plan, &[series.clone()], 2.0, 300.0).is_err()); + let unique = RawSeries { + samples: vec![(1.0, 2.0)], + ..series + }; + assert!(execute(&plan, &[unique.clone(), unique], 2.0, 300.0).is_err()); +} + +/// Counting two series with equal numeric values must return two, not one distinct value. +#[test] +fn count_preserves_equal_valued_series_multiplicity() { + let plan = ExactPromqlPlan::bind("count(smoke_gauge)").unwrap(); + let data: Vec<_> = ["a", "b"] + .into_iter() + .map(|job| RawSeries { + labels: [ + ("__name__".into(), "smoke_gauge".into()), + ("job".into(), job.into()), + ] + .into_iter() + .collect(), + samples: vec![(240.0, 5.0)], + }) + .collect(); + let result = execute(&plan, &data, 240.0, 300.0).unwrap(); + assert_eq!(result.len(), 1); + assert_eq!(result[0].value, 2.0); +} + +/// Negative ratios select from the upper end: -1 must retain the entire input. +#[test] +fn negative_full_ratio_keeps_every_series() { + let plan = ExactPromqlPlan::bind("limit_ratio(-1, smoke_gauge)").unwrap(); + let data = vec![RawSeries { + labels: [ + ("__name__".into(), "smoke_gauge".into()), + ("job".into(), "a".into()), + ] + .into_iter() + .collect(), + samples: vec![(240.0, 5.0)], + }]; + let result = execute(&plan, &data, 240.0, 300.0).unwrap(); + assert_eq!(result.len(), 1); +} diff --git a/tools/promql-smoke/README.md b/tools/promql-smoke/README.md new file mode 100644 index 00000000..fedd2b9d --- /dev/null +++ b/tools/promql-smoke/README.md @@ -0,0 +1,157 @@ +# PromQL aggregation and rollup smoke tests + +Small fixtures for **Prometheus 3.5.0**: **14 aggregation operators**, **25 range-vector +functions**, **105 queries**, **14 series**, **64 samples**. Every query has explicit +expected labels and values in [cases.json](cases.json), plus a note explaining the +case. `null` means a missing sample, not zero. + +“Rollup” here means every registered Prometheus function accepting a +`ValueTypeMatrix` argument. This is function-name coverage using float samples, +not full PromQL conformance: native histograms, mixed sample types, staleness, +and all combinations of modifiers are not covered. Instant-vector histogram +helpers and scalar/math/label functions are outside this scope. + +The catalog is extracted from the **v3.5.0** official parser registry, with source +URLs and SHA-256 hashes in [catalog.json](catalog.json). Generation fails if any +catalog entry has no case. `--verify-catalog` also downloads the pinned sources +and verifies both their hashes and their registered names against the catalog. + +| Category | Functions | +| --- | --- | +| Aggregations | `sum`, `avg`, `count`, `min`, `max`, `group`, `stddev`, `stdvar`, `topk`, `bottomk`, `count_values`, `quantile` | +| Experimental aggregations | `limitk`, `limit_ratio` | +| Time-window aggregations | `avg_over_time`, `min_over_time`, `max_over_time`, `sum_over_time`, `count_over_time`, `quantile_over_time`, `stddev_over_time`, `stdvar_over_time`, `last_over_time`, `present_over_time` | +| Other range-vector functions | `absent_over_time`, `changes`, `delta`, `deriv`, `idelta`, `increase`, `irate`, `predict_linear`, `rate`, `resets` | +| Experimental range-vector functions | `double_exponential_smoothing`, `mad_over_time`, `ts_of_min_over_time`, `ts_of_max_over_time`, `ts_of_last_over_time` | + +The extra cases cover empty inputs, grouping, repeated values, sparse sampling, +single-sample ranges, counter resets and zero-point extrapolation, left-open +window boundaries, interpolated p99, tied timestamp extrema, and nonfinite values. +The two original gauges remain `1,2,3,4,5` and `10,20,30,40,50`. + +## Run official reference tests + +From the repository root, using the official **3.5.0** promtool binary: + +```bash +python3 tools/promql-smoke/run.py --promtool /path/to/promtool --verify-catalog +``` + +This checks the binary version and runs both suites, enabling +`promql-experimental-functions` only for the experimental suite. Ordinary runs +can omit `--verify-catalog` to work offline. Outputs default to +`/tmp/asap-promql-smoke`; change that with `--output-dir`. + +If using Docker, generate once and run both suites: + +```bash +python3 tools/promql-smoke/run.py --verify-catalog + +docker run --rm -v /tmp/asap-promql-smoke:/tests:ro --entrypoint promtool \ + prom/prometheus:v3.5.0 test rules /tests/rules.test.yml + +docker run --rm -v /tmp/asap-promql-smoke:/tests:ro --entrypoint promtool \ + prom/prometheus:v3.5.0 --enable-feature=promql-experimental-functions \ + test rules /tests/experimental.test.yml +``` + +The Python `--promtool` runner additionally saves per-case JUnit results, logs, +and `reference-results.json`. It returns nonzero on failure. Expected finite +values use promtool's one-bit floating-point tolerance. + +Promtool 3.5 does not consider NaN equal to NaN. That one reference case uses +`x != bool x` to prove the result is NaN; its original expression and raw NaN +expectations are retained for planner and HTTP tests. The override is explicit +in `cases.json`. The infinity cases compare raw values directly. + +## Bind and execute through ASAPPlanner and the backend + +The workspace pins the published planner commit +`f27b16a747e5d7fcd70a5510075c0cd062f0dcea` +([ASAPPlanner PR #413](https://github.com/ProjectASAP/ASAPPlanner/pull/413)). +This commit applies PR #413 on top of the backend’s existing planner pin +`029ff2fe041172c94c2d32c90b185bc83c5e8a57`, preserving its interfaces. +No adjacent planner checkout or local Cargo patch is required. The planner +preserves `irate`, series-count semantics, and special quantile parameters in +the canonical tree. + +```bash +cargo +1.98.0 test --locked -p data_plane --test promql_exact_execution +cargo +1.98.0 run --locked -p control_plane --example promql_smoke -- --require-bound + +# Save canonical trees, exact plans, and the original summary binder diagnostics. +cargo +1.98.0 run --locked -p control_plane --example promql_smoke -- \ + --json --require-bound > /tmp/asap-promql-smoke/planner-results.json +``` + +`ExactPromqlPlan::bind` calls the backend's `parse_query_expr_canonical` with +`AccuracyTarget::Exact`, then compiles that canonical tree into typed native +kernels. `BOUND_EXACT` means an executable exact plan; unsupported shapes are +`EXACT_BIND_REJECTED`. `--require-bound` fails on any rejection. Original sketch +binder diagnostics remain under `summary_status`; they do not determine exact +execution support. + +The Rust execution test evaluates all 105 queries against raw timestamped float +samples and compares full labels, values, and requested ordering with the same +fixture expectations verified by official Prometheus. Additional regression tests +cover invalid snapshots, unsupported modifiers, equal-valued series counts, and +negative sampling ratios. + +## Run the local native HTTP smoke server + +In one terminal: + +```bash +cargo +1.98.0 run --locked -p data_plane --example promql_exact_smoke +``` + +In another: + +```bash +python3 tools/promql-smoke/run.py --backend-url http://127.0.0.1:18081 +``` + +This example loads `cases.json` directly and serves `/api/v1/query` and +`/api/v1/query_range` using the canonical binder and backend exact executor. +It performs no Prometheus forwarding. Responses identify `data_source: asap_exact`. +Optional positional arguments are the fixture path and listening address. +This is a local test entry point; production storage and routing are not wired +into this new raw-sample execution path. Exact plans require raw samples, which +cannot in general be reconstructed from sketches. + +## Compare backend HTTP results + +After loading `samples.openmetrics` through the deployment's ingest path: + +```bash +python3 tools/promql-smoke/run.py --backend-url http://127.0.0.1:8080 +``` + +The runner does **not** load data or install a plan into the backend. It queries +all cases, including experimental ones, and reports unsupported queries as +failures. Fixture timestamps are in seconds, beginning at `1788825600`, sampled +every 60 seconds, and evaluated at `1788825840`. The official promtool suite uses +relative times starting at zero; `ts_of_*` expected values are shifted to epoch +time for HTTP comparisons. + +The comparator checks all labels including metric names, result type, the full +series set, evaluation timestamps, finite values (`rtol=atol=1e-12`), NaN/Inf, +and explicitly requested topk/bottomk ordering. Missing series are not replaced +with zero. Full responses, including provenance annotations, are saved in +`backend-results.json`. A matching response can still be a fallback; inspect its +provenance separately. Approximate sketch answers can fail these exact checks. + +## Validation + +```bash +python3 -m unittest discover -s tools/promql-smoke -p 'test_*.py' -v +``` + +[RESULTS.md](RESULTS.md) records the observed official and planner results from +2026-09-13. It is a snapshot, not a substitute for rerunning after changes. + +Official sources: +[operators](https://prometheus.io/docs/prometheus/3.5/querying/operators/), +[functions](https://prometheus.io/docs/prometheus/3.5/querying/functions/), +[function registry](https://github.com/prometheus/prometheus/blob/v3.5.0/promql/parser/functions.go), +[aggregation registry](https://github.com/prometheus/prometheus/blob/v3.5.0/promql/parser/lex.go). diff --git a/tools/promql-smoke/RESULTS.md b/tools/promql-smoke/RESULTS.md new file mode 100644 index 00000000..3acc44a3 --- /dev/null +++ b/tools/promql-smoke/RESULTS.md @@ -0,0 +1,57 @@ +# Recorded smoke results + +Run date: 2026-09-13. + +- Fixture SHA-256: `7df76b148e7b1d4c11260f23679d5138d1bb6c14ededf9ff8d1d9415f8146eeb`. +- Reference: official Prometheus/promtool 3.5.0 (`8be3a9560fbdd18a94dedec4b747c35178177202`). +- Backend PR base: `8cf1890b` on `origin/main`. +- Planner dependency: published commit `f27b16a747e5d7fcd70a5510075c0cd062f0dcea` + (the fix from [PR #413](https://github.com/ProjectASAP/ASAPPlanner/pull/413) + applied to the existing backend planner pin); no local path patch. + +| Check | Result | +| --- | --- | +| Official promtool | 84 stable + 21 experimental cases passed | +| Official Prometheus HTTP against isolated fixture TSDB | 105/105 passed | +| Canonical tree → exact kernel binding | 105/105 `BOUND_EXACT` | +| Backend native exact execution against official-verified expectations | 105/105 passed | +| Local native backend HTTP `/api/v1/query` | 105/105 passed; every response identifies `asap_exact` | +| Local native HTTP range consistency | 105/105 range queries match per-step instant results | +| Native executor regression tests | 5/5 passed, including the 105-case corpus | +| Planner frontend regression/conformance/lowering/equivalence tests | 162/162 passed | +| Python comparator/coverage tests | 8/8 passed | + +All 14 aggregation operators and 25 range-vector functions in the pinned catalog +have an executable exact binding. This measures float-sample function coverage, +not full PromQL conformance. Native histograms, mixed types, staleness, offsets, +`@`, subqueries, and explicit vector matching are outside this exact smoke path; +unsupported query shapes are rejected rather than silently stripped. + +The exact plan consumes the real ASAPPlanner canonical tree. Execution uses native +backend kernels and timestamped raw samples; it does not forward queries to +Prometheus or read expected results from the fixture. The HTTP check uses the +`promql_exact_smoke` example, not the production ingestion/storage/router path. +Production use still needs a raw-sample source and routing integration. + +## Semantic behavior exercised + +- Existing upstream `irate` retains a distinct canonical intent from `rate`. +- Existing upstream `count` counts series, including equal-valued series, instead of distinct numbers. +- Quantile phi outside [0,1] and NaN survives lowering and produces the defined + `-Inf`, `+Inf`, or `NaN` result in the smoke cases. +- Negative `limit_ratio` uses the upper hash interval; `-1` keeps every series. + +The original summary/sketch binder diagnostics remain in `summary_status` in the +planner JSON report. Exact bindings do not imply these functions can execute from +existing sketches alone. + +Original smoke logs are under `/tmp/asap-promql-smoke/`, including +`reference-results.json`, `planner-results.json`, `backend-results.json`, +`native-exact-results.log`, `planner-regressions-after.log`, +`planner-types-mapping.log`, and `http-reference/reference-http-results.json`. +These temporary artifacts may be removed; reproduce the checks with [README.md](README.md). + +PR-branch reruns of promtool, canonical binding, native execution, native HTTP +instant/range checks, planner frontend tests, and Python tests are recorded under +`/tmp/promql-pr-smoke/` and `/tmp/promql-pr-*.log`. The official HTTP reference +check was recorded with the identical fixture before the rebase. diff --git a/tools/promql-smoke/cases.json b/tools/promql-smoke/cases.json new file mode 100644 index 00000000..3b13f6ee --- /dev/null +++ b/tools/promql-smoke/cases.json @@ -0,0 +1,1849 @@ +{ + "prometheus_version": "3.5.0", + "start": 1788825600, + "interval": 60, + "eval_offset": 240, + "series": [ + { + "labels": { + "__name__": "smoke_gauge", + "job": "a" + }, + "values": [ + 1, + 2, + 3, + 4, + 5 + ] + }, + { + "labels": { + "__name__": "smoke_gauge", + "job": "b" + }, + "values": [ + 10, + 20, + 30, + 40, + 50 + ] + }, + { + "labels": { + "__name__": "smoke_counter_total", + "job": "steady" + }, + "values": [ + 60, + 120, + 180, + 240, + 300 + ] + }, + { + "labels": { + "__name__": "smoke_counter_total", + "job": "reset" + }, + "values": [ + 60, + 120, + 180, + 30, + 90 + ] + }, + { + "labels": { + "__name__": "smoke_counter_total", + "job": "last_reset" + }, + "values": [ + 60, + 120, + 180, + 240, + 30 + ] + }, + { + "labels": { + "__name__": "smoke_zero_counter_total", + "job": "zero" + }, + "values": [ + 0, + 60, + 120, + 180, + 240 + ] + }, + { + "labels": { + "__name__": "smoke_constant", + "job": "constant" + }, + "values": [ + 7, + 7, + 7, + 7, + 7 + ] + }, + { + "labels": { + "__name__": "smoke_repeat", + "job": "repeat" + }, + "values": [ + 3, + 1, + 3, + 1, + 2 + ] + }, + { + "labels": { + "__name__": "smoke_sparse", + "job": "sparse" + }, + "values": [ + 1, + null, + 3, + null, + 5 + ] + }, + { + "labels": { + "__name__": "smoke_single", + "job": "single" + }, + "values": [ + null, + null, + null, + null, + 5 + ] + }, + { + "labels": { + "__name__": "smoke_group", + "job": "a", + "instance": "x" + }, + "values": [ + 1, + 1, + 1, + 1, + 1 + ] + }, + { + "labels": { + "__name__": "smoke_group", + "job": "a", + "instance": "y" + }, + "values": [ + 3, + 3, + 3, + 3, + 3 + ] + }, + { + "labels": { + "__name__": "smoke_group", + "job": "b", + "instance": "x" + }, + "values": [ + 2, + 2, + 2, + 2, + 2 + ] + }, + { + "labels": { + "__name__": "smoke_group", + "job": "b", + "instance": "y" + }, + "values": [ + 6, + 6, + 6, + 6, + 6 + ] + } + ], + "queries": [ + { + "id": "selector", + "category": "selector", + "function": "selector", + "experimental": false, + "expr": "smoke_gauge{job=\"a\"}", + "expected": [ + { + "labels": { + "__name__": "smoke_gauge", + "job": "a" + }, + "value": 5 + } + ], + "note": "Baseline label filter." + }, + { + "id": "agg_sum", + "category": "aggregation", + "function": "sum", + "experimental": false, + "expr": "sum(smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 55 + } + ], + "note": "Aggregate the final values 5 and 50." + }, + { + "id": "agg_avg", + "category": "aggregation", + "function": "avg", + "experimental": false, + "expr": "avg(smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 27.5 + } + ], + "note": "Aggregate the final values 5 and 50." + }, + { + "id": "agg_count", + "category": "aggregation", + "function": "count", + "experimental": false, + "expr": "count(smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 2 + } + ], + "note": "Aggregate the final values 5 and 50." + }, + { + "id": "agg_min", + "category": "aggregation", + "function": "min", + "experimental": false, + "expr": "min(smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 5 + } + ], + "note": "Aggregate the final values 5 and 50." + }, + { + "id": "agg_max", + "category": "aggregation", + "function": "max", + "experimental": false, + "expr": "max(smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 50 + } + ], + "note": "Aggregate the final values 5 and 50." + }, + { + "id": "agg_group", + "category": "aggregation", + "function": "group", + "experimental": false, + "expr": "group(smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 1 + } + ], + "note": "Aggregate the final values 5 and 50." + }, + { + "id": "agg_stddev", + "category": "aggregation", + "function": "stddev", + "experimental": false, + "expr": "stddev(smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 22.5 + } + ], + "note": "Aggregate the final values 5 and 50." + }, + { + "id": "agg_stdvar", + "category": "aggregation", + "function": "stdvar", + "experimental": false, + "expr": "stdvar(smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 506.25 + } + ], + "note": "Aggregate the final values 5 and 50." + }, + { + "id": "agg_topk", + "category": "aggregation", + "function": "topk", + "experimental": false, + "expr": "topk(1, smoke_gauge)", + "expected": [ + { + "labels": { + "__name__": "smoke_gauge", + "job": "b" + }, + "value": 50 + } + ], + "note": "Largest series; preserve its labels and metric name." + }, + { + "id": "agg_bottomk", + "category": "aggregation", + "function": "bottomk", + "experimental": false, + "expr": "bottomk(1, smoke_gauge)", + "expected": [ + { + "labels": { + "__name__": "smoke_gauge", + "job": "a" + }, + "value": 5 + } + ], + "note": "Smallest series; preserve its labels and metric name." + }, + { + "id": "agg_count_values", + "category": "aggregation", + "function": "count_values", + "experimental": false, + "expr": "count_values(\"sample\", smoke_gauge % 5)", + "expected": [ + { + "labels": { + "sample": "0" + }, + "value": 2 + } + ], + "note": "Both final values modulo 5 are zero; count repeated values." + }, + { + "id": "agg_quantile", + "category": "aggregation", + "function": "quantile", + "experimental": false, + "expr": "quantile(0.5, smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 27.5 + } + ], + "note": "Median of 5 and 50 is linearly interpolated." + }, + { + "id": "agg_limitk", + "category": "aggregation", + "function": "limitk", + "experimental": true, + "expr": "limitk(1, smoke_gauge{job=\"a\"})", + "expected": [ + { + "labels": { + "__name__": "smoke_gauge", + "job": "a" + }, + "value": 5 + } + ], + "note": "Singleton selection has an unambiguous identity; separate case checks sampling two series." + }, + { + "id": "agg_limit_ratio", + "category": "aggregation", + "function": "limit_ratio", + "experimental": true, + "expr": "limit_ratio(1, smoke_gauge)", + "expected": [ + { + "labels": { + "job": "a", + "__name__": "smoke_gauge" + }, + "value": 5 + }, + { + "labels": { + "job": "b", + "__name__": "smoke_gauge" + }, + "value": 50 + } + ], + "note": "Ratio 1 selects the entire input without changing labels." + }, + { + "id": "rollup_avg_over_time", + "category": "rollup", + "function": "avg_over_time", + "experimental": false, + "expr": "avg_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 3 + }, + { + "labels": { + "job": "b" + }, + "value": 30 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_count_over_time", + "category": "rollup", + "function": "count_over_time", + "experimental": false, + "expr": "count_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 5 + }, + { + "labels": { + "job": "b" + }, + "value": 5 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_last_over_time", + "category": "rollup", + "function": "last_over_time", + "experimental": false, + "expr": "last_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a", + "__name__": "smoke_gauge" + }, + "value": 5 + }, + { + "labels": { + "job": "b", + "__name__": "smoke_gauge" + }, + "value": 50 + } + ], + "note": "last_over_time preserves the metric name, unlike the other numeric rollups." + }, + { + "id": "rollup_max_over_time", + "category": "rollup", + "function": "max_over_time", + "experimental": false, + "expr": "max_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 5 + }, + { + "labels": { + "job": "b" + }, + "value": 50 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_min_over_time", + "category": "rollup", + "function": "min_over_time", + "experimental": false, + "expr": "min_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 1 + }, + { + "labels": { + "job": "b" + }, + "value": 10 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_present_over_time", + "category": "rollup", + "function": "present_over_time", + "experimental": false, + "expr": "present_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 1 + }, + { + "labels": { + "job": "b" + }, + "value": 1 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_stddev_over_time", + "category": "rollup", + "function": "stddev_over_time", + "experimental": false, + "expr": "stddev_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 1.4142135623730951 + }, + { + "labels": { + "job": "b" + }, + "value": 14.142135623730951 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_stdvar_over_time", + "category": "rollup", + "function": "stdvar_over_time", + "experimental": false, + "expr": "stdvar_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 2 + }, + { + "labels": { + "job": "b" + }, + "value": 200 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_sum_over_time", + "category": "rollup", + "function": "sum_over_time", + "experimental": false, + "expr": "sum_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 15 + }, + { + "labels": { + "job": "b" + }, + "value": 150 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_mad_over_time", + "category": "rollup", + "function": "mad_over_time", + "experimental": true, + "expr": "mad_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 1 + }, + { + "labels": { + "job": "b" + }, + "value": 10 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_delta", + "category": "rollup", + "function": "delta", + "experimental": false, + "expr": "delta(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 5 + }, + { + "labels": { + "job": "b" + }, + "value": 50 + } + ], + "note": "Observed difference 4/40 extrapolated from 240s to the 300s window: 5/50." + }, + { + "id": "rollup_idelta", + "category": "rollup", + "function": "idelta", + "experimental": false, + "expr": "idelta(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 1 + }, + { + "labels": { + "job": "b" + }, + "value": 10 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_deriv", + "category": "rollup", + "function": "deriv", + "experimental": false, + "expr": "deriv(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 0.016666666666666666 + }, + { + "labels": { + "job": "b" + }, + "value": 0.16666666666666666 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_changes", + "category": "rollup", + "function": "changes", + "experimental": false, + "expr": "changes(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 4 + }, + { + "labels": { + "job": "b" + }, + "value": 4 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_quantile_over_time", + "category": "rollup", + "function": "quantile_over_time", + "experimental": false, + "expr": "quantile_over_time(0.5, smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 3 + }, + { + "labels": { + "job": "b" + }, + "value": 30 + } + ], + "note": "Evaluate the five samples at t=240s; [5m] includes t=0." + }, + { + "id": "rollup_predict_linear", + "category": "rollup", + "function": "predict_linear", + "experimental": false, + "expr": "predict_linear(smoke_gauge[5m], 60)", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 6 + }, + { + "labels": { + "job": "b" + }, + "value": 60 + } + ], + "note": "Linear trend projected 60 seconds beyond evaluation time." + }, + { + "id": "rollup_double_exponential_smoothing", + "category": "rollup", + "function": "double_exponential_smoothing", + "experimental": true, + "expr": "double_exponential_smoothing(smoke_gauge[5m], 0.5, 0.5)", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 5 + }, + { + "labels": { + "job": "b" + }, + "value": 50 + } + ], + "note": "The input is a perfect linear trend; smoothing follows it exactly." + }, + { + "id": "rollup_rate", + "category": "rollup", + "function": "rate", + "experimental": false, + "expr": "rate(smoke_counter_total[5m])", + "expected": [ + { + "labels": { + "job": "steady" + }, + "value": 1 + }, + { + "labels": { + "job": "reset" + }, + "value": 0.875 + }, + { + "labels": { + "job": "last_reset" + }, + "value": 0.875 + } + ], + "note": "Compare steady growth, an interior reset, and a reset in the last pair. rate/increase extrapolate over 300s." + }, + { + "id": "rollup_increase", + "category": "rollup", + "function": "increase", + "experimental": false, + "expr": "increase(smoke_counter_total[5m])", + "expected": [ + { + "labels": { + "job": "steady" + }, + "value": 300 + }, + { + "labels": { + "job": "reset" + }, + "value": 262.5 + }, + { + "labels": { + "job": "last_reset" + }, + "value": 262.5 + } + ], + "note": "Compare steady growth, an interior reset, and a reset in the last pair. rate/increase extrapolate over 300s." + }, + { + "id": "rollup_irate", + "category": "rollup", + "function": "irate", + "experimental": false, + "expr": "irate(smoke_counter_total[5m])", + "expected": [ + { + "labels": { + "job": "steady" + }, + "value": 1 + }, + { + "labels": { + "job": "reset" + }, + "value": 1 + }, + { + "labels": { + "job": "last_reset" + }, + "value": 0.5 + } + ], + "note": "Compare steady growth, an interior reset, and a reset in the last pair. rate/increase extrapolate over 300s." + }, + { + "id": "rollup_resets", + "category": "rollup", + "function": "resets", + "experimental": false, + "expr": "resets(smoke_counter_total[5m])", + "expected": [ + { + "labels": { + "job": "steady" + }, + "value": 0 + }, + { + "labels": { + "job": "reset" + }, + "value": 1 + }, + { + "labels": { + "job": "last_reset" + }, + "value": 1 + } + ], + "note": "Compare steady growth, an interior reset, and a reset in the last pair. rate/increase extrapolate over 300s." + }, + { + "id": "rollup_absent_over_time", + "category": "rollup", + "function": "absent_over_time", + "experimental": false, + "expr": "absent_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [ + { + "labels": { + "job": "missing" + }, + "value": 1 + } + ], + "note": "Missing range yields 1 and derives equality-matcher labels." + }, + { + "id": "rollup_ts_of_max_over_time", + "category": "rollup", + "function": "ts_of_max_over_time", + "experimental": true, + "expr": "ts_of_max_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 240 + }, + { + "labels": { + "job": "b" + }, + "value": 240 + } + ], + "note": "Sample timestamps are relative to fixture start; HTTP expected values add the epoch start.", + "value_is_timestamp": true + }, + { + "id": "rollup_ts_of_min_over_time", + "category": "rollup", + "function": "ts_of_min_over_time", + "experimental": true, + "expr": "ts_of_min_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 0 + }, + { + "labels": { + "job": "b" + }, + "value": 0 + } + ], + "note": "Sample timestamps are relative to fixture start; HTTP expected values add the epoch start.", + "value_is_timestamp": true + }, + { + "id": "rollup_ts_of_last_over_time", + "category": "rollup", + "function": "ts_of_last_over_time", + "experimental": true, + "expr": "ts_of_last_over_time(smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 240 + }, + { + "labels": { + "job": "b" + }, + "value": 240 + } + ], + "note": "Sample timestamps are relative to fixture start; HTTP expected values add the epoch start.", + "value_is_timestamp": true + }, + { + "id": "empty_avg_over_time", + "category": "rollup", + "function": "avg_over_time", + "experimental": false, + "expr": "avg_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_count_over_time", + "category": "rollup", + "function": "count_over_time", + "experimental": false, + "expr": "count_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_last_over_time", + "category": "rollup", + "function": "last_over_time", + "experimental": false, + "expr": "last_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_max_over_time", + "category": "rollup", + "function": "max_over_time", + "experimental": false, + "expr": "max_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_min_over_time", + "category": "rollup", + "function": "min_over_time", + "experimental": false, + "expr": "min_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_present_over_time", + "category": "rollup", + "function": "present_over_time", + "experimental": false, + "expr": "present_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_stddev_over_time", + "category": "rollup", + "function": "stddev_over_time", + "experimental": false, + "expr": "stddev_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_stdvar_over_time", + "category": "rollup", + "function": "stdvar_over_time", + "experimental": false, + "expr": "stdvar_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_sum_over_time", + "category": "rollup", + "function": "sum_over_time", + "experimental": false, + "expr": "sum_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_mad_over_time", + "category": "rollup", + "function": "mad_over_time", + "experimental": true, + "expr": "mad_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_delta", + "category": "rollup", + "function": "delta", + "experimental": false, + "expr": "delta(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_idelta", + "category": "rollup", + "function": "idelta", + "experimental": false, + "expr": "idelta(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_deriv", + "category": "rollup", + "function": "deriv", + "experimental": false, + "expr": "deriv(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_changes", + "category": "rollup", + "function": "changes", + "experimental": false, + "expr": "changes(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_quantile_over_time", + "category": "rollup", + "function": "quantile_over_time", + "experimental": false, + "expr": "quantile_over_time(0.5, smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_predict_linear", + "category": "rollup", + "function": "predict_linear", + "experimental": false, + "expr": "predict_linear(smoke_missing{job=\"missing\"}[5m], 60)", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_double_exponential_smoothing", + "category": "rollup", + "function": "double_exponential_smoothing", + "experimental": true, + "expr": "double_exponential_smoothing(smoke_missing{job=\"missing\"}[5m], 0.5, 0.5)", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_rate", + "category": "rollup", + "function": "rate", + "experimental": false, + "expr": "rate(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_increase", + "category": "rollup", + "function": "increase", + "experimental": false, + "expr": "increase(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_irate", + "category": "rollup", + "function": "irate", + "experimental": false, + "expr": "irate(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_resets", + "category": "rollup", + "function": "resets", + "experimental": false, + "expr": "resets(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero." + }, + { + "id": "empty_absent_over_time", + "category": "rollup", + "function": "absent_over_time", + "experimental": false, + "expr": "absent_over_time(smoke_gauge[5m])", + "expected": [], + "note": "An existing range must not trigger absence." + }, + { + "id": "empty_ts_of_max_over_time", + "category": "rollup", + "function": "ts_of_max_over_time", + "experimental": true, + "expr": "ts_of_max_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero.", + "value_is_timestamp": true + }, + { + "id": "empty_ts_of_min_over_time", + "category": "rollup", + "function": "ts_of_min_over_time", + "experimental": true, + "expr": "ts_of_min_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero.", + "value_is_timestamp": true + }, + { + "id": "empty_ts_of_last_over_time", + "category": "rollup", + "function": "ts_of_last_over_time", + "experimental": true, + "expr": "ts_of_last_over_time(smoke_missing{job=\"missing\"}[5m])", + "expected": [], + "note": "An empty input must produce no series, not a fabricated zero.", + "value_is_timestamp": true + }, + { + "id": "empty_sum", + "category": "aggregation", + "function": "sum", + "experimental": false, + "expr": "sum(smoke_missing)", + "expected": [], + "note": "An aggregate over no series yields an empty vector." + }, + { + "id": "empty_count", + "category": "aggregation", + "function": "count", + "experimental": false, + "expr": "count(smoke_missing)", + "expected": [], + "note": "An aggregate over no series yields an empty vector." + }, + { + "id": "empty_group", + "category": "aggregation", + "function": "group", + "experimental": false, + "expr": "group(smoke_missing)", + "expected": [], + "note": "An aggregate over no series yields an empty vector." + }, + { + "id": "empty_topk", + "category": "aggregation", + "function": "topk", + "experimental": false, + "expr": "topk(1, smoke_missing)", + "expected": [], + "note": "An aggregate over no series yields an empty vector." + }, + { + "id": "empty_quantile", + "category": "aggregation", + "function": "quantile", + "experimental": false, + "expr": "quantile(0.5, smoke_missing)", + "expected": [], + "note": "An aggregate over no series yields an empty vector." + }, + { + "id": "empty_limitk", + "category": "aggregation", + "function": "limitk", + "experimental": true, + "expr": "limitk(1, smoke_missing)", + "expected": [], + "note": "An aggregate over no series yields an empty vector." + }, + { + "id": "empty_limit_ratio", + "category": "aggregation", + "function": "limit_ratio", + "experimental": true, + "expr": "limit_ratio(1, smoke_missing)", + "expected": [], + "note": "An aggregate over no series yields an empty vector." + }, + { + "id": "sum_by", + "category": "aggregation", + "function": "sum", + "experimental": false, + "expr": "sum by (job) (smoke_group)", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 4 + }, + { + "labels": { + "job": "b" + }, + "value": 8 + } + ], + "note": "Group four series into two jobs." + }, + { + "id": "sum_without", + "category": "aggregation", + "function": "sum", + "experimental": false, + "expr": "sum without (instance) (smoke_group)", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 4 + }, + { + "labels": { + "job": "b" + }, + "value": 8 + } + ], + "note": "Drop instance and metric name from output labels." + }, + { + "id": "avg_by", + "category": "aggregation", + "function": "avg", + "experimental": false, + "expr": "avg by (job) (smoke_group)", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 2 + }, + { + "labels": { + "job": "b" + }, + "value": 4 + } + ], + "note": "Average separately within each job." + }, + { + "id": "count_by", + "category": "aggregation", + "function": "count", + "experimental": false, + "expr": "count by (job) (smoke_group)", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 2 + }, + { + "labels": { + "job": "b" + }, + "value": 2 + } + ], + "note": "Count series separately within each job." + }, + { + "id": "quantile_p99", + "category": "aggregation", + "function": "quantile", + "experimental": false, + "expr": "quantile(0.99, smoke_gauge)", + "expected": [ + { + "labels": {}, + "value": 49.55 + } + ], + "note": "Interpolate between 5 and 50." + }, + { + "id": "limitk_count", + "category": "aggregation", + "function": "limitk", + "experimental": true, + "expr": "count(limitk(1, smoke_gauge))", + "expected": [ + { + "labels": {}, + "value": 1 + } + ], + "note": "Select exactly one of two series without depending on a hash-selected identity." + }, + { + "id": "limit_ratio_complement", + "category": "aggregation", + "function": "limit_ratio", + "experimental": true, + "expr": "sum(limit_ratio(0.5, smoke_gauge) or limit_ratio(-0.5, smoke_gauge))", + "expected": [ + { + "labels": {}, + "value": 55 + } + ], + "note": "Positive and negative ratios cover the full input." + }, + { + "id": "limit_ratio_disjoint", + "category": "aggregation", + "function": "limit_ratio", + "experimental": true, + "expr": "limit_ratio(0.5, smoke_gauge) and limit_ratio(-0.5, smoke_gauge)", + "expected": [], + "note": "Complementary subsets must not overlap." + }, + { + "id": "p99", + "category": "rollup", + "function": "quantile_over_time", + "experimental": false, + "expr": "quantile_over_time(0.99, smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 4.96 + }, + { + "labels": { + "job": "b" + }, + "value": 49.6 + } + ], + "note": "Small-sample p99 requires interpolation." + }, + { + "id": "window_left_open", + "category": "rollup", + "function": "sum_over_time", + "experimental": false, + "expr": "sum_over_time(smoke_gauge[4m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 14 + }, + { + "labels": { + "job": "b" + }, + "value": 140 + } + ], + "note": "The sample at t=0 is exactly on the left boundary and is excluded." + }, + { + "id": "window_single", + "category": "rollup", + "function": "count_over_time", + "experimental": false, + "expr": "count_over_time(smoke_gauge[1m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": 1 + }, + { + "labels": { + "job": "b" + }, + "value": 1 + } + ], + "note": "Only t=240 is included; t=180 is excluded." + }, + { + "id": "sparse_average", + "category": "rollup", + "function": "avg_over_time", + "experimental": false, + "expr": "avg_over_time(smoke_sparse[5m])", + "expected": [ + { + "labels": { + "job": "sparse" + }, + "value": 3 + } + ], + "note": "Missing samples are omitted, not zero-filled." + }, + { + "id": "sparse_count", + "category": "rollup", + "function": "count_over_time", + "experimental": false, + "expr": "count_over_time(smoke_sparse[5m])", + "expected": [ + { + "labels": { + "job": "sparse" + }, + "value": 3 + } + ], + "note": "Count only the three present samples." + }, + { + "id": "constant_changes", + "category": "rollup", + "function": "changes", + "experimental": false, + "expr": "changes(smoke_constant[5m])", + "expected": [ + { + "labels": { + "job": "constant" + }, + "value": 0 + } + ], + "note": "Equal adjacent values do not count as changes." + }, + { + "id": "constant_stddev", + "category": "rollup", + "function": "stddev_over_time", + "experimental": false, + "expr": "stddev_over_time(smoke_constant[5m])", + "expected": [ + { + "labels": { + "job": "constant" + }, + "value": 0 + } + ], + "note": "A constant series has zero standard deviation." + }, + { + "id": "zero_rate", + "category": "rollup", + "function": "rate", + "experimental": false, + "expr": "rate(smoke_zero_counter_total[5m])", + "expected": [ + { + "labels": { + "job": "zero" + }, + "value": 0.8 + } + ], + "note": "Counter starts at zero: extrapolation must not invent a negative prior counter." + }, + { + "id": "zero_increase", + "category": "rollup", + "function": "increase", + "experimental": false, + "expr": "increase(smoke_zero_counter_total[5m])", + "expected": [ + { + "labels": { + "job": "zero" + }, + "value": 240 + } + ], + "note": "Zero-point clamping limits the extrapolated increase to 240." + }, + { + "id": "single_delta", + "category": "rollup", + "function": "delta", + "experimental": false, + "expr": "delta(smoke_single[5m])", + "expected": [], + "note": "A single sample is insufficient; return no series." + }, + { + "id": "single_deriv", + "category": "rollup", + "function": "deriv", + "experimental": false, + "expr": "deriv(smoke_single[5m])", + "expected": [], + "note": "A single sample is insufficient; return no series." + }, + { + "id": "single_idelta", + "category": "rollup", + "function": "idelta", + "experimental": false, + "expr": "idelta(smoke_single[5m])", + "expected": [], + "note": "A single sample is insufficient; return no series." + }, + { + "id": "single_rate", + "category": "rollup", + "function": "rate", + "experimental": false, + "expr": "rate(smoke_single[5m])", + "expected": [], + "note": "A single sample is insufficient; return no series." + }, + { + "id": "single_irate", + "category": "rollup", + "function": "irate", + "experimental": false, + "expr": "irate(smoke_single[5m])", + "expected": [], + "note": "A single sample is insufficient; return no series." + }, + { + "id": "single_increase", + "category": "rollup", + "function": "increase", + "experimental": false, + "expr": "increase(smoke_single[5m])", + "expected": [], + "note": "A single sample is insufficient; return no series." + }, + { + "id": "single_predict_linear", + "category": "rollup", + "function": "predict_linear", + "experimental": false, + "expr": "predict_linear(smoke_single[5m], 60)", + "expected": [], + "note": "A single sample is insufficient; return no series." + }, + { + "id": "single_double_exponential_smoothing", + "category": "rollup", + "function": "double_exponential_smoothing", + "experimental": true, + "expr": "double_exponential_smoothing(smoke_single[5m], 0.5, 0.5)", + "expected": [], + "note": "A single sample is insufficient; return no series." + }, + { + "id": "ties_ts_of_max_over_time", + "category": "rollup", + "function": "ts_of_max_over_time", + "experimental": true, + "expr": "ts_of_max_over_time(smoke_repeat[5m])", + "expected": [ + { + "labels": { + "job": "repeat" + }, + "value": 120 + } + ], + "note": "For tied extrema choose the latest sample timestamp.", + "value_is_timestamp": true + }, + { + "id": "ties_ts_of_min_over_time", + "category": "rollup", + "function": "ts_of_min_over_time", + "experimental": true, + "expr": "ts_of_min_over_time(smoke_repeat[5m])", + "expected": [ + { + "labels": { + "job": "repeat" + }, + "value": 180 + } + ], + "note": "For tied extrema choose the latest sample timestamp.", + "value_is_timestamp": true + }, + { + "id": "ties_ts_of_last_over_time", + "category": "rollup", + "function": "ts_of_last_over_time", + "experimental": true, + "expr": "ts_of_last_over_time(smoke_repeat[5m])", + "expected": [ + { + "labels": { + "job": "repeat" + }, + "value": 240 + } + ], + "note": "For tied extrema choose the latest sample timestamp.", + "value_is_timestamp": true + }, + { + "id": "phi_-0.1", + "category": "rollup", + "function": "quantile_over_time", + "experimental": false, + "expr": "quantile_over_time(-0.1, smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": "-Inf" + }, + { + "labels": { + "job": "b" + }, + "value": "-Inf" + } + ], + "note": "Out-of-range quantiles return the corresponding infinity." + }, + { + "id": "phi_1.1", + "category": "rollup", + "function": "quantile_over_time", + "experimental": false, + "expr": "quantile_over_time(1.1, smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": "+Inf" + }, + { + "labels": { + "job": "b" + }, + "value": "+Inf" + } + ], + "note": "Out-of-range quantiles return the corresponding infinity." + }, + { + "id": "phi_nan", + "category": "rollup", + "function": "quantile_over_time", + "experimental": false, + "expr": "quantile_over_time(NaN, smoke_gauge[5m])", + "expected": [ + { + "labels": { + "job": "a" + }, + "value": "NaN" + }, + { + "labels": { + "job": "b" + }, + "value": "NaN" + } + ], + "note": "NaN quantile parameter returns NaN. promtool 3.5 cannot compare NaN expectations; the reference expression checks x != x with bool, which is true only for NaN. HTTP comparison checks raw NaN directly.", + "promtool_expr": "(quantile_over_time(NaN, smoke_gauge[5m])) != bool (quantile_over_time(NaN, smoke_gauge[5m]))", + "promtool_expected": [ + { + "labels": { + "job": "a" + }, + "value": 1 + }, + { + "labels": { + "job": "b" + }, + "value": 1 + } + ] + }, + { + "id": "topk_order", + "category": "aggregation", + "function": "topk", + "experimental": false, + "expr": "topk(2, smoke_gauge)", + "expected": [ + { + "labels": { + "__name__": "smoke_gauge", + "job": "b" + }, + "value": 50 + }, + { + "labels": { + "__name__": "smoke_gauge", + "job": "a" + }, + "value": 5 + } + ], + "note": "Instant topk returns descending values; HTTP comparator also verifies order.", + "ordered": true + }, + { + "id": "bottomk_order", + "category": "aggregation", + "function": "bottomk", + "experimental": false, + "expr": "bottomk(2, smoke_gauge)", + "expected": [ + { + "labels": { + "job": "a", + "__name__": "smoke_gauge" + }, + "value": 5 + }, + { + "labels": { + "job": "b", + "__name__": "smoke_gauge" + }, + "value": 50 + } + ], + "note": "Instant bottomk returns ascending values; HTTP comparator also verifies order.", + "ordered": true + } + ] +} diff --git a/tools/promql-smoke/catalog.json b/tools/promql-smoke/catalog.json new file mode 100644 index 00000000..2437d66d --- /dev/null +++ b/tools/promql-smoke/catalog.json @@ -0,0 +1,174 @@ +{ + "prometheus_version": "3.5.0", + "scope": "All aggregation operators and all functions with a ValueTypeMatrix argument; float-sample smoke coverage, not full type/edge-case conformance.", + "sources": { + "functions.go": { + "url": "https://raw.githubusercontent.com/prometheus/prometheus/v3.5.0/promql/parser/functions.go", + "sha256": "11881bfeb3093bea274f16f55a681bb1510a81c9cdc795756ad7256b2ef514cf" + }, + "lex.go": { + "url": "https://raw.githubusercontent.com/prometheus/prometheus/v3.5.0/promql/parser/lex.go", + "sha256": "a8d67c46cf53a10b2b3721d58a7734868f67f0467bbe6469e59bd97c5b1ac5fc" + } + }, + "aggregation": [ + { + "name": "sum", + "experimental": false + }, + { + "name": "avg", + "experimental": false + }, + { + "name": "count", + "experimental": false + }, + { + "name": "min", + "experimental": false + }, + { + "name": "max", + "experimental": false + }, + { + "name": "group", + "experimental": false + }, + { + "name": "stddev", + "experimental": false + }, + { + "name": "stdvar", + "experimental": false + }, + { + "name": "topk", + "experimental": false + }, + { + "name": "bottomk", + "experimental": false + }, + { + "name": "count_values", + "experimental": false + }, + { + "name": "quantile", + "experimental": false + }, + { + "name": "limitk", + "experimental": true + }, + { + "name": "limit_ratio", + "experimental": true + } + ], + "rollup": [ + { + "name": "absent_over_time", + "experimental": false + }, + { + "name": "avg_over_time", + "experimental": false + }, + { + "name": "changes", + "experimental": false + }, + { + "name": "count_over_time", + "experimental": false + }, + { + "name": "delta", + "experimental": false + }, + { + "name": "deriv", + "experimental": false + }, + { + "name": "double_exponential_smoothing", + "experimental": true + }, + { + "name": "idelta", + "experimental": false + }, + { + "name": "increase", + "experimental": false + }, + { + "name": "irate", + "experimental": false + }, + { + "name": "last_over_time", + "experimental": false + }, + { + "name": "mad_over_time", + "experimental": true + }, + { + "name": "max_over_time", + "experimental": false + }, + { + "name": "min_over_time", + "experimental": false + }, + { + "name": "ts_of_max_over_time", + "experimental": true + }, + { + "name": "ts_of_min_over_time", + "experimental": true + }, + { + "name": "ts_of_last_over_time", + "experimental": true + }, + { + "name": "predict_linear", + "experimental": false + }, + { + "name": "present_over_time", + "experimental": false + }, + { + "name": "quantile_over_time", + "experimental": false + }, + { + "name": "rate", + "experimental": false + }, + { + "name": "resets", + "experimental": false + }, + { + "name": "stddev_over_time", + "experimental": false + }, + { + "name": "stdvar_over_time", + "experimental": false + }, + { + "name": "sum_over_time", + "experimental": false + } + ] +} diff --git a/tools/promql-smoke/run.py b/tools/promql-smoke/run.py new file mode 100644 index 00000000..6a952fb9 --- /dev/null +++ b/tools/promql-smoke/run.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 +"""Run small, explicit fixtures for every Prometheus 3.5 aggregation and rollup.""" +import argparse +import hashlib +import json +import math +from pathlib import Path +import re +import subprocess +import urllib.error +import urllib.parse +import urllib.request +import xml.etree.ElementTree as ET + +HERE = Path(__file__).resolve().parent +FEATURE = "promql-experimental-functions" +SPECIAL_YAML = {"NaN": ".nan", "+Inf": ".inf", "-Inf": "-.inf"} + + +def selector(labels): + name = labels.get("__name__", "") + rest = ",".join( + f"{key}={json.dumps(value)}" + for key, value in sorted(labels.items()) if key != "__name__" + ) + return name + "{" + rest + "}" + + +def save_json(path, value): + path.write_text(json.dumps(value, indent=2, allow_nan=False) + "\n") + + +def check_coverage(cases, catalog, verify_source=False): + """Fail if any registered function is missing, even if all present tests pass.""" + if cases["prometheus_version"] != catalog["prometheus_version"]: + raise ValueError("Fixture and catalog versions differ") + ids = [query["id"] for query in cases["queries"]] + if len(ids) != len(set(ids)): + raise ValueError("Duplicate case IDs") + report = {"prometheus_version": catalog["prometheus_version"], "scope": catalog["scope"]} + for category in ("aggregation", "rollup"): + entries = {entry["name"]: entry for entry in catalog[category]} + covered = {} + for query in cases["queries"]: + if query["category"] != category: + continue + name = query["function"] + if name not in entries: + raise ValueError(f"Unregistered {category}: {name}") + if query["experimental"] != entries[name]["experimental"]: + raise ValueError(f"Incorrect experimental flag: {query['id']}") + if not re.search(rf"\b{re.escape(name)}\s*(?:\(|by\b|without\b)", query["expr"]): + raise ValueError(f"Case does not exercise its declared function: {query['id']}") + covered.setdefault(name, []).append(query["id"]) + missing = sorted(entries.keys() - covered.keys()) + if missing: + raise ValueError(f"Missing {category} coverage: {missing}") + report[category] = {"covered": len(covered), "total": len(entries), "cases": covered} + report["source_verified"] = False + if verify_source: + texts = {} + for name, source in catalog["sources"].items(): + with urllib.request.urlopen(source["url"], timeout=30) as response: + raw = response.read() + actual = hashlib.sha256(raw).hexdigest() + if actual != source["sha256"]: + raise ValueError(f"Upstream source hash changed: {source['url']}") + texts[name] = raw.decode() + blocks = re.findall(r'\n\t"([^"]+)": \{(.*?)\n\t\},', texts["functions.go"], re.S) + rollups = { + name: bool(re.search(r"Experimental:\s*true", body)) + for name, body in blocks if "ValueTypeMatrix" in body + } + block = texts["lex.go"].split("// Aggregators.")[1].split("// Keywords.")[0] + aggregates = { + name: name in ("limitk", "limit_ratio") + for name in re.findall(r'"([^"]+)":', block) + } + for category, actual in (("aggregation", aggregates), ("rollup", rollups)): + expected = {entry["name"]: entry["experimental"] for entry in catalog[category]} + if actual != expected: + raise ValueError(f"Catalog does not match official {category} registry") + report["source_verified"] = True + report["case_count"] = len(cases["queries"]) + report["experimental_case_count"] = sum(q["experimental"] for q in cases["queries"]) + return report + + +def expected_samples(query, start=0): + return [ + {"labels": sample["labels"], "value": ( + sample["value"] + start if query.get("value_is_timestamp") else sample["value"] + )} + for sample in query["expected"] + ] + + +def generate(cases, out): + lines = [] + previous_metric = None + count = 0 + for series in cases["series"]: + metric = series["labels"]["__name__"] + if metric != previous_metric: + family, kind = (metric[:-6], "counter") if metric.endswith("_total") else (metric, "gauge") + lines.append(f"# TYPE {family} {kind}") + previous_metric = metric + for index, value in enumerate(series["values"]): + if value is None: + continue # A missing sample is not a zero-valued observation. + timestamp = cases["start"] + index * cases["interval"] + lines.append(f'{selector(series["labels"])} {value} {timestamp}') + count += 1 + (out / "samples.openmetrics").write_text("\n".join(lines + ["# EOF", ""])) + inputs = [ + {"series": selector(series["labels"]), "values": " ".join( + "_" if value is None else str(value) for value in series["values"] + )} + for series in cases["series"] + ] + suites = [] + for experimental in (False, True): + groups = [] + for query in cases["queries"]: + if query["experimental"] != experimental: + continue + groups.append({ + "name": query["id"], "interval": f'{cases["interval"]}s', + "input_series": inputs, + "promql_expr_test": [{ + "expr": query.get("promtool_expr", query["expr"]), + "eval_time": f'{cases["eval_offset"]}s', + "exp_samples": [ + {"labels": selector(s["labels"]), "value": s["value"]} + for s in query.get("promtool_expected", expected_samples(query)) + ], + }], + }) + path = out / ("experimental.test.yml" if experimental else "rules.test.yml") + # JSON is YAML; special float expectations require YAML numeric scalars. + rendered = json.dumps({"fuzzy_compare": True, "tests": groups}, indent=2) + for value, yaml in SPECIAL_YAML.items(): + rendered = rendered.replace(f'"value": "{value}"', f'"value": {yaml}') + path.write_text(rendered + "\n") + suites.append((path, experimental, len(groups))) + print(f"Generated {len(cases['series'])} series / {count} samples / {len(cases['queries'])} queries in {out}", flush=True) + return suites + + +def reference_checks(promtool, suites, out, version): + version_run = subprocess.run([promtool, "--version"], capture_output=True, text=True, check=True) + version_text = version_run.stdout + version_run.stderr + if not re.search(rf"version {re.escape(version)}(?:\s|\(|,|$)", version_text): + raise ValueError(f"Expected promtool {version}, got: {version_text.strip()}") + results = [] + for path, experimental, count in suites: + command = [promtool] + if experimental: + command.append(f"--enable-feature={FEATURE}") + command += ["test", "rules", f"--junit={path.with_suffix('.xml')}", str(path)] + run = subprocess.run(command, capture_output=True, text=True) + log = run.stdout + run.stderr + path.with_suffix(".log").write_text(log) + status = "PASS" if run.returncode == 0 else "FAIL" + if run.returncode: + print(log, flush=True) + case_results = [] + junit = path.with_suffix(".xml") + if junit.exists(): + for case in ET.parse(junit).iter("testcase"): + passed = not any(case.find(tag) is not None for tag in ("failure", "error", "skipped")) + case_results.append({"id": case.attrib["name"], "status": "PASS" if passed else "FAIL"}) + if len(case_results) != count or any(row["status"] != "PASS" for row in case_results): + status = "FAIL" + print(f"Prometheus {path.name}: {status} ({count} cases)", flush=True) + results.append({"suite": path.name, "case_count": count, "status": status, "cases": case_results, + "command": command, "exit_code": run.returncode, "log": log}) + report = {"version": version_text.strip(), "suites": results} + save_json(out / "reference-results.json", report) + return all(row["status"] == "PASS" for row in results) + + +def compare_vector(body, expected, evaluation, ordered=False): + if body.get("status") != "success": + raise ValueError(f"Query error: {body.get('errorType')}: {body.get('error')}") + if body["data"]["resultType"] != "vector": + raise ValueError(f"Expected vector, got {body['data']['resultType']}") + actual = body["data"]["result"] + key = lambda labels: tuple(sorted(labels.items())) + wanted = {key(sample["labels"]): sample["value"] for sample in expected} + if len(actual) != len(wanted): + raise ValueError(f"Expected {len(wanted)} series, got {len(actual)}") + seen = set() + for series in actual: + labels = key(series["metric"]) + if labels in seen or labels not in wanted: + raise ValueError(f"Unexpected or duplicate labels: {labels}") + seen.add(labels) + timestamp, value = series["value"] + if float(timestamp) != evaluation: + raise ValueError(f"Expected timestamp {evaluation}, got {timestamp}") + actual_value, reference_value = float(value), float(wanted[labels]) + equal = ( + math.isnan(actual_value) and math.isnan(reference_value) + if math.isnan(reference_value) + else math.isclose(actual_value, reference_value, rel_tol=1e-12, abs_tol=1e-12) + ) + if not equal: + raise ValueError(f"{labels}: expected {reference_value}, got {actual_value}") + if ordered and [key(s["metric"]) for s in actual] != [key(s["labels"]) for s in expected]: + raise ValueError("Series order differs") + + +def backend_checks(base_url, cases, out): + evaluation = cases["start"] + cases["eval_offset"] + results = [] + for query in cases["queries"]: + expected = expected_samples(query, cases["start"]) + record = {"id": query["id"], "query": query["expr"], "expected": expected} + url = base_url.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode( + {"query": query["expr"], "time": evaluation} + ) + try: + try: + response = urllib.request.urlopen(url, timeout=15) + except urllib.error.HTTPError as error: + response = error # Retain the actual error body in the report. + with response: + record["http_status"] = response.code + body = json.load(response) + record["response"] = body + if record["http_status"] != 200: + raise ValueError(f"HTTP {record['http_status']}: {body}") + compare_vector(body, expected, evaluation, query.get("ordered", False)) + record["status"] = "PASS" + except (ValueError, KeyError, TypeError, OSError) as error: + record.update(status="FAIL", error=str(error)) + results.append(record) + print(record["status"], query["id"], record.get("error", ""), flush=True) + save_json(out / "backend-results.json", results) + print("Exact HTTP result checks only; inspect saved responses for execution/fallback provenance.") + return all(row["status"] == "PASS" for row in results) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output-dir", type=Path, default=Path("/tmp/asap-promql-smoke")) + parser.add_argument("--promtool", help="Run both stable and experimental official expression suites") + parser.add_argument("--backend-url", help="Query a backend already loaded with these samples") + parser.add_argument("--verify-catalog", action="store_true", help="Verify pinned official registry source hashes online") + args = parser.parse_args() + cases = json.loads((HERE / "cases.json").read_text()) + catalog = json.loads((HERE / "catalog.json").read_text()) + out = args.output_dir.resolve() + out.mkdir(parents=True, exist_ok=True) + coverage = check_coverage(cases, catalog, args.verify_catalog) + save_json(out / "coverage.json", coverage) + print("Coverage: " + ", ".join( + f"{coverage[category]['covered']}/{coverage[category]['total']} {category}" + for category in ("aggregation", "rollup") + )) + suites = generate(cases, out) + success = True + if args.promtool: + success = reference_checks(args.promtool, suites, out, catalog["prometheus_version"]) + if args.backend_url: + success = backend_checks(args.backend_url, cases, out) and success + if not success: + raise SystemExit(1) + + +if __name__ == "__main__": + main() diff --git a/tools/promql-smoke/test_runner.py b/tools/promql-smoke/test_runner.py new file mode 100644 index 00000000..07c6e37c --- /dev/null +++ b/tools/promql-smoke/test_runner.py @@ -0,0 +1,90 @@ +"""Regression checks for failures that could otherwise turn a mismatch into a pass.""" +import copy +import json +from pathlib import Path +import unittest + +from run import check_coverage, compare_vector, expected_samples + +HERE = Path(__file__).resolve().parent + + +def vector(*samples): + return {"status": "success", "data": {"resultType": "vector", "result": [ + {"metric": labels, "value": [1788825840, str(value)]} + for labels, value in samples + ]}} + + +class ComparatorTests(unittest.TestCase): + def test_nonfinite_values(self): + for value in ("NaN", "+Inf", "-Inf"): + compare_vector(vector(({}, value)), [{"labels": {}, "value": value}], 1788825840) + for actual, wanted in (("NaN", 0), (0, "NaN"), ("-Inf", "+Inf"), (0, "+Inf")): + with self.subTest(actual=actual, wanted=wanted), self.assertRaises(ValueError): + compare_vector(vector(({}, actual)), [{"labels": {}, "value": wanted}], 1788825840) + + def test_empty_is_not_zero(self): + compare_vector(vector(), [], 1788825840) + with self.assertRaises(ValueError): + compare_vector(vector(({}, 0)), [], 1788825840) + + def test_checks_every_series_and_full_labels(self): + expected = [{"labels": {"job": "a"}, "value": 5}, {"labels": {"job": "b"}, "value": 50}] + bad_results = [ + vector(({"job": "a"}, 5), ({"job": "b"}, 51)), + vector(({"job": "a"}, 5), ({"job": "a"}, 50)), + vector(({"job": "a"}, 5), ({"job": "b", "__name__": "wrong"}, 50)), + ] + for body in bad_results: + with self.subTest(body=body), self.assertRaises(ValueError): + compare_vector(body, expected, 1788825840) + + def test_timestamp_value_uses_epoch_but_sample_time_is_evaluation(self): + query = {"value_is_timestamp": True, "expected": [{"labels": {"job": "a"}, "value": 120}]} + expected = expected_samples(query, 1788825600) + self.assertEqual(expected[0]["value"], 1788825720) + compare_vector(vector(({"job": "a"}, 1788825720)), expected, 1788825840) + with self.assertRaises(ValueError): + compare_vector(vector(({"job": "a"}, 120)), expected, 1788825840) + wrong_time = vector(({"job": "a"}, 1788825720)) + wrong_time["data"]["result"][0]["value"][0] -= 60 + with self.assertRaises(ValueError): + compare_vector(wrong_time, expected, 1788825840) + + def test_topk_order_is_only_checked_when_requested(self): + expected = [{"labels": {"job": "b"}, "value": 50}, {"labels": {"job": "a"}, "value": 5}] + reverse = vector(({"job": "a"}, 5), ({"job": "b"}, 50)) + compare_vector(reverse, expected, 1788825840) + with self.assertRaises(ValueError): + compare_vector(reverse, expected, 1788825840, ordered=True) + + def test_error_or_wrong_type_does_not_pass_as_empty(self): + for body in ({"status": "error", "error": "unsupported"}, + {"status": "success", "data": {"resultType": "scalar", "result": [0, "0"]}}): + with self.subTest(body=body), self.assertRaises(ValueError): + compare_vector(body, [], 1788825840) + + +class CoverageTests(unittest.TestCase): + def setUp(self): + self.cases = json.loads((HERE / "cases.json").read_text()) + self.catalog = json.loads((HERE / "catalog.json").read_text()) + + def test_missing_function_cannot_be_reported_as_complete(self): + self.cases["queries"] = [q for q in self.cases["queries"] if q["function"] != "rate"] + with self.assertRaisesRegex(ValueError, "Missing rollup coverage"): + check_coverage(self.cases, self.catalog) + + def test_experimental_flag_and_duplicate_ids_are_checked(self): + bad = copy.deepcopy(self.cases) + next(q for q in bad["queries"] if q["function"] == "limitk")["experimental"] = False + with self.assertRaisesRegex(ValueError, "experimental flag"): + check_coverage(bad, self.catalog) + self.cases["queries"].append(self.cases["queries"][0]) + with self.assertRaisesRegex(ValueError, "Duplicate case IDs"): + check_coverage(self.cases, self.catalog) + + +if __name__ == "__main__": + unittest.main() From b6a540261bb297a46911dc2469c420a9606022eb Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 21:53:08 -0600 Subject: [PATCH 3/3] test(promql): satisfy clippy and record PR branch validation --- data_plane/tests/promql_exact_execution.rs | 2 +- tools/promql-smoke/.gitignore | 1 + tools/promql-smoke/RESULTS.md | 2 ++ 3 files changed, 4 insertions(+), 1 deletion(-) create mode 100644 tools/promql-smoke/.gitignore diff --git a/data_plane/tests/promql_exact_execution.rs b/data_plane/tests/promql_exact_execution.rs index 41a89b08..21f55c4e 100644 --- a/data_plane/tests/promql_exact_execution.rs +++ b/data_plane/tests/promql_exact_execution.rs @@ -108,7 +108,7 @@ fn invalid_raw_snapshots_are_rejected() { .collect(), samples: vec![(1.0, 2.0), (1.0, 3.0)], }; - assert!(execute(&plan, &[series.clone()], 2.0, 300.0).is_err()); + assert!(execute(&plan, std::slice::from_ref(&series), 2.0, 300.0).is_err()); let unique = RawSeries { samples: vec![(1.0, 2.0)], ..series diff --git a/tools/promql-smoke/.gitignore b/tools/promql-smoke/.gitignore new file mode 100644 index 00000000..c18dd8d8 --- /dev/null +++ b/tools/promql-smoke/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/tools/promql-smoke/RESULTS.md b/tools/promql-smoke/RESULTS.md index 3acc44a3..f7f7d3a1 100644 --- a/tools/promql-smoke/RESULTS.md +++ b/tools/promql-smoke/RESULTS.md @@ -20,6 +20,8 @@ Run date: 2026-09-13. | Native executor regression tests | 5/5 passed, including the 105-case corpus | | Planner frontend regression/conformance/lowering/equivalence tests | 162/162 passed | | Python comparator/coverage tests | 8/8 passed | +| Backend query parser regression tests | 7/7 passed | +| Targeted Clippy (`-D warnings`) and Cargo format check | Passed | All 14 aggregation operators and 25 range-vector functions in the pinned catalog have an executable exact binding. This measures float-sample function coverage,