mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-11-14 06:46:56 +00:00
feat(promql): evaluate topk, bottomk, scalar, absent and instant sorts over evaluated series
Signed-off-by: Dennis Zhuang <killme2008@gmail.com>
This commit is contained in:
Dennis Zhuang
@@ -19,3 +19,4 @@ pub mod grid_series;
|
||||
mod metrics;
|
||||
pub mod range_array;
|
||||
pub mod range_eval;
|
||||
pub mod select;
|
||||
|
||||
@@ -0,0 +1,450 @@
|
||||
// Copyright 2023 Greptime Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
//! Operators that depend on the whole set of evaluated series at each grid point: `topk`,
|
||||
//! `bottomk`, `scalar`, `absent` and the instant-result sorts.
|
||||
//!
|
||||
//! Semantics follow Prometheus (`aggregationK`, `funcScalar`, `funcAbsent`,
|
||||
//! `createLabelsForAbsentFunction`, `funcSort*`). A scalar is a grid series without labels that
|
||||
//! has a value at every point.
|
||||
|
||||
use std::cmp::Ordering;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::error::{DataFusionError, Result};
|
||||
use datatypes::series::samples::{SampleRef, Series, SeriesSchema};
|
||||
use datatypes::series::time::TimeAxis;
|
||||
use datatypes::series::{Label, MetricName};
|
||||
|
||||
use crate::grid_series::GridSeriesBuilder;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
/// The labels that form an aggregation group.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum Grouping {
|
||||
/// `by(...)`: only these labels, `__name__` included when listed.
|
||||
By(Vec<Arc<str>>),
|
||||
/// `without(...)`: every label except these and `__name__`.
|
||||
Without(Vec<Arc<str>>),
|
||||
}
|
||||
|
||||
/// The order `sort_instant` arranges an instant result in.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum SortOrder {
|
||||
/// `sort`: ascending values, NaN last.
|
||||
Asc,
|
||||
/// `sort_desc`: descending values, NaN last.
|
||||
Desc,
|
||||
/// `sort_by_label`: natural order of the listed label values, then the label sets.
|
||||
ByLabel(Vec<Arc<str>>),
|
||||
/// `sort_by_label_desc`.
|
||||
ByLabelDesc(Vec<Arc<str>>),
|
||||
}
|
||||
|
||||
/// `topk(k, ...)` (`top`) or `bottomk(k, ...)` with a constant `k`.
|
||||
///
|
||||
/// At every point each group keeps the `k` largest (or smallest) values, NaN ranking last for
|
||||
/// both; the selected samples keep their series' labels and values. Prometheus breaks ties by
|
||||
/// the order its storage returns series in, which is not a property of the series; here ties
|
||||
/// go to the series whose label set sorts first.
|
||||
pub fn topk(
|
||||
k: f64,
|
||||
top: bool,
|
||||
grouping: &Grouping,
|
||||
schema: &SeriesSchema,
|
||||
series: &[Series],
|
||||
) -> Result<Vec<Series>> {
|
||||
single_field(schema, "topk/bottomk")?;
|
||||
let points = grid_points(schema, series)?;
|
||||
// Prometheus converts k with int64() and caps it at the number of input series.
|
||||
let k = if k.is_nan() || k < 1.0 {
|
||||
0
|
||||
} else {
|
||||
(k.trunc().min(series.len() as f64)) as usize
|
||||
};
|
||||
if k == 0 {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let mut order: Vec<usize> = (0..series.len()).collect();
|
||||
order.sort_by(|&a, &b| compare_labels(&series[a].metric, &series[b].metric));
|
||||
let mut groups: HashMap<GroupKey, Vec<usize>> = HashMap::new();
|
||||
for index in order {
|
||||
groups
|
||||
.entry(group_key(&series[index].metric, grouping))
|
||||
.or_default()
|
||||
.push(index);
|
||||
}
|
||||
|
||||
let mut selected: Vec<Vec<usize>> = vec![Vec::new(); series.len()];
|
||||
let mut candidates: Vec<(usize, f64)> = Vec::new();
|
||||
for members in groups.values() {
|
||||
for point in 0..points {
|
||||
candidates.clear();
|
||||
for &member in members {
|
||||
if let Some(value) = value_at(&series[member], point)? {
|
||||
candidates.push((member, value));
|
||||
}
|
||||
}
|
||||
// Members are in label order, so a stable sort hands ties to the earlier label set.
|
||||
candidates.sort_by(|a, b| rank(a.1, b.1, top));
|
||||
for &(member, _) in candidates.iter().take(k) {
|
||||
selected[member].push(point);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let output = Arc::new(schema.clone());
|
||||
let mut builder = GridSeriesBuilder::try_new(output)?;
|
||||
let mut result = Vec::new();
|
||||
for (index, points) in selected.into_iter().enumerate() {
|
||||
if points.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let mut sorted = points;
|
||||
sorted.sort_unstable();
|
||||
for point in sorted {
|
||||
builder.push(point, &[value_at(&series[index], point)?])?;
|
||||
}
|
||||
if let Some(built) = builder.finish(series[index].metric.clone())? {
|
||||
result.push(built);
|
||||
}
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
/// `scalar(v)`: the value of the only series present at each point, NaN otherwise.
|
||||
pub fn scalar(schema: &SeriesSchema, series: &[Series]) -> Result<Series> {
|
||||
single_field(schema, "scalar")?;
|
||||
let points = grid_points(schema, series)?;
|
||||
let mut values: Vec<Option<f64>> = vec![None; points];
|
||||
let mut counts = vec![0u8; points];
|
||||
for item in series {
|
||||
let TimeAxis::Grid(range) = &item.time else {
|
||||
unreachable!("checked by grid_points");
|
||||
};
|
||||
for point in range.start()..range.start() + range.len() {
|
||||
if let Some(value) = value_at(item, point)? {
|
||||
counts[point] = counts[point].saturating_add(1);
|
||||
values[point] = Some(value);
|
||||
}
|
||||
}
|
||||
}
|
||||
let mut builder = GridSeriesBuilder::try_new(Arc::new(schema.clone()))?;
|
||||
for (point, (value, count)) in values.into_iter().zip(counts).enumerate() {
|
||||
let value = if count == 1 { value } else { Some(GO_NAN) };
|
||||
builder.push(point, &[value])?;
|
||||
}
|
||||
let empty = MetricName::try_new(None, Vec::new()).map_err(external)?;
|
||||
Ok(builder
|
||||
.finish(empty)?
|
||||
.expect("every point of a scalar has a value"))
|
||||
}
|
||||
|
||||
/// `absent(v)`: 1 with `labels` at every point where no series is present.
|
||||
pub fn absent(
|
||||
schema: &SeriesSchema,
|
||||
series: &[Series],
|
||||
labels: MetricName,
|
||||
) -> Result<Option<Series>> {
|
||||
let points = grid_points(schema, series)?;
|
||||
let mut present = vec![false; points];
|
||||
for item in series {
|
||||
let TimeAxis::Grid(range) = &item.time else {
|
||||
unreachable!("checked by grid_points");
|
||||
};
|
||||
let points = &mut present[range.start()..range.start() + range.len()];
|
||||
for (row, present) in points.iter_mut().enumerate() {
|
||||
if item
|
||||
.fields
|
||||
.iter()
|
||||
.any(|field| !matches!(field.sample(row), SampleRef::Missing))
|
||||
{
|
||||
*present = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
let ones = vec![Some(1.0); schema.fields.len()];
|
||||
let mut builder = GridSeriesBuilder::try_new(Arc::new(schema.clone()))?;
|
||||
for (point, present) in present.into_iter().enumerate() {
|
||||
if !present {
|
||||
builder.push(point, &ones)?;
|
||||
}
|
||||
}
|
||||
builder.finish(labels)
|
||||
}
|
||||
|
||||
/// A label matcher of the selector `absent` is applied to.
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct LabelMatcher {
|
||||
pub name: Arc<str>,
|
||||
pub op: MatchOp,
|
||||
pub value: Arc<str>,
|
||||
/// Whether the matcher accepts an empty value, i.e. a missing label; the caller evaluates
|
||||
/// regular expressions.
|
||||
pub matches_empty: bool,
|
||||
}
|
||||
|
||||
/// The operator of a [`LabelMatcher`].
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum MatchOp {
|
||||
Equal,
|
||||
NotEqual,
|
||||
Regex,
|
||||
NotRegex,
|
||||
}
|
||||
|
||||
/// The labels of an `absent` result, as Prometheus `createLabelsForAbsentFunction` derives them:
|
||||
/// the first equality matcher of a name sets that label, any other matcher of the name removes
|
||||
/// it, and `__name__` is skipped.
|
||||
///
|
||||
/// Prometheus reads the matchers after TSDB's `PostingsForMatchers` has reordered them in place,
|
||||
/// moving the matchers that narrow the result before those that only subtract from it; that is
|
||||
/// the order users see whenever the storage holds data, and the order used here. It changes the
|
||||
/// result only when a name has both an equality and another matcher.
|
||||
pub fn absent_labels(matchers: &[LabelMatcher]) -> Result<MetricName> {
|
||||
let must_be_set: Vec<&str> = matchers
|
||||
.iter()
|
||||
.filter(|matcher| !matcher.matches_empty)
|
||||
.map(|matcher| &*matcher.name)
|
||||
.collect();
|
||||
let subtracting = |matcher: &LabelMatcher| {
|
||||
!must_be_set.contains(&&*matcher.name)
|
||||
|| (matches!(matcher.op, MatchOp::NotEqual | MatchOp::NotRegex)
|
||||
&& matcher.matches_empty)
|
||||
};
|
||||
let ordered = matchers
|
||||
.iter()
|
||||
.filter(|matcher| !subtracting(matcher))
|
||||
.chain(matchers.iter().filter(|matcher| subtracting(matcher)));
|
||||
|
||||
let mut labels: Vec<Label> = Vec::new();
|
||||
let mut set: Vec<&str> = Vec::new();
|
||||
for matcher in ordered {
|
||||
if &*matcher.name == "__name__" {
|
||||
continue;
|
||||
}
|
||||
if matcher.op == MatchOp::Equal && !set.contains(&&*matcher.name) {
|
||||
labels.retain(|label| label.name != matcher.name);
|
||||
labels.push(Label {
|
||||
name: matcher.name.clone(),
|
||||
value: matcher.value.clone(),
|
||||
});
|
||||
set.push(&matcher.name);
|
||||
} else {
|
||||
labels.retain(|label| label.name != matcher.name);
|
||||
}
|
||||
}
|
||||
labels.retain(|label| !label.value.is_empty());
|
||||
MetricName::try_new(None, labels).map_err(external)
|
||||
}
|
||||
|
||||
/// Orders the series of an instant result. A range result is returned unchanged: Prometheus
|
||||
/// does not sort matrices.
|
||||
pub fn sort_instant(
|
||||
order: &SortOrder,
|
||||
schema: &SeriesSchema,
|
||||
mut series: Vec<Series>,
|
||||
) -> Result<Vec<Series>> {
|
||||
let points = grid_points(schema, &series)?;
|
||||
if points != 1 {
|
||||
return Ok(series);
|
||||
}
|
||||
match order {
|
||||
SortOrder::Asc | SortOrder::Desc => {
|
||||
single_field(schema, "sort")?;
|
||||
let mut keyed = Vec::with_capacity(series.len());
|
||||
for item in series {
|
||||
let value = value_at(&item, 0)?;
|
||||
keyed.push((value.unwrap_or(f64::NAN), item));
|
||||
}
|
||||
let descending = *order == SortOrder::Desc;
|
||||
keyed.sort_by(|(a, _), (b, _)| match (a.is_nan(), b.is_nan()) {
|
||||
(true, true) => Ordering::Equal,
|
||||
(true, false) => Ordering::Greater,
|
||||
(false, true) => Ordering::Less,
|
||||
_ if descending => b.total_cmp(a),
|
||||
_ => a.total_cmp(b),
|
||||
});
|
||||
Ok(keyed.into_iter().map(|(_, item)| item).collect())
|
||||
}
|
||||
SortOrder::ByLabel(labels) | SortOrder::ByLabelDesc(labels) => {
|
||||
let descending = matches!(order, SortOrder::ByLabelDesc(_));
|
||||
series.sort_by(|a, b| {
|
||||
for label in labels {
|
||||
let (a, b) = (
|
||||
a.metric.get(label).unwrap_or(""),
|
||||
b.metric.get(label).unwrap_or(""),
|
||||
);
|
||||
if a == b {
|
||||
continue;
|
||||
}
|
||||
let less = natural_less(a, b);
|
||||
return if less != descending {
|
||||
Ordering::Less
|
||||
} else {
|
||||
Ordering::Greater
|
||||
};
|
||||
}
|
||||
let order = compare_labels(&a.metric, &b.metric);
|
||||
if descending { order.reverse() } else { order }
|
||||
});
|
||||
Ok(series)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Go's `math.NaN()` bit pattern, which Prometheus returns for an undefined scalar.
|
||||
const GO_NAN: f64 = f64::from_bits(0x7FF8_0000_0000_0001);
|
||||
|
||||
/// `true` when `a` ranks before `b` for `topk` (`top`) or `bottomk`; NaN ranks last for both.
|
||||
fn rank(a: f64, b: f64, top: bool) -> Ordering {
|
||||
match (a.is_nan(), b.is_nan()) {
|
||||
(true, true) => Ordering::Equal,
|
||||
(true, false) => Ordering::Greater,
|
||||
(false, true) => Ordering::Less,
|
||||
_ if top => b.partial_cmp(&a).expect("not NaN"),
|
||||
_ => a.partial_cmp(&b).expect("not NaN"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Label pairs identifying a group; empty values are absent, as in Prometheus.
|
||||
type GroupKey = Vec<(Arc<str>, Arc<str>)>;
|
||||
|
||||
fn group_key(metric: &MetricName, grouping: &Grouping) -> GroupKey {
|
||||
let mut key: Vec<_> = labels_with_name(metric)
|
||||
.into_iter()
|
||||
.filter(|(name, value)| {
|
||||
!value.is_empty()
|
||||
&& match grouping {
|
||||
Grouping::By(names) => names.contains(name),
|
||||
Grouping::Without(names) => &**name != "__name__" && !names.contains(name),
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
key.sort();
|
||||
key
|
||||
}
|
||||
|
||||
/// The label pairs of `metric` with `__name__` among them, sorted by name.
|
||||
fn labels_with_name(metric: &MetricName) -> Vec<(Arc<str>, Arc<str>)> {
|
||||
let mut labels: Vec<_> = metric
|
||||
.tags()
|
||||
.iter()
|
||||
.map(|tag| (tag.name.clone(), tag.value.clone()))
|
||||
.collect();
|
||||
if let Some(name) = metric.name() {
|
||||
labels.push((Arc::from("__name__"), Arc::from(name)));
|
||||
}
|
||||
labels.sort();
|
||||
labels
|
||||
}
|
||||
|
||||
/// Prometheus `labels.Compare`: label pairs in name order, then the shorter set first.
|
||||
fn compare_labels(a: &MetricName, b: &MetricName) -> Ordering {
|
||||
labels_with_name(a).cmp(&labels_with_name(b))
|
||||
}
|
||||
|
||||
/// `github.com/facette/natsort.Compare`, which `sort_by_label` uses: digit runs compare as
|
||||
/// integers when both parse, other runs as strings.
|
||||
fn natural_less(a: &str, b: &str) -> bool {
|
||||
let (chunks_a, chunks_b) = (chunks(a), chunks(b));
|
||||
for (i, chunk_a) in chunks_a.iter().enumerate() {
|
||||
let Some(chunk_b) = chunks_b.get(i) else {
|
||||
return false;
|
||||
};
|
||||
let equal = match (chunk_a.parse::<i64>(), chunk_b.parse::<i64>()) {
|
||||
(Ok(a), Ok(b)) if a != b => return a < b,
|
||||
(Ok(_), Ok(_)) => true,
|
||||
_ => chunk_a == chunk_b,
|
||||
};
|
||||
if !equal {
|
||||
return chunk_a < chunk_b;
|
||||
}
|
||||
if i == chunks_a.len() - 1 {
|
||||
return true;
|
||||
} else if i == chunks_b.len() - 1 {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
/// Maximal runs of ASCII digits and of other characters, like the regexp `(\d+|\D+)`.
|
||||
fn chunks(s: &str) -> Vec<&str> {
|
||||
let mut chunks = Vec::new();
|
||||
let mut start = 0;
|
||||
let bytes = s.as_bytes();
|
||||
for i in 1..=bytes.len() {
|
||||
if i == bytes.len() || bytes[i].is_ascii_digit() != bytes[start].is_ascii_digit() {
|
||||
chunks.push(&s[start..i]);
|
||||
start = i;
|
||||
}
|
||||
}
|
||||
chunks
|
||||
}
|
||||
|
||||
fn single_field(schema: &SeriesSchema, operator: &str) -> Result<()> {
|
||||
if schema.fields.len() != 1 {
|
||||
return Err(DataFusionError::NotImplemented(format!(
|
||||
"{operator} over series with {} fields",
|
||||
schema.fields.len()
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Checks every series against `schema` and returns the grid length.
|
||||
fn grid_points(schema: &SeriesSchema, series: &[Series]) -> Result<usize> {
|
||||
let grid = schema.eval_grid.as_ref().ok_or_else(|| {
|
||||
DataFusionError::Execution("select operators need an evaluation grid".into())
|
||||
})?;
|
||||
for item in series {
|
||||
match &item.time {
|
||||
TimeAxis::Grid(range)
|
||||
if range.grid().spec() == grid.spec()
|
||||
&& item.fields.len() == schema.fields.len() => {}
|
||||
_ => {
|
||||
return Err(DataFusionError::Execution(
|
||||
"series does not match the select operator schema".into(),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(grid.timestamps().len())
|
||||
}
|
||||
|
||||
/// The first field's value of `series` at grid `point`.
|
||||
fn value_at(series: &Series, point: usize) -> Result<Option<f64>> {
|
||||
let TimeAxis::Grid(range) = &series.time else {
|
||||
unreachable!("checked by grid_points");
|
||||
};
|
||||
if !(range.start()..range.start() + range.len()).contains(&point) {
|
||||
return Ok(None);
|
||||
}
|
||||
match series.fields.as_slice()[0].sample(point - range.start()) {
|
||||
SampleRef::Missing => Ok(None),
|
||||
SampleRef::Float(value) => Ok(Some(value)),
|
||||
SampleRef::Histogram(_) => Err(DataFusionError::NotImplemented(
|
||||
"select operators over histogram samples".into(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn external(error: datatypes::error::Error) -> DataFusionError {
|
||||
DataFusionError::External(Box::new(error))
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,273 @@
|
||||
// Copyright 2023 Greptime Team
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::error::DataFusionError;
|
||||
use datatypes::series::samples::{FieldSpec, Series, SeriesSchema};
|
||||
use datatypes::series::time::{EvalGrid, GridSpec};
|
||||
use datatypes::series::{Label, MetricName};
|
||||
|
||||
use super::{
|
||||
Grouping, LabelMatcher, MatchOp, SortOrder, absent, absent_labels, natural_less, scalar,
|
||||
sort_instant, topk,
|
||||
};
|
||||
use crate::grid_series::GridSeriesBuilder;
|
||||
use crate::grid_series::reference::{Dense, assert_series_eq};
|
||||
|
||||
fn schema(points: u64, fields: usize) -> SeriesSchema {
|
||||
SeriesSchema {
|
||||
fields: (0..fields)
|
||||
.map(|index| FieldSpec {
|
||||
name: Arc::from(format!("f{index}")),
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
.into(),
|
||||
eval_grid: Some(Arc::new(
|
||||
EvalGrid::try_new(GridSpec::try_new(100_000, 10_000, points).unwrap()).unwrap(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Parses `name=value,...`, with `__name__` as the metric name; `-` is no labels.
|
||||
fn metric(labels: &str) -> MetricName {
|
||||
let mut name = None;
|
||||
let mut tags = Vec::new();
|
||||
for pair in labels.split(',').filter(|pair| *pair != "-") {
|
||||
let (label, value) = pair.split_once('=').unwrap();
|
||||
if label == "__name__" {
|
||||
name = Some(Arc::from(value));
|
||||
} else {
|
||||
tags.push(Label {
|
||||
name: Arc::from(label),
|
||||
value: Arc::from(value),
|
||||
});
|
||||
}
|
||||
}
|
||||
MetricName::try_new(name, tags).unwrap()
|
||||
}
|
||||
|
||||
fn dense(points: usize, values: &str) -> Dense {
|
||||
let mut dense = vec![vec![None]; points];
|
||||
for point in values.split_whitespace() {
|
||||
let (index, bits) = point.split_once(':').unwrap();
|
||||
dense[index.parse::<usize>().unwrap()][0] =
|
||||
Some(f64::from_bits(u64::from_str_radix(bits, 16).unwrap()));
|
||||
}
|
||||
dense
|
||||
}
|
||||
|
||||
fn build(schema: &SeriesSchema, metric: MetricName, dense: &Dense) -> Option<Series> {
|
||||
let mut builder = GridSeriesBuilder::try_new(Arc::new(schema.clone())).unwrap();
|
||||
for (index, row) in dense.iter().enumerate() {
|
||||
builder.push(index, row).unwrap();
|
||||
}
|
||||
builder.finish(metric).unwrap()
|
||||
}
|
||||
|
||||
fn names(list: &str) -> Vec<Arc<str>> {
|
||||
list.split(',')
|
||||
.filter(|name| *name != "-")
|
||||
.map(Arc::from)
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// `topk`/`bottomk`, `scalar` and `absent` evaluated by the Prometheus v3.15.0 engine; see the
|
||||
/// header of `prometheus_cases.txt`.
|
||||
#[test]
|
||||
fn select_operators_match_prometheus() {
|
||||
let text = include_str!("prometheus_cases.txt");
|
||||
let mut checked = 0;
|
||||
for case in text.split("end\n").filter(|case| case.contains("case ")) {
|
||||
let lines: Vec<&str> = case.lines().filter(|line| !line.starts_with('#')).collect();
|
||||
let spec: Vec<&str> = lines[0].split_whitespace().skip(1).collect();
|
||||
let points: usize = lines[1].split_whitespace().nth(3).unwrap().parse().unwrap();
|
||||
let schema = schema(points as u64, 1);
|
||||
let mut inputs = Vec::new();
|
||||
let mut expected: BTreeMap<String, Dense> = BTreeMap::new();
|
||||
for line in &lines[2..] {
|
||||
let (kind, rest) = line.split_once(' ').unwrap();
|
||||
let (labels, values) = rest.split_once(" ;").unwrap();
|
||||
match kind {
|
||||
"series" => inputs.push((metric(labels), dense(points, values))),
|
||||
"result" => {
|
||||
expected.insert(labels.to_string(), dense(points, values));
|
||||
}
|
||||
other => panic!("unexpected line {other} in {case}"),
|
||||
}
|
||||
}
|
||||
let series: Vec<Series> = inputs
|
||||
.iter()
|
||||
.filter_map(|(metric, dense)| build(&schema, metric.clone(), dense))
|
||||
.collect();
|
||||
|
||||
let actual: Vec<Series> = match spec[0] {
|
||||
"topk" => {
|
||||
let grouping = match spec[3] {
|
||||
"without" => Grouping::Without(names(spec[4])),
|
||||
"by" => Grouping::By(names(spec[4])),
|
||||
_ => Grouping::By(Vec::new()),
|
||||
};
|
||||
let k: f64 = spec[2].parse().unwrap();
|
||||
topk(k, spec[1] == "topk", &grouping, &schema, &series).unwrap()
|
||||
}
|
||||
"scalar" => vec![scalar(&schema, &series).unwrap()],
|
||||
"absent" => {
|
||||
let matchers: Vec<LabelMatcher> = spec
|
||||
.get(1)
|
||||
.filter(|matchers| **matchers != "-")
|
||||
.map(|matchers| {
|
||||
matchers
|
||||
.split(',')
|
||||
.map(|matcher| {
|
||||
let (name, op, value) = match matcher.split_once("!=") {
|
||||
Some((name, value)) => (name, MatchOp::NotEqual, value),
|
||||
None => {
|
||||
let (name, value) = matcher.split_once('=').unwrap();
|
||||
(name, MatchOp::Equal, value)
|
||||
}
|
||||
};
|
||||
let value = value.trim_matches('"');
|
||||
LabelMatcher {
|
||||
name: Arc::from(name),
|
||||
op,
|
||||
value: Arc::from(value),
|
||||
matches_empty: (op == MatchOp::Equal) == value.is_empty(),
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
let selected: Vec<Series> = series
|
||||
.iter()
|
||||
.filter(|item| {
|
||||
matchers.iter().all(|matcher| {
|
||||
let value = item.metric.get(&matcher.name).unwrap_or("");
|
||||
(value == &*matcher.value) == (matcher.op == MatchOp::Equal)
|
||||
})
|
||||
})
|
||||
.cloned()
|
||||
.collect();
|
||||
absent(&schema, &selected, absent_labels(&matchers).unwrap())
|
||||
.unwrap()
|
||||
.into_iter()
|
||||
.collect()
|
||||
}
|
||||
other => panic!("unknown case {other}"),
|
||||
};
|
||||
|
||||
let actual: BTreeMap<String, Series> = actual
|
||||
.into_iter()
|
||||
.map(|series| (render(&series.metric), series))
|
||||
.collect();
|
||||
assert_eq!(
|
||||
actual.keys().collect::<Vec<_>>(),
|
||||
expected.keys().collect::<Vec<_>>(),
|
||||
"{case}"
|
||||
);
|
||||
for (labels, dense) in &expected {
|
||||
assert_series_eq(actual.get(labels), dense, &metric(labels));
|
||||
checked += 1;
|
||||
}
|
||||
}
|
||||
assert!(checked > 200, "{checked} result series checked");
|
||||
}
|
||||
|
||||
/// The fixture's label rendering: `name=value` pairs in name order, `-` when empty.
|
||||
fn render(metric: &MetricName) -> String {
|
||||
let mut labels: Vec<(String, String)> = metric
|
||||
.tags()
|
||||
.iter()
|
||||
.map(|tag| (tag.name.to_string(), tag.value.to_string()))
|
||||
.collect();
|
||||
if let Some(name) = metric.name() {
|
||||
labels.push(("__name__".to_string(), name.to_string()));
|
||||
}
|
||||
labels.sort();
|
||||
if labels.is_empty() {
|
||||
return "-".to_string();
|
||||
}
|
||||
labels
|
||||
.iter()
|
||||
.map(|(name, value)| format!("{name}={value}"))
|
||||
.collect::<Vec<_>>()
|
||||
.join(",")
|
||||
}
|
||||
|
||||
fn instant(values: &[(&str, Option<f64>)]) -> (SeriesSchema, Vec<Series>) {
|
||||
let schema = schema(1, 1);
|
||||
let series = values
|
||||
.iter()
|
||||
.map(|(labels, value)| build(&schema, metric(labels), &vec![vec![*value]]).unwrap())
|
||||
.collect();
|
||||
(schema, series)
|
||||
}
|
||||
|
||||
fn rendered(series: &[Series]) -> Vec<String> {
|
||||
series.iter().map(|series| render(&series.metric)).collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sorts_instant_results_like_prometheus() {
|
||||
let (schema, series) = instant(&[
|
||||
("host=a", Some(2.0)),
|
||||
("host=b", Some(f64::NAN)),
|
||||
("host=c", Some(-1.0)),
|
||||
]);
|
||||
let ascending = sort_instant(&SortOrder::Asc, &schema, series.clone()).unwrap();
|
||||
assert_eq!(rendered(&ascending), ["host=c", "host=a", "host=b"]);
|
||||
let descending = sort_instant(&SortOrder::Desc, &schema, series).unwrap();
|
||||
assert_eq!(rendered(&descending), ["host=a", "host=c", "host=b"]);
|
||||
|
||||
let (schema, series) = instant(&[
|
||||
("host=x10", Some(1.0)),
|
||||
("host=x9", Some(1.0)),
|
||||
("host=x9,job=a", Some(1.0)),
|
||||
("host=x", Some(1.0)),
|
||||
]);
|
||||
let by_host = sort_instant(&SortOrder::ByLabel(names("host")), &schema, series.clone());
|
||||
assert_eq!(
|
||||
rendered(&by_host.unwrap()),
|
||||
["host=x", "host=x9", "host=x9,job=a", "host=x10"]
|
||||
);
|
||||
let by_host_desc = sort_instant(&SortOrder::ByLabelDesc(names("host")), &schema, series);
|
||||
assert_eq!(
|
||||
rendered(&by_host_desc.unwrap()),
|
||||
["host=x10", "host=x9,job=a", "host=x9", "host=x"]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn natural_order_follows_natsort() {
|
||||
assert!(natural_less("a", "a1"));
|
||||
assert!(!natural_less("a1", "a"));
|
||||
assert!(natural_less("x9", "x10"));
|
||||
assert!(!natural_less("10", "9"));
|
||||
// A digit run too long for an i64 compares as a string, like Go's strconv.Atoi failing.
|
||||
assert!(natural_less("20000000000000000000", "3"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_value_operators_over_several_fields() {
|
||||
let schema = schema(1, 2);
|
||||
assert!(matches!(
|
||||
topk(1.0, true, &Grouping::By(Vec::new()), &schema, &[]),
|
||||
Err(DataFusionError::NotImplemented(_))
|
||||
));
|
||||
assert!(matches!(
|
||||
scalar(&schema, &[]),
|
||||
Err(DataFusionError::NotImplemented(_))
|
||||
));
|
||||
}
|
||||
Reference in New Issue
Block a user