From 25cd53a57beb8406a4eec9980b3d5e91ca6dd091 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Mon, 28 Sep 2026 07:38:51 +0000 Subject: [PATCH] fix(promql): match Prometheus for set operators, sgn, irate, min/max, absent, and histogram_quantile (#14923) --- src/config/src/meta/promql/value.rs | 21 +- src/promql/src/aggregations/extrema.rs | 21 +- src/promql/src/aggregations/mod.rs | 10 +- src/promql/src/binary/vector_operations.rs | 197 ++++++++++++++++--- src/promql/src/engine/call.rs | 10 + src/promql/src/engine/mod.rs | 29 +++ src/promql/src/functions/absent.rs | 139 +++++++++++-- src/promql/src/functions/absent_over_time.rs | 167 ++++++++-------- src/promql/src/functions/histogram.rs | 20 +- src/promql/src/functions/idelta.rs | 8 +- src/promql/src/functions/irate.rs | 31 ++- src/promql/src/functions/math_operations.rs | 17 +- src/promql/src/functions/mod.rs | 4 +- 13 files changed, 497 insertions(+), 177 deletions(-) diff --git a/src/config/src/meta/promql/value.rs b/src/config/src/meta/promql/value.rs index d07334e936..faea8de5bd 100644 --- a/src/config/src/meta/promql/value.rs +++ b/src/config/src/meta/promql/value.rs @@ -1019,10 +1019,19 @@ pub fn signature(labels: &Labels) -> u64 { /// matching `names`. // REFACTORME: make this a method of `Metric` pub fn signature_without_labels(labels: &Labels, exclude_names: &[&str]) -> u64 { + hash_labels(labels, |name| !exclude_names.contains(&name)) +} + +/// [`signature`] of only the labels named in `include_names`. +pub fn signature_with_labels(labels: &Labels, include_names: &[&str]) -> u64 { + hash_labels(labels, |name| include_names.contains(&name)) +} + +fn hash_labels(labels: &Labels, include: impl Fn(&str) -> bool) -> u64 { let mut hasher = crate::utils::hash::gxhash::new_hasher(); labels .iter() - .filter(|item| !exclude_names.contains(&item.name.as_str())) + .filter(|item| include(item.name.as_str())) .for_each(|item| { hasher.write(item.name.as_bytes()); hasher.write(item.value.as_bytes()); @@ -1093,6 +1102,16 @@ mod tests { sig_without_ac, signature_without_labels(&labels, &["a", "c"]) ); + + let kept: Labels = labels + .iter() + .filter(|label| label.name == "a" || label.name == "c") + .cloned() + .collect(); + assert_eq!( + signature_with_labels(&labels, &["a", "c"]), + signature(&kept) + ); } #[test] diff --git a/src/promql/src/aggregations/extrema.rs b/src/promql/src/aggregations/extrema.rs index 52853f920b..53dc02a938 100644 --- a/src/promql/src/aggregations/extrema.rs +++ b/src/promql/src/aggregations/extrema.rs @@ -24,24 +24,21 @@ pub struct ExtremaAccumulator { impl ExtremaAccumulator { pub(super) fn new(slots: usize) -> Self { - let initial = if IS_MAX { - f64::NEG_INFINITY - } else { - f64::INFINITY - }; Self { - values: vec![initial; slots], + values: vec![f64::NAN; slots], present: vec![false; slots], } } fn push(&mut self, slot: usize, value: f64) { - // Strict comparisons preserve the first signed zero and ignore NaN values. - let replace = if IS_MAX { - value > self.values[slot] - } else { - value < self.values[slot] - }; + // Strict comparisons keep the first signed zero; NaN survives only if every value is NaN. + let current = self.values[slot]; + let replace = current.is_nan() + || if IS_MAX { + value > current + } else { + value < current + }; if replace { self.values[slot] = value; } diff --git a/src/promql/src/aggregations/mod.rs b/src/promql/src/aggregations/mod.rs index 54c1a59420..74e5a639a7 100644 --- a/src/promql/src/aggregations/mod.rs +++ b/src/promql/src/aggregations/mod.rs @@ -624,10 +624,12 @@ mod tests { #[test] fn test_extrema_merge_preserves_nan_and_signed_zero_behavior() { - fn check(func: impl AggFunc, initial: f64) { + fn check(func: impl AggFunc) { for (values, expected) in [ - (vec![f64::NAN], initial), + (vec![f64::NAN], f64::NAN), + (vec![f64::NAN, f64::NAN], f64::NAN), (vec![f64::NAN, 7.0, f64::NAN], 7.0), + (vec![7.0, f64::NAN], 7.0), (vec![-0.0, 0.0], -0.0), (vec![0.0, -0.0], 0.0), ] { @@ -651,8 +653,8 @@ mod tests { } } } - check(Min, f64::INFINITY); - check(Max, f64::NEG_INFINITY); + check(Min); + check(Max); } #[test] diff --git a/src/promql/src/binary/vector_operations.rs b/src/promql/src/binary/vector_operations.rs index 09dbc575be..7f7536f993 100644 --- a/src/promql/src/binary/vector_operations.rs +++ b/src/promql/src/binary/vector_operations.rs @@ -17,7 +17,10 @@ use std::collections::{HashMap, HashSet, hash_map::Entry}; use config::meta::promql::{ NAME_LABEL, - value::{Labels, LabelsExt, RangeValue, Sample, Value}, + value::{ + Labels, LabelsExt, RangeValue, Sample, Value, signature_with_labels, + signature_without_labels, + }, }; use datafusion::error::{DataFusionError, Result}; use promql_parser::parser::{BinaryExpr, VectorMatchCardinality, token}; @@ -178,12 +181,11 @@ fn vector_or(expr: &BinaryExpr, left: Vec, right: Vec) - if right.is_empty() { return Ok(Value::Matrix(left)); } - // Add all right-hand side elements which have not been added from the left-hand - // side. - let unmatched = filter_set_series(right, &left, false); + let unmatched = filter_set_series(right, &left, false, match_signature(expr)); let mut output = left; output.extend(unmatched); - Ok(Value::Matrix(output)) + // equal labels imply an equal signature, so left and right fragments never share a step + Ok(Value::Matrix(merge_same_labelset(output)?)) } /// matrix1 unless matrix2 results in a matrix consisting of the elements of @@ -202,7 +204,12 @@ fn vector_unless( if left.is_empty() || right.is_empty() { return Ok(Value::Matrix(left)); } - Ok(Value::Matrix(filter_set_series(left, &right, false))) + Ok(Value::Matrix(filter_set_series( + left, + &right, + false, + match_signature(expr), + ))) } /// matrix1 and matrix2 results in a matrix consisting of the elements of @@ -217,14 +224,12 @@ fn vector_and(expr: &BinaryExpr, left: Vec, right: Vec) if left.is_empty() || right.is_empty() { return Ok(Value::Matrix(vec![])); } - let output = filter_set_series(left, &right, true) - .into_par_iter() - .map(|mut range| { - range.labels = range.labels.without_metric_name(); - range - }) - .collect(); - Ok(Value::Matrix(output)) + Ok(Value::Matrix(filter_set_series( + left, + &right, + true, + match_signature(expr), + ))) } fn validate_set_matching(expr: &BinaryExpr) -> Result<()> { @@ -236,43 +241,72 @@ fn validate_set_matching(expr: &BinaryExpr) -> Result<()> { Ok(()) } +/// Keeps `input` samples whose step matches (`keep_matches`) or misses an `other` sample. fn filter_set_series( input: Vec, other: &[RangeValue], keep_matches: bool, + signature: impl Fn(&Labels) -> u64 + Sync, ) -> Vec { - let signatures: HashSet = other - .par_iter() - .map(|series| series.labels.signature()) - .collect(); + let mut other_steps: HashMap> = HashMap::with_capacity(other.len()); + for range in other { + other_steps + .entry(signature(&range.labels)) + .or_default() + .extend(range.samples.iter().map(|sample| sample.timestamp)); + } input .into_par_iter() - .filter(|series| signatures.contains(&series.labels.signature()) == keep_matches) + .filter_map(|mut range| { + let steps = other_steps.get(&signature(&range.labels)); + range.samples.retain(|sample| { + steps.is_some_and(|steps| steps.contains(&sample.timestamp)) == keep_matches + }); + (!range.samples.is_empty()).then_some(range) + }) .collect() } +/// Signature of the labels `on`/`ignoring` compare; without `on`, `__name__` is never compared. +fn match_signature(expr: &BinaryExpr) -> impl Fn(&Labels) -> u64 + Sync + '_ { + let on = expr.is_matching_on(); + let mut names: Vec<&str> = expr + .modifier + .as_ref() + .and_then(|modifier| modifier.matching.as_ref()) + .map(|matching| { + matching + .labels() + .labels + .iter() + .map(String::as_str) + .collect() + }) + .unwrap_or_default(); + if !on { + names.push(NAME_LABEL); + } + move |labels: &Labels| { + if on { + signature_with_labels(labels, &names) + } else { + signature_without_labels(labels, &names) + } + } +} + fn vector_arithmetic_operators( expr: &BinaryExpr, left: Vec, right: Vec, ) -> Result { - let is_matching_on = expr.is_matching_on(); let matching_labels: Vec = expr .modifier .as_ref() .and_then(|modifier| modifier.matching.as_ref()) .map(|matching| matching.labels().labels.clone()) .unwrap_or_default(); - let mut excluded_labels = matching_labels.clone(); - excluded_labels.push(NAME_LABEL.to_string()); - - let match_signature = |labels: &Labels| { - if is_matching_on { - labels.keep(&matching_labels).signature() - } else { - labels.delete(&excluded_labels).signature() - } - }; + let match_signature = match_signature(expr); let card = expr.modifier.as_ref().map(|modifier| &modifier.card); // group_right makes the lhs the "one" side, so the rhs series drive the output @@ -1373,4 +1407,107 @@ mod tests { assert_eq!(sample_pairs(&result[0]), vec![(1, 5.0), (3, 10.0)]); assert_eq!(sample_pairs(&result[1]), vec![(2, 5.0)]); } + + #[test] + fn test_and_keeps_metric_name_and_ignores_it_when_matching() { + let left = vec![range_at( + &[(1, 1.0)], + vec![("__name__", "up"), ("job", "api")], + )]; + let right = vec![range_at( + &[(1, 2.0)], + vec![("__name__", "ready"), ("job", "api")], + )]; + let result = eval_bin_op("up and ready", left, right); + assert_eq!(result.len(), 1); + assert_eq!( + label_pairs(&result[0]), + pairs(&[("__name__", "up"), ("job", "api")]) + ); + assert_eq!(sample_pairs(&result[0]), vec![(1, 1.0)]); + } + + #[test] + fn test_set_operators_honor_on_and_ignoring() { + let left = || { + vec![ + range_at(&[(1, 1.0)], vec![("instance", "a"), ("job", "api")]), + range_at(&[(1, 2.0)], vec![("instance", "b"), ("job", "db")]), + ] + }; + let right = || { + vec![range_at( + &[(1, 9.0)], + vec![("instance", "x"), ("job", "api")], + )] + }; + for (query, expected) in [ + ("a and b", vec![]), + ( + "a and on (job) b", + vec![pairs(&[("instance", "a"), ("job", "api")])], + ), + ( + "a and ignoring (instance) b", + vec![pairs(&[("instance", "a"), ("job", "api")])], + ), + ( + "a unless on (job) b", + vec![pairs(&[("instance", "b"), ("job", "db")])], + ), + ( + "a or on (job) b", + vec![ + pairs(&[("instance", "a"), ("job", "api")]), + pairs(&[("instance", "b"), ("job", "db")]), + ], + ), + ( + "a or b", + vec![ + pairs(&[("instance", "a"), ("job", "api")]), + pairs(&[("instance", "b"), ("job", "db")]), + pairs(&[("instance", "x"), ("job", "api")]), + ], + ), + ] { + let mut labels: Vec<_> = eval_bin_op(query, left(), right()) + .iter() + .map(label_pairs) + .collect(); + labels.sort(); + assert_eq!(labels, expected, "{query}"); + } + } + + #[test] + fn test_set_operators_match_per_step() { + let left = || { + vec![range_at( + &[(1, 1.0), (2, 2.0), (3, 3.0)], + vec![("job", "api")], + )] + }; + let right = || vec![range_at(&[(2, 20.0), (4, 40.0)], vec![("job", "api")])]; + let samples = |query| { + let mut result = eval_bin_op(query, left(), right()); + result.sort_by_key(|range| range.samples[0].timestamp); + result.iter().flat_map(sample_pairs).collect::>() + }; + assert_eq!(samples("a and b"), vec![(2, 2.0)]); + assert_eq!(samples("a unless b"), vec![(1, 1.0), (3, 3.0)]); + assert_eq!( + samples("a or b"), + vec![(1, 1.0), (2, 2.0), (3, 3.0), (4, 40.0)] + ); + } + + #[test] + fn test_or_merges_fragments_with_the_same_labels() { + let left = vec![range_at(&[(1, 1.0)], vec![])]; + let right = vec![range_at(&[(1, 2.0), (2, 2.0)], vec![])]; + let result = eval_bin_op("a or b", left, right); + assert_eq!(result.len(), 1); + assert_eq!(sample_pairs(&result[0]), vec![(1, 1.0), (2, 2.0)]); + } } diff --git a/src/promql/src/engine/call.rs b/src/promql/src/engine/call.rs index 5a8be00d99..78e3515be4 100644 --- a/src/promql/src/engine/call.rs +++ b/src/promql/src/engine/call.rs @@ -81,6 +81,16 @@ impl Engine { async fn call_builtin(&mut self, func_name: Func, args: &FunctionArgs) -> Result { match func_name { + Func::Absent | Func::AbsentOverTime => { + self.ensure_args_len(args, 1, "Invalid args, expected one argument")?; + let labels = functions::absent_labels(&args.args[0]); + let input = self.call_expr_arg(args, 0).await?; + if func_name == Func::Absent { + functions::absent(input, labels, &self.eval_ctx) + } else { + functions::absent_over_time(input, labels, &self.eval_ctx) + } + } Func::Clamp => { let err = "Invalid args, expected clamp(v instant-vector, min scalar, max scalar)"; self.ensure_args_len(args, 3, err)?; diff --git a/src/promql/src/engine/mod.rs b/src/promql/src/engine/mod.rs index 08796cc3c1..e7ffc68c57 100644 --- a/src/promql/src/engine/mod.rs +++ b/src/promql/src/engine/mod.rs @@ -689,6 +689,35 @@ pub(crate) mod tests { assert_eq!(series.iter().map(|s| s.samples.len()).sum::(), 4); } + #[tokio::test] + async fn test_or_fills_gaps_into_one_series() { + let series = matrix( + eval_on_empty("(vector(time()) < 1640995260) or vector(2)", 3) + .await + .unwrap(), + ); + assert_eq!(series.len(), 1); + let values: Vec = series[0].samples.iter().map(|s| s.value).collect(); + assert_eq!(values, vec![1640995200.0, 2.0, 2.0]); + + let query = "sum_over_time(((vector(time()) < 1640995260) or vector(2))[2m:1m])"; + let series = matrix(eval_on_empty(query, 2).await.unwrap()); + assert_eq!(series.len(), 1); + assert_eq!(series[0].samples.last().unwrap().value, 1640995202.0); + } + + #[tokio::test] + async fn test_absent_drops_empty_matcher_labels() { + let series = matrix( + eval_on_empty(r#"absent(missing{job=""}) + vector(1)"#, 1) + .await + .unwrap(), + ); + assert_eq!(series.len(), 1); + assert!(series[0].labels.is_empty()); + assert_eq!(series[0].samples[0].value, 2.0); + } + fn single_value(value: Value) -> f64 { match value { Value::Float(f) => f, diff --git a/src/promql/src/functions/absent.rs b/src/promql/src/functions/absent.rs index 3fd203fff1..3f70bc4f0e 100644 --- a/src/promql/src/functions/absent.rs +++ b/src/promql/src/functions/absent.rs @@ -13,14 +13,21 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::collections::BTreeSet; +use std::{ + collections::{BTreeMap, BTreeSet, HashSet}, + sync::Arc, +}; -use config::meta::promql::value::{EvalContext, Labels, RangeValue, Sample, Value}; +use config::meta::promql::{ + NAME_LABEL, + value::{EvalContext, Label, Labels, RangeValue, Sample, Value}, +}; use datafusion::error::{DataFusionError, Result}; +use promql_parser::{label::MatchOp, parser::Expr}; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#absent /// Returns 1 for each timestamp where the input vector has no data -pub(crate) fn absent(data: Value, eval_ctx: &EvalContext) -> Result { +pub(crate) fn absent(data: Value, labels: Labels, eval_ctx: &EvalContext) -> Result { let matrix = match data { Value::Matrix(matrix) => matrix, Value::None => vec![], @@ -31,29 +38,68 @@ pub(crate) fn absent(data: Value, eval_ctx: &EvalContext) -> Result { ))); } }; - // Collect all timestamps that have data across all series let present: BTreeSet<_> = matrix .iter() .flat_map(|series| series.samples.iter().map(|sample| sample.timestamp)) .collect(); - // Generate samples for timestamps that DON'T have data - let samples: Vec<_> = eval_ctx - .timestamps() + Ok(absent_series( + labels, + eval_ctx + .timestamps() + .into_iter() + .filter(|timestamp| !present.contains(timestamp)), + )) +} + +/// Prometheus' `createLabelsForAbsentFunction`: a selector's equality matchers, minus repeated +/// names. +pub(crate) fn absent_labels(expr: &Expr) -> Labels { + let selector = match expr { + Expr::Paren(paren) => return absent_labels(&paren.expr), + Expr::VectorSelector(vs) => vs, + Expr::MatrixSelector(ms) => &ms.vs, + _ => return Labels::default(), + }; + let mut labels = BTreeMap::new(); + let mut seen = HashSet::new(); + for matcher in &selector.matchers.matchers { + if matcher.name == NAME_LABEL { + continue; + } + // an empty value unsets the label, as Prometheus' labels.Builder does + if matches!(matcher.op, MatchOp::Equal) + && seen.insert(matcher.name.as_str()) + && !matcher.value.is_empty() + { + labels.insert(matcher.name.as_str(), matcher.value.as_str()); + } else { + labels.remove(matcher.name.as_str()); + } + } + labels .into_iter() - .filter(|timestamp| !present.contains(timestamp)) + .map(|(name, value)| { + Arc::new(Label { + name: name.to_string(), + value: value.to_string(), + }) + }) + .collect() +} + +pub(super) fn absent_series(labels: Labels, timestamps: impl Iterator) -> Value { + let samples: Vec<_> = timestamps .map(|timestamp| Sample::new(timestamp, 1.0)) .collect(); - // If all timestamps have data, return None (empty result) if samples.is_empty() { - return Ok(Value::None); + return Value::None; } - // Return 1.0 for timestamps where data is absent - Ok(Value::Matrix(vec![RangeValue { - labels: Labels::default(), + Value::Matrix(vec![RangeValue { + labels, samples, exemplars: None, time_window: None, - }])) + }]) } #[cfg(test)] @@ -64,7 +110,7 @@ mod tests { fn test_absent_instant_with_zero_step() { let ctx = EvalContext::new(1_000_000, 1_000_000, 0, "test".into()); for input in [Value::None, Value::Matrix(vec![])] { - let Value::Matrix(matrix) = absent(input, &ctx).unwrap() else { + let Value::Matrix(matrix) = absent(input, Labels::default(), &ctx).unwrap() else { panic!("expected one absent sample"); }; assert_eq!(matrix.len(), 1); @@ -78,7 +124,10 @@ mod tests { exemplars: None, time_window: None, }]); - assert!(matches!(absent(input, &ctx).unwrap(), Value::None)); + assert!(matches!( + absent(input, Labels::default(), &ctx).unwrap(), + Value::None + )); } fn create_eval_ctx() -> EvalContext { @@ -95,7 +144,7 @@ mod tests { // Empty matrix should return 1.0 for all timestamps in range let eval_ctx = create_eval_ctx(); let value = Value::Matrix(vec![]); - let result = absent(value, &eval_ctx).unwrap(); + let result = absent(value, Labels::default(), &eval_ctx).unwrap(); if let Value::Matrix(matrix) = result { assert_eq!(matrix.len(), 1); @@ -117,7 +166,7 @@ mod tests { // None should return 1.0 for all timestamps let eval_ctx = create_eval_ctx(); let value = Value::None; - let result = absent(value, &eval_ctx).unwrap(); + let result = absent(value, Labels::default(), &eval_ctx).unwrap(); if let Value::Matrix(matrix) = result { assert_eq!(matrix.len(), 1); @@ -145,7 +194,7 @@ mod tests { exemplars: None, time_window: None, }]); - let result = absent(value, &eval_ctx).unwrap(); + let result = absent(value, Labels::default(), &eval_ctx).unwrap(); assert!(matches!(result, Value::None)); } @@ -163,7 +212,7 @@ mod tests { exemplars: None, time_window: None, }]); - let result = absent(value, &eval_ctx).unwrap(); + let result = absent(value, Labels::default(), &eval_ctx).unwrap(); if let Value::Matrix(matrix) = result { assert_eq!(matrix.len(), 1); @@ -181,7 +230,55 @@ mod tests { // Invalid input type should return error let eval_ctx = create_eval_ctx(); let value = Value::Float(5.0); - let result = absent(value, &eval_ctx); + let result = absent(value, Labels::default(), &eval_ctx); assert!(result.is_err()); } + + fn labels_of(query: &str) -> Vec<(String, String)> { + let Expr::Call(call) = promql_parser::parser::parse(query).unwrap() else { + panic!("not a call: {query}"); + }; + absent_labels(&call.args.args[0]) + .iter() + .map(|label| (label.name.clone(), label.value.clone())) + .collect() + } + + #[test] + fn test_absent_labels_follow_equality_matchers() { + let pairs = |labels: &[(&str, &str)]| -> Vec<(String, String)> { + labels + .iter() + .map(|(name, value)| (name.to_string(), value.to_string())) + .collect() + }; + assert_eq!(labels_of(r#"absent(up{job="x"})"#), pairs(&[("job", "x")])); + assert_eq!( + labels_of(r#"absent(up{job="x",instance=~".*",env!="dev",a="1"})"#), + pairs(&[("a", "1"), ("job", "x")]) + ); + assert_eq!( + labels_of(r#"absent(up{job="a",job="b",foo="bar"})"#), + pairs(&[("foo", "bar")]) + ); + assert_eq!( + labels_of(r#"absent({__name__="up",job="x"})"#), + pairs(&[("job", "x")]) + ); + assert_eq!( + labels_of(r#"absent_over_time(up{job="x"}[5m])"#), + pairs(&[("job", "x")]) + ); + assert_eq!( + labels_of(r#"absent(((up{job="x"})))"#), + pairs(&[("job", "x")]) + ); + assert!(labels_of(r#"absent(sum(up{job="x"}))"#).is_empty()); + assert_eq!( + labels_of(r#"absent(up{job="",env="a"})"#), + pairs(&[("env", "a")]) + ); + assert!(labels_of(r#"absent(up{job="",job="x"})"#).is_empty()); + assert!(labels_of(r#"absent_over_time(up{job="x"}[5m:1m])"#).is_empty()); + } } diff --git a/src/promql/src/functions/absent_over_time.rs b/src/promql/src/functions/absent_over_time.rs index 1da99ff713..85ec63604b 100644 --- a/src/promql/src/functions/absent_over_time.rs +++ b/src/promql/src/functions/absent_over_time.rs @@ -13,58 +13,71 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::time::Duration; +use config::meta::promql::value::{EvalContext, Labels, Value}; +use datafusion::error::{DataFusionError, Result}; -use config::meta::promql::value::{EvalContext, Labels, RangeValue, Sample, Value}; -use datafusion::error::Result; - -use crate::functions::RangeFunc; +use super::{SeriesRange, absent::absent_series, present_over_time::PresentOverTimeFunc}; /// https://prometheus.io/docs/prometheus/latest/querying/functions/#absent_over_time -pub(crate) fn absent_over_time(data: Value, eval_ctx: &EvalContext) -> Result { - // A selector matching nothing arrives as None or as an empty matrix, and eval_range walks - // series, so it has none to report on - the absence this function exists for is the one case - // it cannot see. Answer that here, before delegating the per-series case to it. - let nothing_matched = match &data { - Value::None => true, - Value::Matrix(matrix) => matrix.is_empty(), - _ => false, +pub(crate) fn absent_over_time( + data: Value, + labels: Labels, + eval_ctx: &EvalContext, +) -> Result { + let matrix = match data { + Value::Matrix(matrix) => matrix, + Value::None => vec![], + other => { + return Err(DataFusionError::Plan(format!( + "absent_over_time: matrix argument expected but got {}", + other.get_type() + ))); + } }; - if nothing_matched { - return Ok(Value::Matrix(vec![RangeValue { - labels: Labels::default(), - samples: eval_ctx - .timestamps() - .into_iter() - .map(|timestamp| Sample::new(timestamp, 1.0)) - .collect(), - exemplars: None, - time_window: None, - }])); - } - - super::eval_range(data, AbsentOverTimeFunc, eval_ctx) -} - -pub struct AbsentOverTimeFunc; - -impl RangeFunc for AbsentOverTimeFunc { - fn name(&self) -> &'static str { - "absent_over_time" - } - - fn exec(&self, samples: &[Sample], _eval_ts: i64, _range: &Duration) -> Option { - samples.is_empty().then_some(1.0) + let timestamps = eval_ctx.timestamps(); + let mut present = vec![false; timestamps.len()]; + let mut missing = timestamps.len(); + for series in &matrix { + if missing == 0 { + return Ok(Value::None); + } + let range = series + .time_window + .as_ref() + .ok_or_else(|| { + DataFusionError::Internal("absent_over_time: series without a range".into()) + })? + .range; + for (slot, _) in SeriesRange::new( + &series.samples, + &PresentOverTimeFunc, + range, + eval_ctx, + ×tamps, + ) { + if !present[slot] { + present[slot] = true; + missing -= 1; + } + } } + Ok(absent_series( + labels, + timestamps + .into_iter() + .zip(present) + .filter_map(|(timestamp, present)| (!present).then_some(timestamp)), + )) } #[cfg(test)] mod tests { - use config::meta::promql::value::TimeWindow; + use std::time::Duration; + + use config::meta::promql::value::{RangeValue, Sample, TimeWindow}; use super::*; - // Test helper fn reported(value: &Value) -> Vec<(i64, f64)> { match value { Value::Matrix(v) => v @@ -72,20 +85,31 @@ mod tests { .flat_map(|series| series.samples.iter()) .map(|sample| (sample.timestamp, sample.value)) .collect(), + Value::None => vec![], _ => panic!("Expected Matrix result"), } } - // Test helper + fn series(timestamps: &[i64]) -> RangeValue { + RangeValue { + labels: Labels::default(), + samples: timestamps.iter().map(|&ts| Sample::new(ts, 5.0)).collect(), + exemplars: None, + time_window: Some(TimeWindow { + range: Duration::from_micros(2000), + offset: Duration::ZERO, + }), + } + } + fn absent_over_time_test_helper(data: Value) -> Result { let eval_ctx = EvalContext::new(3000, 3000, 0, "test".to_string()); - absent_over_time(data, &eval_ctx) + absent_over_time(data, Labels::default(), &eval_ctx) } #[test] fn test_absent_over_time_value_none_input() { let result = absent_over_time_test_helper(Value::None).unwrap(); - match &result { Value::Matrix(v) => assert_eq!( v.len(), @@ -95,13 +119,14 @@ mod tests { _ => panic!("Expected Matrix result"), } assert_eq!(reported(&result), vec![(3000, 1.0)]); + let result = absent_over_time_test_helper(Value::Matrix(vec![])).unwrap(); + assert_eq!(reported(&result), vec![(3000, 1.0)]); } #[test] fn test_absent_over_time_reports_every_step_of_a_range_query() { let eval_ctx = EvalContext::new(1000, 3000, 1000, "test".to_string()); - let result = absent_over_time(Value::None, &eval_ctx).unwrap(); - + let result = absent_over_time(Value::None, Labels::default(), &eval_ctx).unwrap(); assert_eq!( reported(&result), vec![(1000, 1.0), (2000, 1.0), (3000, 1.0)] @@ -110,21 +135,23 @@ mod tests { #[test] fn test_absent_over_time_is_silent_while_the_metric_reports() { - let range_value = RangeValue { - labels: Labels::default(), - samples: vec![Sample::new(3000, 5.0)], - exemplars: None, - time_window: Some(TimeWindow { - range: Duration::from_secs(2), - offset: Duration::ZERO, - }), - }; - let result = absent_over_time_test_helper(Value::Matrix(vec![range_value])).unwrap(); + let result = absent_over_time_test_helper(Value::Matrix(vec![series(&[3000])])).unwrap(); + assert!( + reported(&result).is_empty(), + "present data must report no absence" + ); + } - match result { - Value::Matrix(v) => assert!(v.is_empty(), "present data must report no absence"), - _ => panic!("Expected Matrix result"), - } + #[test] + fn test_absent_over_time_reports_gaps_of_an_existing_series() { + let eval_ctx = EvalContext::new(2000, 10000, 2000, "test".to_string()); + // windows are (t - 2000, t], so 3000 covers step 4000 and 9000 covers step 10000 + let data = Value::Matrix(vec![series(&[3000]), series(&[9000])]); + let result = absent_over_time(data, Labels::default(), &eval_ctx).unwrap(); + assert_eq!( + reported(&result), + vec![(2000, 1.0), (6000, 1.0), (8000, 1.0)] + ); } #[test] @@ -132,26 +159,4 @@ mod tests { let result = absent_over_time_test_helper(Value::Float(1.0)); assert!(result.is_err()); } - - #[test] - fn test_absent_over_time_exec_samples_present_returns_none() { - let func = AbsentOverTimeFunc; - let samples = vec![Sample::new(1000, 5.0)]; - assert!(func.exec(&samples, 0, &Duration::ZERO).is_none()); - } - - #[test] - fn test_absent_over_time_exec_empty_returns_one() { - let func = AbsentOverTimeFunc; - assert_eq!(func.exec(&[], 0, &Duration::ZERO), Some(1.0)); - } - - #[test] - fn test_absent_over_time_function() { - // Test with empty matrix - should return 1.0 - let empty_matrix = Value::Matrix(vec![]); - let result = absent_over_time_test_helper(empty_matrix).unwrap(); - - assert_eq!(reported(&result), vec![(3000, 1.0)]); - } } diff --git a/src/promql/src/functions/histogram.rs b/src/promql/src/functions/histogram.rs index 65e03dc2a0..de161c0b3d 100644 --- a/src/promql/src/functions/histogram.rs +++ b/src/promql/src/functions/histogram.rs @@ -118,12 +118,11 @@ pub(crate) fn histogram_quantile( { *cursor += 1; } - let sample = bucket_rv + if let Some(sample) = bucket_rv .samples .get(*cursor) .filter(|sample| sample.timestamp == eval_ts) - .or_else(|| bucket_rv.samples.first()); - if let Some(sample) = sample { + { buckets.push(Bucket::new(*upper_bound, sample.value)); } } @@ -317,12 +316,11 @@ mod tests { { *cursor += 1; } - let sample = bucket_rv + if let Some(sample) = bucket_rv .samples .get(*cursor) .filter(|sample| sample.timestamp == eval_ts) - .or_else(|| bucket_rv.samples.first()); - if let Some(sample) = sample { + { buckets.push(Bucket::new(*upper_bound, sample.value)); } } @@ -752,11 +750,9 @@ mod tests { time_window: None, }; let input = Value::Matrix(vec![ - // Intentionally put +Inf before the finite bound. Timestamp 2 is - // absent, so the established fallback-to-first-sample behavior is - // exercised as well as the cursor fast path. + // +Inf comes first on purpose; no bucket has a sample at 2, and only +Inf at 3 series("+Inf", vec![Sample::new(1, 10.0), Sample::new(3, 30.0)]), - series("1", vec![Sample::new(1, 5.0), Sample::new(3, 15.0)]), + series("1", vec![Sample::new(1, 5.0)]), ]); let Value::Matrix(result) = @@ -769,9 +765,9 @@ mod tests { result[0] .samples .iter() - .map(|sample| (sample.timestamp, sample.value)) + .map(|sample| (sample.timestamp, sample.value.to_bits())) .collect::>(), - vec![(1, 1.0), (2, 1.0), (3, 1.0)], + vec![(1, 1.0f64.to_bits()), (3, f64::NAN.to_bits())], ); } diff --git a/src/promql/src/functions/idelta.rs b/src/promql/src/functions/idelta.rs index 9df3823e9d..58263b6e18 100644 --- a/src/promql/src/functions/idelta.rs +++ b/src/promql/src/functions/idelta.rs @@ -30,7 +30,7 @@ impl RangeFunc for IdeltaFunc { let [.., previous, last] = samples else { return None; }; - Some(last.value - previous.value) + (last.timestamp != previous.timestamp).then_some(last.value - previous.value) } } @@ -106,4 +106,10 @@ mod tests { _ => panic!("Expected Matrix result"), } } + + #[test] + fn test_idelta_exec_same_timestamp_returns_none() { + let samples = vec![Sample::new(1000, 10.0), Sample::new(1000, 20.0)]; + assert!(IdeltaFunc.exec(&samples, 1000, &Duration::ZERO).is_none()); + } } diff --git a/src/promql/src/functions/irate.rs b/src/promql/src/functions/irate.rs index d9dc4d5f9f..52eab1d0b6 100644 --- a/src/promql/src/functions/irate.rs +++ b/src/promql/src/functions/irate.rs @@ -30,14 +30,14 @@ impl RangeFunc for IrateFunc { let [.., previous, last] = samples else { return None; }; - let dt_seconds = (last.timestamp - previous.timestamp) as f64 / 1_000_000.0; - if dt_seconds == 0.0 { - return Some(0.0); + if last.timestamp == previous.timestamp { + return None; } - let dt_value = if last.value - previous.value >= 0.0 { - last.value - previous.value - } else { + let dt_seconds = (last.timestamp - previous.timestamp) as f64 / 1_000_000.0; + let dt_value = if last.value < previous.value { last.value + } else { + last.value - previous.value }; Some(dt_value / dt_seconds) } @@ -120,11 +120,24 @@ mod tests { } #[test] - fn test_irate_exec_same_timestamp_returns_zero() { + fn test_irate_exec_same_timestamp_returns_none() { let func = IrateFunc; let samples = vec![Sample::new(1000, 10.0), Sample::new(1000, 20.0)]; - let result = func.exec(&samples, 1000, &Duration::ZERO); - assert_eq!(result, Some(0.0)); + assert!(func.exec(&samples, 1000, &Duration::ZERO).is_none()); + } + + #[test] + fn test_irate_exec_nan_previous_propagates() { + let func = IrateFunc; + let samples = vec![ + Sample::new(1_000_000, f64::NAN), + Sample::new(2_000_000, 5.0), + ]; + assert!( + func.exec(&samples, 2_000_000, &Duration::ZERO) + .unwrap() + .is_nan() + ); } #[test] diff --git a/src/promql/src/functions/math_operations.rs b/src/promql/src/functions/math_operations.rs index b19305b632..9629d5637c 100644 --- a/src/promql/src/functions/math_operations.rs +++ b/src/promql/src/functions/math_operations.rs @@ -59,7 +59,16 @@ pub(crate) fn round(data: Value, to_nearest: &ScalarParam) -> Result { } pub(crate) fn sgn(data: Value) -> Result { - exec(data, f64::signum) + // f64::signum maps ±0 to ±1; Prometheus returns zeros and NaN unchanged. + exec(data, |value| { + if value > 0.0 { + 1.0 + } else if value < 0.0 { + -1.0 + } else { + value + } + }) } /// Apply a given simple match function to a float type @@ -151,7 +160,9 @@ mod tests { assert_eq!(round(12.4), 10.0); assert_eq!(apply(sgn, 5.0), 1.0); - assert_eq!(apply(sgn, 0.0), 1.0); + assert_eq!(apply(sgn, 0.0).to_bits(), 0.0f64.to_bits()); + assert_eq!(apply(sgn, -0.0).to_bits(), (-0.0f64).to_bits()); + assert!(apply(sgn, f64::NAN).is_nan()); assert_eq!(apply(sgn, -5.0), -1.0); } @@ -304,7 +315,7 @@ mod tests { assert_eq!(result_matrix.len(), 3); assert_eq!(result_matrix[0].samples[0].value, 1.0); assert_eq!(result_matrix[1].samples[0].value, -1.0); - assert_eq!(result_matrix[2].samples[0].value, 1.0); + assert_eq!(result_matrix[2].samples[0].value, 0.0); } else { panic!("Expected Matrix result"); } diff --git a/src/promql/src/functions/mod.rs b/src/promql/src/functions/mod.rs index a842037055..fc7af1de2f 100644 --- a/src/promql/src/functions/mod.rs +++ b/src/promql/src/functions/mod.rs @@ -56,7 +56,7 @@ mod sum_over_time; mod time_operations; mod vector; -pub(crate) use absent::absent; +pub(crate) use absent::{absent, absent_labels}; pub(crate) use absent_over_time::absent_over_time; pub(crate) use clamp::clamp; pub(crate) use histogram::histogram_quantile; @@ -162,8 +162,6 @@ impl Func { Self::Sgn => SingleArgFunc::Value(sgn), Self::Sqrt => SingleArgFunc::Value(sqrt), Self::Timestamp => SingleArgFunc::Value(timestamp), - Self::Absent => SingleArgFunc::Context(absent), - Self::AbsentOverTime => SingleArgFunc::Context(absent_over_time), Self::Scalar => SingleArgFunc::Context(scalar), Self::Vector => SingleArgFunc::Context(vector), Self::DayOfMonth => SingleArgFunc::Date(day_of_month),