starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 16:10:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(promql): match Prometheus for set operators, sgn, irate, min/max, absent, and histogram_quantile (#14923)

Browse Source
This commit is contained in:
Huaijin
2026-09-28 07:38:51 +00:00
committed by GitHub
parent f4395f42f4
commit 25cd53a57b
13 changed files with 497 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/src/meta/promql/value.rs
+20 -1
View File
@@ -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]
src/promql/src/aggregations/extrema.rs
+9 -12
View File
@@ -24,24 +24,21 @@ pub struct ExtremaAccumulator<const IS_MAX: bool> {
impl<const IS_MAX: bool> ExtremaAccumulator<IS_MAX> {
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;
}
src/promql/src/aggregations/mod.rs
+6 -4
View File
@@ -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]
src/promql/src/binary/vector_operations.rs
+167 -30
View File
@@ -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<RangeValue>, right: Vec<RangeValue>) -
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<RangeValue>, right: Vec<RangeValue>)
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<RangeValue>,
other: &[RangeValue],
keep_matches: bool,
signature: impl Fn(&Labels) -> u64 + Sync,
) -> Vec<RangeValue> {
let signatures: HashSet<u64> = other
.par_iter()
.map(|series| series.labels.signature())
.collect();
let mut other_steps: HashMap<u64, HashSet<i64>> = 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<RangeValue>,
right: Vec<RangeValue>,
) -> Result<Value> {
let is_matching_on = expr.is_matching_on();
let matching_labels: Vec<String> = 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::<Vec<_>>()
};
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)]);
}
}
src/promql/src/engine/call.rs
+10
View File
@@ -81,6 +81,16 @@ impl Engine {
async fn call_builtin(&mut self, func_name: Func, args: &FunctionArgs) -> Result<Value> {
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)?;
src/promql/src/engine/mod.rs
+29
View File
@@ -689,6 +689,35 @@ pub(crate) mod tests {
assert_eq!(series.iter().map(|s| s.samples.len()).sum::<usize>(), 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<f64> = 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,
src/promql/src/functions/absent.rs
+118 -21
View File
@@ -13,14 +13,21 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
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<Value> {
pub(crate) fn absent(data: Value, labels: Labels, eval_ctx: &EvalContext) -> Result<Value> {
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<Value> {
)));
}
};
// 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<Item = i64>) -> 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());
}
}
src/promql/src/functions/absent_over_time.rs
+86 -81
View File
@@ -13,58 +13,71 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
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<Value> {
// 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<Value> {
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<f64> {
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,
&timestamps,
) {
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<Value> {
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)]);
}
}
src/promql/src/functions/histogram.rs
+8 -12
View File
@@ -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<_>>(),
vec![(1, 1.0), (2, 1.0), (3, 1.0)],
vec![(1, 1.0f64.to_bits()), (3, f64::NAN.to_bits())],
);
}
src/promql/src/functions/idelta.rs
+7 -1
View File
@@ -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());
}
}
src/promql/src/functions/irate.rs
+22 -9
View File
@@ -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]
src/promql/src/functions/math_operations.rs
+14 -3
View File
@@ -59,7 +59,16 @@ pub(crate) fn round(data: Value, to_nearest: &ScalarParam) -> Result<Value> {
}
pub(crate) fn sgn(data: Value) -> Result<Value> {
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");
}
src/promql/src/functions/mod.rs
+1 -3
View File
@@ -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),