mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-11-14 06:46:56 +00:00
feat(promql): evaluate time, date and label rewriting functions over grid series (#9684)
I hereby agree to the terms of the [GreptimeDB CLA](https://github.com/GreptimeTeam/.github/blob/main/CLA.md). ## Refer to a related PR or issue link (optional) Targets `narthex`. Stacked on #9673 for `GridSeriesBuilder` and the test reference; review only `feat(promql): evaluate time, date and label rewriting functions over grid series`. ## What's changed and what's your intention? ### Problem The native path has no kernels for `time()`, the date functions (`minute`, `hour`, `day_of_month`, `day_of_week`, `day_of_year`, `days_in_month`, `month`, `year`) or `label_replace`/`label_join`. ### Change - `map_time::time` and `map_time::date_of_eval_time` produce one unlabelled series over the schema's grid from the evaluation timestamps (`ms / 1000` for `time()`, whole seconds truncated towards zero for the date functions, as Prometheus `funcTime`/`dateWrapper`). - `map_time::map_date` reads every float value as Unix seconds and drops `__name__`, `__type__` and `__unit__`. The value goes through Go's `int64(f)` as on amd64 (NaN and out-of-range values become `i64::MIN`) and a port of Go's `time` absolute-date arithmetic, including its wrapping for instants Go cannot represent, which decides the result for those inputs. All-float series are mapped chunk by chunk without copying the time axis; histogram samples are dropped and the series is trimmed again. - `map_label::label_replace` anchors the regex as `^(?s:...)$`, expands `$1`/`${name}`/`$$` with Go's `regexp.Expand` rules, and leaves unmatched series unchanged. `map_label::label_join` joins source values, an absent label joining as `""`. An empty result removes the label in both. Empty source or destination label names and invalid regexes are `Plan` errors with Prometheus' messages. - `grid_series::merge_same_labelset` follows Prometheus `mergeSeriesWithSameLabelset`: series that share a label set (empty values counting as absent) become one series, merged field by field; a field with values from two of them at one point fails with "vector cannot contain metrics with the same labelset". The label functions call it; `map_date` leaves it to the caller, as other per-series kernels that drop `__name__` do. - `promql` now depends on the workspace `regex` crate, which is already in the build graph. Review focus: - `map_time::DateFunction::apply` and `civil`: the port of Go's `absSeconds`/`absDays` arithmetic; `days_in_month` uses the proleptic Gregorian leap rule on the computed year. - `map_time::go_int64`: NaN and values outside `[-2^63, 2^63)` map to `i64::MIN`. - `merge_same_labelset`: the fast path returns the input untouched when no label sets collide; colliding groups are rebuilt row by row and fail on any shared point. - `label_replace` expands its template with Go's `regexp.Expand` rules: a reference with a leading zero (`$01`) or of more than nine digits names a group, unlike `regex::Captures::expand`. ### Non-goals - `timestamp()` (EVAL-TS), which needs sample timestamps from instant selection. - Merging histogram series that collide after a label function (`NotImplemented`). - Planner wiring. ### Constraints - Every per-series kernel that drops `__name__` (range functions, date functions, math) can leave two series with one label set. Prometheus merges or rejects them per step; on the native path that is `merge_same_labelset`, which the caller must apply after such kernels. - Regexes compile with the Rust `regex` crate, as the ordinary planner does. Perl classes (`\d`, `\w`, `\s`, `\b`) are Unicode-aware there and ASCII-only in Go's RE2, so non-ASCII label values can match differently. The dialects also differ in what compiles: `\Q...\E` is rejected, `(?x)` and repeat counts above 1000 are accepted, and `\<`, `\>` are word boundaries instead of literal characters. The rustdoc of `label_replace` lists the same differences. - `go_int64` follows amd64. On arm64 Go saturates (`NaN` becomes 0), so Prometheus' own results for non-finite inputs differ by architecture. ### Verification - `cargo nextest run -p promql`: 315 passed. - `cargo clippy -p promql --all-targets --all-features -- -D warnings`, `cargo fmt --check`, `hawkeye check`: clean. - `time_functions_match_prometheus`: 200 fixtures produced by the Prometheus v3.15.0 engine built for amd64 (`map_time/prometheus_cases.txt`): `time()` and argument-less date functions over grids from 1685 to 2254, and date functions over values including NaN, ±Inf, ±1e300, ±9.2e18, leap-day edges of 1600/1900/2000/2100/2400 and years 0 and 10000. Values compare bit for bit after `merge_same_labelset`; 6 cases expect the duplicate-labelset error. - `label_functions_match_prometheus`: 200 fixtures (`map_label/prometheus_cases.txt`) with random destinations including `__name__`, absent sources, templates with valid and invalid `$` references, anchoring-sensitive regexes and invalid ones; 25 expect errors, the rest compare label sets and value bits. - Mutation check: each of these fails a test: Go's NaN and out-of-range conversion replaced with a saturating cast or NaN-to-0; floor instead of truncating division of the evaluation time; whole-second `time()`; dropping the 400-year rule; leap day counted in January and February; hour from the Unix seconds instead of Go's absolute seconds; no leap February in `days_in_month`; leap year taken from the March-based year; keeping `__name__`; mapping only the first chunk; regex without the `(?s:...)` group or without anchors; setting an empty label instead of removing it; skipping absent `label_join` sources; skipping the merge; skipping the duplicate check; comparing empty label values as present. - `replacements_expand_like_go`: 16 templates with expectations taken from Go, including `$1²`, where Go's Unicode classes end the name before `²`. - `merge_combines_disjoint_fields_at_one_point`: different fields at one point merge, one field present twice fails. ## PR Checklist Please convert it to a draft if some of the following conditions are not met. - [x] I have written the necessary rustdoc comments. - [x] I have added the necessary unit tests and integration tests. - [ ] This PR requires documentation updates. - [x] API changes are backward compatible. - [x] Schema or data changes are backward compatible. - [ ] This PR needs to be backported to release branches, and I have added the `backport-<target>` labels (e.g. `backport-v1.3` targets `release/v1.3`).
This commit is contained in:
dennis zhuang
Generated
+1
@@ -11371,6 +11371,7 @@ dependencies = [
|
||||
"prometheus 0.14.0",
|
||||
"prost 0.14.1",
|
||||
"rand 0.9.4",
|
||||
"regex",
|
||||
"snafu 0.8.6",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
@@ -24,6 +24,7 @@ greptime-proto.workspace = true
|
||||
lazy_static.workspace = true
|
||||
prometheus.workspace = true
|
||||
prost.workspace = true
|
||||
regex.workspace = true
|
||||
snafu.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
|
||||
@@ -14,13 +14,15 @@
|
||||
|
||||
//! Building evaluated float series that cover only their result range on a shared grid.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::collections::hash_map::Entry;
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::arrow::array::Float64Array;
|
||||
use datafusion::arrow::array::builder::NullBufferBuilder;
|
||||
use datafusion::error::{DataFusionError, Result};
|
||||
use datatypes::series::MetricName;
|
||||
use datatypes::series::samples::{Samples, Series, SeriesFields, SeriesSchema};
|
||||
use datatypes::series::samples::{SampleRef, Samples, Series, SeriesFields, SeriesSchema};
|
||||
use datatypes::series::time::{EvalGrid, GridRange, TimeAxis};
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -140,6 +142,124 @@ impl GridSeriesBuilder {
|
||||
}
|
||||
}
|
||||
|
||||
type LabelSet = Vec<(Arc<str>, Arc<str>)>;
|
||||
|
||||
/// The label set Prometheus compares, which has no empty values.
|
||||
fn label_set(metric: &MetricName) -> LabelSet {
|
||||
let name = metric
|
||||
.name()
|
||||
.filter(|name| !name.is_empty())
|
||||
.map(|name| (Arc::from("__name__"), Arc::from(name)));
|
||||
name.into_iter()
|
||||
.chain(
|
||||
metric
|
||||
.tags()
|
||||
.iter()
|
||||
.filter(|tag| !tag.value.is_empty())
|
||||
.map(|tag| (tag.name.clone(), tag.value.clone())),
|
||||
)
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Merges series that share a label set into one, as Prometheus does after a function that
|
||||
/// rewrites labels or drops `__name__`; two of them holding one field at one point fail.
|
||||
pub fn merge_same_labelset(schema: &Arc<SeriesSchema>, series: Vec<Series>) -> Result<Vec<Series>> {
|
||||
let mut groups: HashMap<LabelSet, Vec<usize>> = HashMap::with_capacity(series.len());
|
||||
let mut collided = false;
|
||||
for (index, series) in series.iter().enumerate() {
|
||||
match groups.entry(label_set(&series.metric)) {
|
||||
Entry::Occupied(mut entry) => {
|
||||
entry.get_mut().push(index);
|
||||
collided = true;
|
||||
}
|
||||
Entry::Vacant(entry) => {
|
||||
entry.insert(vec![index]);
|
||||
}
|
||||
}
|
||||
}
|
||||
if !collided {
|
||||
return Ok(series);
|
||||
}
|
||||
|
||||
let mut series: Vec<Option<Series>> = series.into_iter().map(Some).collect();
|
||||
let mut output = Vec::with_capacity(groups.len());
|
||||
let mut builder = GridSeriesBuilder::try_new(schema.clone())?;
|
||||
let mut rows = Vec::new();
|
||||
let mut values = vec![None; schema.fields.len()];
|
||||
let mut merged = vec![None; schema.fields.len()];
|
||||
for (labels, members) in groups {
|
||||
if let [member] = members[..] {
|
||||
output.extend(series[member].take());
|
||||
continue;
|
||||
}
|
||||
rows.clear();
|
||||
for &member in &members {
|
||||
let member_series = series[member]
|
||||
.as_ref()
|
||||
.expect("each series is in one group");
|
||||
let TimeAxis::Grid(range) = &member_series.time else {
|
||||
return Err(DataFusionError::Execution(
|
||||
"merging requires evaluated series".into(),
|
||||
));
|
||||
};
|
||||
for row in 0..range.len() {
|
||||
if read_row(member_series, row, &mut values)? {
|
||||
rows.push((range.start() + row, member, row));
|
||||
}
|
||||
}
|
||||
}
|
||||
rows.sort_unstable_by_key(|&(point, _, _)| point);
|
||||
let mut start = 0;
|
||||
while start < rows.len() {
|
||||
let point = rows[start].0;
|
||||
let end = start + rows[start..].partition_point(|&(p, _, _)| p == point);
|
||||
// Rows of one point merge field by field; only a field present twice collides.
|
||||
merged.iter_mut().for_each(|value| *value = None);
|
||||
for &(_, member, row) in &rows[start..end] {
|
||||
read_row(series[member].as_ref().unwrap(), row, &mut values)?;
|
||||
for (slot, value) in merged.iter_mut().zip(&values) {
|
||||
if value.is_some() {
|
||||
if slot.is_some() {
|
||||
let labels: Vec<_> = labels
|
||||
.iter()
|
||||
.map(|(name, value)| format!("{name}=\"{value}\""))
|
||||
.collect();
|
||||
return Err(DataFusionError::Execution(format!(
|
||||
"vector cannot contain metrics with the same labelset {{{}}}",
|
||||
labels.join(", ")
|
||||
)));
|
||||
}
|
||||
*slot = *value;
|
||||
}
|
||||
}
|
||||
}
|
||||
builder.push(point, &merged)?;
|
||||
start = end;
|
||||
}
|
||||
let metric = series[members[0]].as_ref().unwrap().metric.clone();
|
||||
output.extend(builder.finish(metric)?);
|
||||
}
|
||||
Ok(output)
|
||||
}
|
||||
|
||||
/// Reads the float fields of `row`; returns whether any is present.
|
||||
fn read_row(series: &Series, row: usize, values: &mut [Option<f64>]) -> Result<bool> {
|
||||
let mut present = false;
|
||||
for (value, samples) in values.iter_mut().zip(series.fields.iter()) {
|
||||
*value = match samples.sample(row) {
|
||||
SampleRef::Missing => None,
|
||||
SampleRef::Float(value) => Some(value),
|
||||
SampleRef::Histogram(_) => {
|
||||
return Err(DataFusionError::NotImplemented(
|
||||
"merging histogram series with the same label set".into(),
|
||||
));
|
||||
}
|
||||
};
|
||||
present |= value.is_some();
|
||||
}
|
||||
Ok(present)
|
||||
}
|
||||
|
||||
fn external(error: datatypes::error::Error) -> DataFusionError {
|
||||
DataFusionError::External(Box::new(error))
|
||||
}
|
||||
|
||||
@@ -15,11 +15,16 @@
|
||||
//! Brute-force reference for operators producing grid series: results are compared as dense
|
||||
//! `[grid index][field]` matrices, so a kernel cannot pass by aligning local row numbers.
|
||||
|
||||
use datatypes::series::MetricName;
|
||||
use datatypes::series::samples::{SampleRef, Series};
|
||||
use datatypes::series::time::TimeAxis;
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use datatypes::series::samples::{SampleRef, Series, SeriesSchema};
|
||||
use datatypes::series::time::{EvalGrid, GridSpec, TimeAxis};
|
||||
use datatypes::series::{Label, MetricName};
|
||||
use rand::Rng;
|
||||
|
||||
use crate::grid_series::GridSeriesBuilder;
|
||||
|
||||
/// Values by absolute grid index, then by field; `None` is missing.
|
||||
pub(crate) type Dense = Vec<Vec<Option<f64>>>;
|
||||
|
||||
@@ -141,3 +146,144 @@ pub(crate) fn random_dense(rng: &mut impl Rng, grid_len: usize, fields: usize) -
|
||||
}
|
||||
dense
|
||||
}
|
||||
|
||||
/// Labels with `__name__` as an ordinary entry, the way Prometheus holds them.
|
||||
pub(crate) type Labels = BTreeMap<String, String>;
|
||||
|
||||
/// Value bits by grid index.
|
||||
pub(crate) type Points = BTreeMap<usize, u64>;
|
||||
|
||||
/// A case of a fixture written by a promql-compat `narthex-golden-*` command.
|
||||
pub(crate) struct GoldenCase {
|
||||
/// The operation and its tab-separated arguments.
|
||||
pub(crate) spec: Vec<String>,
|
||||
pub(crate) grid: Arc<EvalGrid>,
|
||||
pub(crate) inputs: Vec<(Labels, Points)>,
|
||||
/// Result series by label set, or the error message.
|
||||
pub(crate) outcome: Result<BTreeMap<Labels, Points>, String>,
|
||||
}
|
||||
|
||||
/// Parses a fixture whose label pairs are separated by `|`.
|
||||
pub(crate) fn golden_cases(text: &str) -> Vec<GoldenCase> {
|
||||
let labels = |text: &str| -> Labels {
|
||||
text.split('|')
|
||||
.filter(|pair| *pair != "-")
|
||||
.map(|pair| {
|
||||
let (name, value) = pair.split_once('=').unwrap();
|
||||
(name.to_string(), value.to_string())
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
let points = |text: &str| -> Points {
|
||||
text.split_whitespace()
|
||||
.map(|point| {
|
||||
let (index, bits) = point.split_once(':').unwrap();
|
||||
(
|
||||
index.parse().unwrap(),
|
||||
u64::from_str_radix(bits, 16).unwrap(),
|
||||
)
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
let mut cases = Vec::new();
|
||||
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 = lines[0].strip_prefix("case ").unwrap();
|
||||
let grid: Vec<i64> = lines[1]
|
||||
.split_whitespace()
|
||||
.skip(1)
|
||||
.map(|value| value.parse().unwrap())
|
||||
.collect();
|
||||
let grid = GridSpec::try_new(grid[0], grid[1], grid[2] as u64).unwrap();
|
||||
let mut inputs = Vec::new();
|
||||
let mut outcome = Ok(BTreeMap::new());
|
||||
for line in &lines[2..] {
|
||||
if let Some(message) = line.strip_prefix("error ") {
|
||||
outcome = Err(message.to_string());
|
||||
continue;
|
||||
}
|
||||
let (kind, rest) = line.split_once(' ').unwrap();
|
||||
let (series_labels, series_points) = rest.split_once(" ;").unwrap();
|
||||
let entry = (labels(series_labels), points(series_points));
|
||||
match kind {
|
||||
"series" => inputs.push(entry),
|
||||
"result" => {
|
||||
outcome.as_mut().unwrap().insert(entry.0, entry.1);
|
||||
}
|
||||
other => panic!("unexpected line {other}"),
|
||||
}
|
||||
}
|
||||
cases.push(GoldenCase {
|
||||
spec: spec.split('\t').map(str::to_string).collect(),
|
||||
grid: Arc::new(EvalGrid::try_new(grid).unwrap()),
|
||||
inputs,
|
||||
outcome,
|
||||
});
|
||||
}
|
||||
cases
|
||||
}
|
||||
|
||||
pub(crate) fn metric_of(labels: &Labels) -> MetricName {
|
||||
let mut labels = labels.clone();
|
||||
let name = labels.remove("__name__").map(Arc::from);
|
||||
MetricName::try_new(
|
||||
name,
|
||||
labels
|
||||
.into_iter()
|
||||
.map(|(name, value)| Label {
|
||||
name: Arc::from(name),
|
||||
value: Arc::from(value),
|
||||
})
|
||||
.collect(),
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
pub(crate) fn labels_of(metric: &MetricName) -> Labels {
|
||||
let mut labels: Labels = metric
|
||||
.tags()
|
||||
.iter()
|
||||
.map(|tag| (tag.name.to_string(), tag.value.to_string()))
|
||||
.collect();
|
||||
if let Some(name) = metric.name() {
|
||||
labels.insert("__name__".to_string(), name.to_string());
|
||||
}
|
||||
labels
|
||||
}
|
||||
|
||||
/// A single-field series of `schema` holding `points`; `None` without points.
|
||||
pub(crate) fn series_of(
|
||||
schema: &Arc<SeriesSchema>,
|
||||
labels: &Labels,
|
||||
points: &Points,
|
||||
) -> Option<Series> {
|
||||
let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap();
|
||||
for (&index, &bits) in points {
|
||||
builder.push(index, &[Some(f64::from_bits(bits))]).unwrap();
|
||||
}
|
||||
builder.finish(metric_of(labels)).unwrap()
|
||||
}
|
||||
|
||||
/// The points of a single-field series, any NaN as one NaN: Prometheus' NaN payloads depend on
|
||||
/// how the Go compiler orders operands.
|
||||
pub(crate) fn points_of(series: &Series) -> Points {
|
||||
let TimeAxis::Grid(range) = &series.time else {
|
||||
panic!("expected a grid series");
|
||||
};
|
||||
series.fields.as_slice()[0]
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter_map(|(row, sample)| match sample {
|
||||
SampleRef::Float(value) => Some((range.start() + row, canonical_bits(value))),
|
||||
_ => None,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub(crate) fn canonical_bits(value: f64) -> u64 {
|
||||
if value.is_nan() {
|
||||
f64::NAN.to_bits()
|
||||
} else {
|
||||
value.to_bits()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,8 +20,8 @@ use datatypes::series::{Label, MetricName};
|
||||
use rand::rngs::StdRng;
|
||||
use rand::{Rng, SeedableRng};
|
||||
|
||||
use super::GridSeriesBuilder;
|
||||
use super::reference::{assert_series_eq, random_dense};
|
||||
use super::reference::{Labels, Points, assert_series_eq, points_of, random_dense, series_of};
|
||||
use super::{GridSeriesBuilder, merge_same_labelset};
|
||||
|
||||
fn grid_schema(fields: usize, grid_len: u64) -> Arc<SeriesSchema> {
|
||||
Arc::new(SeriesSchema {
|
||||
@@ -83,3 +83,62 @@ fn builder_rejects_invalid_rows() {
|
||||
schema.eval_grid = None;
|
||||
assert!(GridSeriesBuilder::try_new(Arc::new(schema)).is_err());
|
||||
}
|
||||
|
||||
/// Label sets compare without empty values, which Prometheus does not have: a series with
|
||||
/// `host=""` and one without `host` merge.
|
||||
#[test]
|
||||
fn merge_treats_empty_label_values_as_absent() {
|
||||
let schema = grid_schema(1, 2);
|
||||
let labels = |host: Option<&str>| {
|
||||
let mut labels = Labels::from([("job".to_string(), "x".to_string())]);
|
||||
if let Some(host) = host {
|
||||
labels.insert("host".to_string(), host.to_string());
|
||||
}
|
||||
labels
|
||||
};
|
||||
let series = vec![
|
||||
series_of(
|
||||
&schema,
|
||||
&labels(Some("")),
|
||||
&Points::from([(0, 1f64.to_bits())]),
|
||||
)
|
||||
.unwrap(),
|
||||
series_of(&schema, &labels(None), &Points::from([(1, 2f64.to_bits())])).unwrap(),
|
||||
];
|
||||
let merged = merge_same_labelset(&schema, series).unwrap();
|
||||
assert_eq!(merged.len(), 1);
|
||||
assert_eq!(
|
||||
points_of(&merged[0]),
|
||||
Points::from([(0, 1f64.to_bits()), (1, 2f64.to_bits())])
|
||||
);
|
||||
}
|
||||
|
||||
/// Series of one label set merge field by field: different fields at one point combine, and
|
||||
/// only one field present in both collides.
|
||||
#[test]
|
||||
fn merge_combines_disjoint_fields_at_one_point() {
|
||||
let schema = grid_schema(2, 1);
|
||||
let series = |row: [Option<f64>; 2]| {
|
||||
let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap();
|
||||
builder.push(0, &row).unwrap();
|
||||
builder.finish(metric(0)).unwrap().unwrap()
|
||||
};
|
||||
let merged = merge_same_labelset(
|
||||
&schema,
|
||||
vec![series([Some(1.0), None]), series([None, Some(2.0)])],
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(merged.len(), 1);
|
||||
assert_series_eq(
|
||||
Some(&merged[0]),
|
||||
&vec![vec![Some(1.0), Some(2.0)]],
|
||||
&metric(0),
|
||||
);
|
||||
assert!(
|
||||
merge_same_labelset(
|
||||
&schema,
|
||||
vec![series([Some(1.0), None]), series([Some(2.0), None])]
|
||||
)
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
|
||||
@@ -17,6 +17,8 @@ pub mod extension_plan;
|
||||
pub mod functions;
|
||||
mod go_sort;
|
||||
pub mod grid_series;
|
||||
pub mod map_label;
|
||||
pub mod map_time;
|
||||
mod metrics;
|
||||
pub mod range_array;
|
||||
pub mod range_eval;
|
||||
|
||||
@@ -0,0 +1,157 @@
|
||||
// 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.
|
||||
|
||||
//! `label_replace` and `label_join` over evaluated series, following Prometheus
|
||||
//! `evalLabelReplace`, `evalLabelJoin` and `mergeSeriesWithSameLabelset`.
|
||||
|
||||
use std::sync::{Arc, LazyLock};
|
||||
|
||||
use datafusion::error::{DataFusionError, Result};
|
||||
use datatypes::series::MetricName;
|
||||
use datatypes::series::samples::{Series, SeriesSchema};
|
||||
use regex::Regex;
|
||||
|
||||
use crate::grid_series::merge_same_labelset;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
/// `label_replace(v, dst, replacement, src, regex)`: where `regex` matches the whole value of
|
||||
/// `src`, `dst` is set to `replacement` with `$1`, `${name}` and `$$` expanded.
|
||||
///
|
||||
/// `regex` compiles with the `regex` crate, whose dialect differs from Go's RE2 used by
|
||||
/// Prometheus: Perl classes (`\d`, `\w`, `\s`, `\b`) are Unicode-aware, `\Q...\E` is rejected,
|
||||
/// `(?x)` and repeat counts above 1000 are accepted, and `\<`, `\>` are word boundaries instead
|
||||
/// of literal characters.
|
||||
pub fn label_replace(
|
||||
schema: &Arc<SeriesSchema>,
|
||||
mut series: Vec<Series>,
|
||||
dst: &str,
|
||||
replacement: &str,
|
||||
src: &str,
|
||||
regex: &str,
|
||||
) -> Result<Vec<Series>> {
|
||||
let compiled = Regex::new(&format!("^(?s:{regex})$")).map_err(|_| {
|
||||
DataFusionError::Plan(format!(
|
||||
"invalid regular expression in label_replace(): {regex}"
|
||||
))
|
||||
})?;
|
||||
if dst.is_empty() {
|
||||
return Err(DataFusionError::Plan(
|
||||
"invalid destination label name in label_replace(): ".into(),
|
||||
));
|
||||
}
|
||||
let mut value = String::new();
|
||||
for series in &mut series {
|
||||
let Some(captures) = compiled.captures(series.metric.get(src).unwrap_or_default()) else {
|
||||
continue;
|
||||
};
|
||||
value.clear();
|
||||
expand(&compiled, &captures, replacement, &mut value);
|
||||
set_label(&mut series.metric, dst, &value)?;
|
||||
}
|
||||
merge_same_labelset(schema, series)
|
||||
}
|
||||
|
||||
/// Go `regexp.Expand` reads a group name as Unicode letters (L), decimal digits (Nd) and `_`,
|
||||
/// narrower than `char::is_alphanumeric`.
|
||||
static GROUP_NAME: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"^[\p{L}\p{Nd}_]*").unwrap());
|
||||
|
||||
/// Go `regexp.Expand`, whose template syntax differs from `regex::Captures::expand`: a numeric
|
||||
/// reference with a leading zero, like `$01`, or of more than nine digits is a group name.
|
||||
fn expand(regex: &Regex, captures: ®ex::Captures<'_>, template: &str, out: &mut String) {
|
||||
let mut rest = template;
|
||||
while let Some(dollar) = rest.find('$') {
|
||||
out.push_str(&rest[..dollar]);
|
||||
rest = &rest[dollar + 1..];
|
||||
if let Some(after) = rest.strip_prefix('$') {
|
||||
out.push('$');
|
||||
rest = after;
|
||||
continue;
|
||||
}
|
||||
let (braced, body) = match rest.strip_prefix('{') {
|
||||
Some(body) => (true, body),
|
||||
None => (false, rest),
|
||||
};
|
||||
let len = GROUP_NAME.find(body).map_or(0, |name| name.end());
|
||||
let name = &body[..len];
|
||||
let closed = !braced || body[len..].starts_with('}');
|
||||
if name.is_empty() || !closed {
|
||||
// Malformed; Go keeps the `$` as text.
|
||||
out.push('$');
|
||||
continue;
|
||||
}
|
||||
rest = &body[len + usize::from(braced)..];
|
||||
let number = (name.bytes().all(|b| b.is_ascii_digit())
|
||||
&& !(name.len() > 1 && name.starts_with('0')))
|
||||
.then(|| name.parse::<usize>().ok())
|
||||
.flatten()
|
||||
.filter(|_| name.len() <= 9);
|
||||
let group = match number {
|
||||
Some(number) => captures.get(number),
|
||||
None => regex
|
||||
.capture_names()
|
||||
.position(|group| group == Some(name))
|
||||
.and_then(|index| captures.get(index)),
|
||||
};
|
||||
if let Some(group) = group {
|
||||
out.push_str(group.as_str());
|
||||
}
|
||||
}
|
||||
out.push_str(rest);
|
||||
}
|
||||
|
||||
/// `label_join(v, dst, separator, src...)`: `dst` is set to the values of `src` joined by
|
||||
/// `separator`, an absent label joining as the empty string.
|
||||
pub fn label_join(
|
||||
schema: &Arc<SeriesSchema>,
|
||||
mut series: Vec<Series>,
|
||||
dst: &str,
|
||||
separator: &str,
|
||||
src: &[&str],
|
||||
) -> Result<Vec<Series>> {
|
||||
if let Some(src) = src.iter().find(|src| src.is_empty()) {
|
||||
return Err(DataFusionError::Plan(format!(
|
||||
"invalid source label name in label_join(): {src}"
|
||||
)));
|
||||
}
|
||||
if dst.is_empty() {
|
||||
return Err(DataFusionError::Plan(
|
||||
"invalid destination label name in label_join(): ".into(),
|
||||
));
|
||||
}
|
||||
let mut value = String::new();
|
||||
for series in &mut series {
|
||||
value.clear();
|
||||
for (index, src) in src.iter().enumerate() {
|
||||
if index > 0 {
|
||||
value.push_str(separator);
|
||||
}
|
||||
value.push_str(series.metric.get(src).unwrap_or_default());
|
||||
}
|
||||
set_label(&mut series.metric, dst, &value)?;
|
||||
}
|
||||
merge_same_labelset(schema, series)
|
||||
}
|
||||
|
||||
/// An empty value removes the label, as Prometheus label sets have no empty values.
|
||||
fn set_label(metric: &mut MetricName, name: &str, value: &str) -> Result<()> {
|
||||
if value.is_empty() {
|
||||
metric.remove(name);
|
||||
return Ok(());
|
||||
}
|
||||
metric
|
||||
.set(name, Arc::from(value))
|
||||
.map_err(|error| DataFusionError::External(Box::new(error)))
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,115 @@
|
||||
// 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 datatypes::series::samples::{FieldSpec, SeriesSchema};
|
||||
|
||||
use super::{label_join, label_replace};
|
||||
use crate::grid_series::reference::{
|
||||
canonical_bits, golden_cases, labels_of, points_of, series_of,
|
||||
};
|
||||
|
||||
#[test]
|
||||
fn label_functions_match_prometheus() {
|
||||
let cases = golden_cases(include_str!("prometheus_cases.txt"));
|
||||
assert_eq!(cases.len(), 200);
|
||||
let (mut matched, mut merged, mut failed) = (0, 0, 0);
|
||||
for case in cases {
|
||||
let schema = Arc::new(SeriesSchema {
|
||||
fields: vec![FieldSpec {
|
||||
name: Arc::from("value"),
|
||||
}]
|
||||
.into(),
|
||||
eval_grid: Some(case.grid.clone()),
|
||||
});
|
||||
let input = case
|
||||
.inputs
|
||||
.iter()
|
||||
.filter_map(|(labels, points)| series_of(&schema, labels, points))
|
||||
.collect::<Vec<_>>();
|
||||
let inputs = input.len();
|
||||
let spec: Vec<&str> = case.spec.iter().map(String::as_str).collect();
|
||||
let actual = match spec[0] {
|
||||
"label_replace" => label_replace(&schema, input, spec[1], spec[2], spec[3], spec[4]),
|
||||
"label_join" => label_join(&schema, input, spec[1], spec[2], &spec[3..]),
|
||||
other => panic!("unknown case {other}"),
|
||||
};
|
||||
match (actual, case.outcome) {
|
||||
(Ok(actual), Ok(expected)) => {
|
||||
if expected.len() < inputs {
|
||||
merged += 1;
|
||||
}
|
||||
let actual: BTreeMap<_, _> = actual
|
||||
.iter()
|
||||
.map(|series| (labels_of(&series.metric), points_of(series)))
|
||||
.collect();
|
||||
let expected: BTreeMap<_, _> = expected
|
||||
.into_iter()
|
||||
.map(|(labels, points)| {
|
||||
let points = points
|
||||
.into_iter()
|
||||
.map(|(index, bits)| (index, canonical_bits(f64::from_bits(bits))))
|
||||
.collect();
|
||||
(labels, points)
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(actual, expected, "case {spec:?}");
|
||||
matched += 1;
|
||||
}
|
||||
(Err(actual), Err(expected)) => {
|
||||
assert!(
|
||||
actual.to_string().contains(&expected),
|
||||
"case {spec:?}: {actual} is not {expected}"
|
||||
);
|
||||
failed += 1;
|
||||
}
|
||||
(actual, expected) => panic!("case {spec:?}: {actual:?} is not {expected:?}"),
|
||||
}
|
||||
}
|
||||
assert!(
|
||||
matched > 150 && merged > 2 && failed > 20,
|
||||
"{matched} matched, {merged} merged, {failed} failed"
|
||||
);
|
||||
}
|
||||
|
||||
/// Replacement templates expand as Go's `regexp.Expand` does; expectations come from Go.
|
||||
#[test]
|
||||
fn replacements_expand_like_go() {
|
||||
let regex = regex::Regex::new("^(?s:(?P<word>[a-z]+)-([0-9]+))$").unwrap();
|
||||
let captures = regex.captures("abc-42").unwrap();
|
||||
for (template, expected) in [
|
||||
("$01", ""),
|
||||
("${01}", ""),
|
||||
("$1", "abc"),
|
||||
("${1}x", "abcx"),
|
||||
("$1x", ""),
|
||||
("$$", "$"),
|
||||
("$", "$"),
|
||||
("${1", "${1"),
|
||||
("$word", "abc"),
|
||||
("${word}!", "abc!"),
|
||||
("$2$1", "42abc"),
|
||||
("$-", "$-"),
|
||||
("$0", "abc-42"),
|
||||
("$9", ""),
|
||||
("$1234567890", ""),
|
||||
("$1²", "abc²"),
|
||||
] {
|
||||
let mut out = String::new();
|
||||
super::expand(®ex, &captures, template, &mut out);
|
||||
assert_eq!(out, expected, "{template}");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,219 @@
|
||||
// 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.
|
||||
|
||||
//! `time()` and the date functions over evaluated series, following Prometheus `funcTime` and
|
||||
//! `dateWrapper`.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::arrow::array::Float64Array;
|
||||
use datafusion::error::{DataFusionError, Result};
|
||||
use datatypes::series::MetricName;
|
||||
use datatypes::series::samples::{
|
||||
DenseSamplesRef, SampleRef, Samples, Series, SeriesFields, SeriesSchema,
|
||||
};
|
||||
use datatypes::series::time::{GridRange, TimeAxis};
|
||||
|
||||
use crate::grid_series::GridSeriesBuilder;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
/// A PromQL date function: `minute(v)`, `year()` and so on.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum DateFunction {
|
||||
DayOfMonth,
|
||||
DayOfWeek,
|
||||
DayOfYear,
|
||||
DaysInMonth,
|
||||
Hour,
|
||||
Minute,
|
||||
Month,
|
||||
Year,
|
||||
}
|
||||
|
||||
/// `time()`: the evaluation time in seconds at every point of the schema's grid, unlabelled.
|
||||
pub fn time(schema: &SeriesSchema) -> Result<Series> {
|
||||
eval_time_series(schema, |ms| ms as f64 / 1000.0)
|
||||
}
|
||||
|
||||
/// A date function without argument, which reads the evaluation time: one unlabelled series
|
||||
/// over the whole grid.
|
||||
pub fn date_of_eval_time(function: DateFunction, schema: &SeriesSchema) -> Result<Series> {
|
||||
// Prometheus truncates the timestamp to whole seconds towards zero.
|
||||
eval_time_series(schema, |ms| function.apply(ms / 1000))
|
||||
}
|
||||
|
||||
/// A date function over `series`: each float value is read as Unix seconds and replaced by the
|
||||
/// date component, and the metric name is dropped. Missing values stay missing; histogram
|
||||
/// samples are dropped, so the result is `None` when no float remains.
|
||||
///
|
||||
/// Series that share a label set once their names are dropped must then go through
|
||||
/// [`merge_same_labelset`](crate::grid_series::merge_same_labelset).
|
||||
pub fn map_date(
|
||||
function: DateFunction,
|
||||
schema: &Arc<SeriesSchema>,
|
||||
series: &Series,
|
||||
) -> Result<Option<Series>> {
|
||||
let TimeAxis::Grid(range) = &series.time else {
|
||||
return Err(DataFusionError::Execution(
|
||||
"date functions require an evaluated series".into(),
|
||||
));
|
||||
};
|
||||
let mut metric = series.metric.clone();
|
||||
metric.remove_all(&METADATA_LABELS);
|
||||
let map = |seconds: f64| function.apply(go_int64(seconds));
|
||||
|
||||
let all_floats = series.fields.iter().all(|samples| {
|
||||
samples
|
||||
.segments()
|
||||
.iter()
|
||||
.all(|segment| matches!(segment.dense(), Some(DenseSamplesRef::Float(_))))
|
||||
});
|
||||
if all_floats {
|
||||
let fields = series
|
||||
.fields
|
||||
.iter()
|
||||
.map(|samples| {
|
||||
let segments = samples
|
||||
.segments()
|
||||
.iter()
|
||||
.map(|segment| {
|
||||
let Some(DenseSamplesRef::Float(floats)) = segment.dense() else {
|
||||
unreachable!("checked above");
|
||||
};
|
||||
Samples::from_float(floats.unary(map))
|
||||
})
|
||||
.collect();
|
||||
Samples::try_from_chunks(segments).map_err(external)
|
||||
})
|
||||
.collect::<Result<Vec<_>>>()?;
|
||||
let fields = SeriesFields::try_from_vec(fields).map_err(external)?;
|
||||
return Series::try_new(schema, metric, series.time.clone(), fields)
|
||||
.map(Some)
|
||||
.map_err(external);
|
||||
}
|
||||
|
||||
// Histogram samples leave holes, which may move the ends of the series.
|
||||
let mut builder = GridSeriesBuilder::try_new(schema.clone())?;
|
||||
let mut values = vec![None; series.fields.len()];
|
||||
for row in 0..range.len() {
|
||||
for (value, samples) in values.iter_mut().zip(series.fields.iter()) {
|
||||
*value = match samples.sample(row) {
|
||||
SampleRef::Float(seconds) => Some(map(seconds)),
|
||||
SampleRef::Missing | SampleRef::Histogram(_) => None,
|
||||
};
|
||||
}
|
||||
builder.push(range.start() + row, &values)?;
|
||||
}
|
||||
builder.finish(metric)
|
||||
}
|
||||
|
||||
fn eval_time_series(schema: &SeriesSchema, value: impl Fn(i64) -> f64) -> Result<Series> {
|
||||
let grid = schema.eval_grid.clone().ok_or_else(|| {
|
||||
DataFusionError::Execution("time functions require an evaluation grid".into())
|
||||
})?;
|
||||
let values: Float64Array = grid.timestamps().iter().map(|&ms| value(ms)).collect();
|
||||
let range = GridRange::try_new(grid.clone(), 0, values.len()).map_err(external)?;
|
||||
let fields = vec![Samples::from_float(values); schema.fields.len()];
|
||||
Series::try_new(
|
||||
schema,
|
||||
MetricName::try_new(None, Vec::new()).map_err(external)?,
|
||||
TimeAxis::Grid(range),
|
||||
SeriesFields::try_from_vec(fields).map_err(external)?,
|
||||
)
|
||||
.map_err(external)
|
||||
}
|
||||
|
||||
/// Go's `int64(f)` on amd64: NaN and values out of range become `i64::MIN`.
|
||||
fn go_int64(value: f64) -> i64 {
|
||||
if (-9_223_372_036_854_775_808.0..9_223_372_036_854_775_808.0).contains(&value) {
|
||||
value as i64
|
||||
} else {
|
||||
i64::MIN
|
||||
}
|
||||
}
|
||||
|
||||
/// Prometheus `schema.IsMetadataLabel`: the labels removed with the metric name.
|
||||
const METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"];
|
||||
|
||||
const SECONDS_PER_DAY: u64 = 86_400;
|
||||
const MARCH_THRU_DECEMBER: u32 = 306;
|
||||
const ABSOLUTE_YEARS: u64 = 292_277_022_400;
|
||||
/// Go's `unixToInternal + internalToAbsolute`.
|
||||
const UNIX_TO_ABSOLUTE: i64 = (1969 * 365 + 1969 / 4 - 1969 / 100 + 1969 / 400) * 86_400
|
||||
+ (ABSOLUTE_YEARS as i64 / 400 * 146_097 + MARCH_THRU_DECEMBER as i64) * 86_400;
|
||||
|
||||
impl DateFunction {
|
||||
/// The component of Unix second `seconds` in UTC, computed the way Go's `time` package does,
|
||||
/// including its wrapping arithmetic for instants out of its range.
|
||||
fn apply(self, seconds: i64) -> f64 {
|
||||
let abs = seconds.wrapping_add(UNIX_TO_ABSOLUTE) as u64;
|
||||
let days = abs / SECONDS_PER_DAY;
|
||||
let value = match self {
|
||||
Self::Hour => (abs % SECONDS_PER_DAY / 3600) as i64,
|
||||
Self::Minute => (abs % 3600 / 60) as i64,
|
||||
Self::DayOfWeek => ((days + 3) % 7) as i64,
|
||||
Self::Year => civil(days).year,
|
||||
Self::Month => civil(days).month as i64,
|
||||
Self::DayOfMonth => civil(days).day as i64,
|
||||
Self::DayOfYear => civil(days).year_day,
|
||||
Self::DaysInMonth => {
|
||||
let Civil { year, month, .. } = civil(days);
|
||||
let leap = year % 4 == 0 && (year % 100 != 0 || year % 400 == 0);
|
||||
match month {
|
||||
2 if leap => 29,
|
||||
2 => 28,
|
||||
4 | 6 | 9 | 11 => 30,
|
||||
_ => 31,
|
||||
}
|
||||
}
|
||||
};
|
||||
value as f64
|
||||
}
|
||||
}
|
||||
|
||||
struct Civil {
|
||||
year: i64,
|
||||
month: u32,
|
||||
day: u32,
|
||||
year_day: i64,
|
||||
}
|
||||
|
||||
/// Go's `absDays.split`, `date` and `yearYday`: days since March 1 of year `-ABSOLUTE_YEARS`.
|
||||
fn civil(days: u64) -> Civil {
|
||||
let d = days.wrapping_mul(4).wrapping_add(3);
|
||||
let century = d / 146_097;
|
||||
let cd = (d % 146_097) as u32 | 3;
|
||||
let product = 2_939_745u64 * cd as u64;
|
||||
let cyear = (product >> 32) as u32;
|
||||
let ayday = product as u32 / 2_939_745 / 4;
|
||||
|
||||
let m = 2141 * ayday + 197_913;
|
||||
let jan_feb = u32::from(ayday >= MARCH_THRU_DECEMBER);
|
||||
let leap = cyear.is_multiple_of(4) && (cyear != 0 || century.is_multiple_of(4));
|
||||
Civil {
|
||||
year: century.wrapping_mul(100).wrapping_sub(ABSOLUTE_YEARS) as i64
|
||||
+ cyear as i64
|
||||
+ jan_feb as i64,
|
||||
month: (m >> 16) - jan_feb * 12,
|
||||
day: 1 + (m & 0xFFFF) / 2141,
|
||||
year_day: ayday as i64 + (1 + 31 + 28) + i64::from(leap && jan_feb == 0)
|
||||
- 365 * jan_feb as i64,
|
||||
}
|
||||
}
|
||||
|
||||
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,175 @@
|
||||
// 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::arrow::array::{Float64Array, UInt32Array};
|
||||
use datatypes::series::histogram::{
|
||||
BucketSideView, CounterResetHint, HistogramBatchBuilder, HistogramView,
|
||||
};
|
||||
use datatypes::series::samples::{FieldSpec, Samples, Series, SeriesFields, SeriesSchema};
|
||||
use datatypes::series::time::{EvalGrid, GridRange, GridSpec, TimeAxis};
|
||||
|
||||
use super::{DateFunction, date_of_eval_time, map_date, time};
|
||||
use crate::grid_series::merge_same_labelset;
|
||||
use crate::grid_series::reference::{
|
||||
Labels, canonical_bits, golden_cases, labels_of, metric_of, points_of, series_of,
|
||||
};
|
||||
|
||||
fn schema(grid: Arc<EvalGrid>) -> Arc<SeriesSchema> {
|
||||
Arc::new(SeriesSchema {
|
||||
fields: vec![FieldSpec {
|
||||
name: Arc::from("value"),
|
||||
}]
|
||||
.into(),
|
||||
eval_grid: Some(grid),
|
||||
})
|
||||
}
|
||||
|
||||
fn function(name: &str) -> DateFunction {
|
||||
match name {
|
||||
"day_of_month" => DateFunction::DayOfMonth,
|
||||
"day_of_week" => DateFunction::DayOfWeek,
|
||||
"day_of_year" => DateFunction::DayOfYear,
|
||||
"days_in_month" => DateFunction::DaysInMonth,
|
||||
"hour" => DateFunction::Hour,
|
||||
"minute" => DateFunction::Minute,
|
||||
"month" => DateFunction::Month,
|
||||
"year" => DateFunction::Year,
|
||||
other => panic!("unknown date function {other}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn time_functions_match_prometheus() {
|
||||
let cases = golden_cases(include_str!("prometheus_cases.txt"));
|
||||
assert_eq!(cases.len(), 200);
|
||||
for case in cases {
|
||||
let schema = schema(case.grid.clone());
|
||||
let actual: Vec<Series> = match (case.spec[0].as_str(), case.spec[1].as_str()) {
|
||||
("eval_time", "time") => vec![time(&schema).unwrap()],
|
||||
("eval_time", name) => vec![date_of_eval_time(function(name), &schema).unwrap()],
|
||||
("date", name) => case
|
||||
.inputs
|
||||
.iter()
|
||||
.filter_map(|(labels, points)| series_of(&schema, labels, points))
|
||||
.filter_map(|series| map_date(function(name), &schema, &series).unwrap())
|
||||
.collect::<Vec<_>>(),
|
||||
(kind, _) => panic!("unknown case {kind}"),
|
||||
};
|
||||
// Dropping `__name__` can give two series one label set.
|
||||
let actual = match (merge_same_labelset(&schema, actual), case.outcome) {
|
||||
(Ok(actual), Ok(expected)) => (actual, expected),
|
||||
(Err(actual), Err(expected)) => {
|
||||
assert!(actual.to_string().contains(&expected), "{actual}");
|
||||
continue;
|
||||
}
|
||||
(actual, expected) => panic!("case {:?}: {actual:?} is not {expected:?}", case.spec),
|
||||
};
|
||||
let (actual, expected) = actual;
|
||||
let actual: BTreeMap<_, _> = actual
|
||||
.iter()
|
||||
.map(|series| (labels_of(&series.metric), points_of(series)))
|
||||
.collect();
|
||||
let expected: BTreeMap<_, _> = expected
|
||||
.into_iter()
|
||||
.map(|(labels, points)| {
|
||||
let points = points
|
||||
.into_iter()
|
||||
.map(|(index, bits)| (index, canonical_bits(f64::from_bits(bits))))
|
||||
.collect();
|
||||
(labels, points)
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(actual, expected, "case {:?}", case.spec);
|
||||
}
|
||||
}
|
||||
|
||||
/// Histogram samples are dropped and the series trimmed to the remaining floats; chunked floats
|
||||
/// are mapped chunk by chunk.
|
||||
#[test]
|
||||
fn map_date_drops_histograms_and_maps_chunks() {
|
||||
let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 1000, 4).unwrap()).unwrap());
|
||||
let schema = schema(grid.clone());
|
||||
let range = GridRange::try_new(grid, 1, 3).unwrap();
|
||||
let series = |fields: Samples| {
|
||||
Series::try_new(
|
||||
&schema,
|
||||
metric_of(
|
||||
&[
|
||||
("__name__", "m"),
|
||||
("__type__", "gauge"),
|
||||
("__unit__", "seconds"),
|
||||
("host", "a"),
|
||||
]
|
||||
.map(|(n, v)| (n.into(), v.into()))
|
||||
.into(),
|
||||
),
|
||||
TimeAxis::Grid(range.clone()),
|
||||
SeriesFields::one(fields),
|
||||
)
|
||||
.unwrap()
|
||||
};
|
||||
let floats = |values: &[f64]| Samples::from_float(Float64Array::from(values.to_vec()));
|
||||
|
||||
let empty = BucketSideView {
|
||||
span_offsets: &[],
|
||||
span_lengths: &[],
|
||||
buckets: &[],
|
||||
};
|
||||
let mut histograms = HistogramBatchBuilder::new();
|
||||
histograms
|
||||
.push(HistogramView {
|
||||
schema: 0,
|
||||
zero_threshold: 0.0,
|
||||
sum: 1.0,
|
||||
count: 1.0,
|
||||
zero_count: 1.0,
|
||||
reset_hint: CounterResetHint::Unknown,
|
||||
start_timestamp: None,
|
||||
custom_values: &[],
|
||||
positive: empty,
|
||||
negative: empty,
|
||||
})
|
||||
.unwrap();
|
||||
let histogram =
|
||||
Samples::try_from_histogram(UInt32Array::from(vec![0]), histograms.finish().unwrap())
|
||||
.unwrap();
|
||||
let mixed = Samples::try_from_chunks(vec![histogram, floats(&[3600.0, 7200.0])]).unwrap();
|
||||
let hours = map_date(DateFunction::Hour, &schema, &series(mixed))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
points_of(&hours),
|
||||
BTreeMap::from([(2, 1f64.to_bits()), (3, 2f64.to_bits())])
|
||||
);
|
||||
assert_eq!(
|
||||
labels_of(&hours.metric),
|
||||
Labels::from([("host".into(), "a".into())])
|
||||
);
|
||||
|
||||
let chunked = Samples::try_from_chunks(vec![floats(&[60.0]), floats(&[120.0, 180.0])]).unwrap();
|
||||
let minutes = map_date(DateFunction::Minute, &schema, &series(chunked))
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
points_of(&minutes),
|
||||
BTreeMap::from([
|
||||
(1, 1f64.to_bits()),
|
||||
(2, 2f64.to_bits()),
|
||||
(3, 3f64.to_bits())
|
||||
])
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user