rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 06:46:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(promql): evaluate classic and native histogram functions over grid series (#9685)

Browse Source 
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 #9684 for `merge_same_labelset` and the
golden fixture helpers; review only the two commits `feat(promql):
evaluate classic histogram_quantile and histogram_fraction over grid
series` and `feat(promql): evaluate native histogram functions over grid
series`.

## What's changed and what's your intention?

### Problem

The native path has no kernels for the histogram functions:
`histogram_quantile` and `histogram_fraction` over classic
(`le`-bucketed) and native histograms, and `histogram_count`,
`histogram_sum`, `histogram_avg`, `histogram_stddev` and
`histogram_stdvar` over native histograms.

### Change

`histogram_eval::evaluate_histograms` takes the evaluated input series
and a `HistogramFunction` (`Quantile(q)` or `Fraction { lower, upper }`,
constants as in the ordinary planner) and follows Prometheus
`funcHistogramQuantile`, `funcHistogramFraction` and `resetHistograms`:

- Float series with the same labels other than `le` form a classic
histogram; a series whose `le` does not parse is ignored. Parsing
follows Go's `strconv.ParseFloat`: finite literals out of the f64 range
(`1e400`) and a signed NaN are rejected, and `_` between digits is
accepted.
- Native histogram samples are evaluated one by one. At a point where a
native histogram and classic buckets of the same labels (`le` aside) are
both present, neither is evaluated.
- Each field is evaluated on its own. Results drop `le`, `__name__`,
`__type__` and `__unit__`; series that then share a label set go through
`merge_same_labelset`.

Classic buckets (`histogram_eval::classic`) port `BucketQuantile` and
`BucketFraction`. At each point the present buckets are sorted with
`go_sort::sort_by` and Go's `<` comparator, so +0/-0 and NaN bounds keep
Go's order, and buckets with equal bounds are coalesced. For quantiles,
counts within a relative 1e-12 of the previous count or below it are
raised to it. The quantile bucket is found with Go's `sort.Search`
bisection, so NaN counts select the same bucket.

Native histograms (`histogram_eval::native`) port `HistogramQuantile`,
`HistogramFraction` and `histogramVariance`. The port includes:

- the bucket order and zero-bucket clipping of
`AllBucketIterator`/`AllReverseBucketIterator`;
- bucket bounds `2^(idx / 2^schema)` from the standard library, with
`f64::MAX` for the last bucket of finite values as in Go's `getBound`;

This covers the reverse iteration for `q >= 0.5`, the count clamp,
custom-bucket edge buckets, and the trailing walk that Prometheus runs
for NaN observations, which leaves the last bucket in place for
interpolation.

`histogram_eval::map_native_histogram` applies
`NativeHistogramFunction::{Count, Sum, Avg, StdDev, StdVar}` to the
histogram samples of one series, dropping float samples and `__name__`;
colliding label sets are left to `merge_same_labelset`, as for other
per-series kernels.

Review focus:

- `classic::bucket_quantile`: the `q` checks come before the `+Inf`
check, the `b == 0 && upper_bound <= 0` branch, and the interpolation
`start + (end - start) * (rank / count)`.
- `classic::ensure_monotonic`: `prev` is not advanced when a count is
raised.
- `native::all_buckets` and `native::bound`: iteration order, clipping
to the zero threshold, and the last finite bucket (`f64::MAX`).
- `native::quantile`: the last visited bucket (empty or not) is the one
interpolated, including after the NaN-observation walk.
- `evaluate_histograms`: the classic/native conflict is decided per
point and field.

### Non-goals

- Native histogram `sum`/`avg` aggregation, rate functions and
arithmetic.
- Non-constant `q`, `lower` and `upper`.
- Planner wiring.

### Constraints

- The ordinary `HistogramFold::evaluate_row` differs from Prometheus for
classic histograms: it errors when the last bucket is not `+Inf`, has no
small-delta tolerance, replaces non-finite counts with the previous
count, and interpolates as `lower + (upper - lower) / count * rank`. The
ordinary native quantile
(`common_query::native_histogram::quantile_histogram_view`) has no
reverse iteration for `q >= 0.5` and derives bounds with `powf` instead
of Go's table. This PR follows Prometheus 3.15.
- `le` values in Go-only float syntax (hex floats such as `0x1p-2`) do
not parse with Rust's `f64::from_str` and are ignored.

### Verification

- `cargo nextest run -p promql`: 318 passed.
- `cargo clippy -p promql --all-targets --all-features -- -D warnings`,
`cargo fmt --check`, `hawkeye check`: clean.
- `classic_histograms_match_prometheus`: 300 fixtures produced by the
Prometheus v3.15.0 engine built for amd64
(`histogram_eval/classic_cases.txt`).
- Bucket layouts: non-positive bounds, a bound written twice
(`1`/`1.0`), no `+Inf`, only `+Inf`, unparsable, out-of-range or missing
`le`, and the `inf` spelling.
- Counts: monotonic, non-monotonic, around the 1e-12 tolerance, NaN,
`+Inf`, negative, zero and subnormal.
- Inputs are passed in reverse order. 32 cases expect the
duplicate-labelset error.
- `native_histograms_match_prometheus`: 420 fixtures
(`histogram_eval/native_cases.txt`, 60 per function).
- Histograms: schemas -4, -1, 0, 1, 3, 8 and custom buckets, with gaps,
zero-length spans, empty buckets, zero thresholds from 0 to 0.5, NaN
sums with uncounted observations, and counts below the bucket total.
- Inputs mix floats and histograms, and include classic buckets of the
same labels.
  - Inputs are read back from the engine, so they carry its span layout.
- Values in both fixtures compare within promql-compat's relative
tolerance (1e-10, 1e-6 for `histogram_stddev`/`histogram_stdvar`), NaN
as any NaN.
- Mutation check, classic: each of these fails a test:
- no coalescing (quantile or fraction), no forced monotonicity, no
small-delta tolerance, an absolute tolerance at zero;
- a linear search, no `+Inf` check, `q` checks after it, no non-positive
first-bucket branch, the ordinary interpolation order;
- accepting `1e400`, keeping `__name__`, a zero lower bound for a
non-positive first fraction bucket, interpolating inside a `-Inf`-wide
bucket, no merge, sorting with NaN bounds last.
- Mutation check, native: each of these fails a test:
- always iterating forward, no NaN-observation walk, no zero-threshold
clipping, no clamp of the count;
  - shifted bucket bounds, no `f64::MAX` for the last finite bucket;
  - a plain variance sum, a geometric midpoint for custom buckets;
- the histogram count as the fraction total under NaN observations,
linear interpolation in exponential buckets;
  - no custom `-Inf` branch, no zero-bucket side adjustment;
- evaluating conflicting classic and native histograms, keeping
`__name__`, no zero bucket in reverse order.
- `histogram_functions_drop_metadata_labels`: `histogram_quantile` drops
all metadata labels, `histogram_count` only `__name__`.
- `classic_bounds_keep_the_first_signed_zero`: bounds +0 and -0 compare
equal and keep input order, so `1 / histogram_quantile(0, ...)` over
`le` "+Inf", "0", "-0" is +Inf.

## 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
2026-10-18 01:03:28 +08:00
committed by GitHub
parent 12e07cf727 041a48d456
commit 0d2f1ff6cd
9 changed files with 8178 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/src/grid_series.rs
+2 -2
View File
@@ -142,10 +142,10 @@ impl GridSeriesBuilder {
}
}
type LabelSet = Vec<(Arc<str>, Arc<str>)>;
pub(crate) type LabelSet = Vec<(Arc<str>, Arc<str>)>;
/// The label set Prometheus compares, which has no empty values.
fn label_set(metric: &MetricName) -> LabelSet {
pub(crate) fn label_set(metric: &MetricName) -> LabelSet {
let name = metric
.name()
.filter(|name| !name.is_empty())
src/promql/src/grid_series/reference.rs
+23 -12
View File
@@ -159,6 +159,8 @@ pub(crate) struct GoldenCase {
pub(crate) spec: Vec<String>,
pub(crate) grid: Arc<EvalGrid>,
pub(crate) inputs: Vec<(Labels, Points)>,
/// Native histogram points of each input, as written by the generator.
pub(crate) histograms: Vec<BTreeMap<usize, String>>,
/// Result series by label set, or the error message.
pub(crate) outcome: Result<BTreeMap<Labels, Points>, String>,
}
@@ -174,16 +176,19 @@ pub(crate) fn golden_cases(text: &str) -> Vec<GoldenCase> {
})
.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 points = |text: &str| -> (Points, BTreeMap<usize, String>) {
let mut floats = Points::new();
let mut histograms = BTreeMap::new();
for point in text.split_whitespace() {
let (index, value) = point.split_once(':').unwrap();
let index = index.parse().unwrap();
if let Some(histogram) = value.strip_prefix('H') {
histograms.insert(index, histogram.to_string());
} else {
floats.insert(index, u64::from_str_radix(value, 16).unwrap());
}
}
(floats, histograms)
};
let mut cases = Vec::new();
for case in text.split("end\n").filter(|case| case.contains("case ")) {
@@ -196,6 +201,7 @@ pub(crate) fn golden_cases(text: &str) -> Vec<GoldenCase> {
.collect();
let grid = GridSpec::try_new(grid[0], grid[1], grid[2] as u64).unwrap();
let mut inputs = Vec::new();
let mut histograms = Vec::new();
let mut outcome = Ok(BTreeMap::new());
for line in &lines[2..] {
if let Some(message) = line.strip_prefix("error ") {
@@ -204,9 +210,13 @@ pub(crate) fn golden_cases(text: &str) -> Vec<GoldenCase> {
}
let (kind, rest) = line.split_once(' ').unwrap();
let (series_labels, series_points) = rest.split_once(" ;").unwrap();
let entry = (labels(series_labels), points(series_points));
let (floats, histogram_points) = points(series_points);
let entry = (labels(series_labels), floats);
match kind {
"series" => inputs.push(entry),
"series" => {
inputs.push(entry);
histograms.push(histogram_points);
}
"result" => {
outcome.as_mut().unwrap().insert(entry.0, entry.1);
}
@@ -217,6 +227,7 @@ pub(crate) fn golden_cases(text: &str) -> Vec<GoldenCase> {
spec: spec.split('\t').map(str::to_string).collect(),
grid: Arc::new(EvalGrid::try_new(grid).unwrap()),
inputs,
histograms,
outcome,
});
}
src/promql/src/histogram_eval.rs
+245
View File
@@ -0,0 +1,245 @@
// 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.
//! Histogram functions over evaluated series: `histogram_quantile` and `histogram_fraction` over
//! classic and native histograms (Prometheus `funcHistogramQuantile`, `funcHistogramFraction`
//! and `resetHistograms`), and `histogram_count`, `histogram_sum`, `histogram_avg`,
//! `histogram_stddev` and `histogram_stdvar` over native histograms.
mod classic;
mod native;
#[cfg(test)]
mod tests;
use std::collections::HashMap;
use std::sync::Arc;
use datafusion::error::{DataFusionError, Result};
use datatypes::series::MetricName;
use datatypes::series::histogram::HistogramView;
use datatypes::series::samples::{DenseSamplesRef, SampleRef, Series, SeriesSchema};
use datatypes::series::time::TimeAxis;
use crate::go_sort;
use crate::grid_series::{GridSeriesBuilder, LabelSet, label_set, merge_same_labelset};
const BUCKET_LABEL: &str = "le";
/// Prometheus `schema.IsMetadataLabel`: the labels removed with the metric name.
const METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"];
/// `histogram_quantile` or `histogram_fraction` with its constant arguments.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum HistogramFunction {
/// `histogram_quantile(q, v)`.
Quantile(f64),
/// `histogram_fraction(lower, upper, v)`.
Fraction { lower: f64, upper: f64 },
}
/// Evaluates `function` over the classic and native histograms in `series`.
///
/// Float series with the same labels other than `le` form a classic histogram; a float series
/// whose `le` does not parse is ignored. Native histogram samples are evaluated one by one. At a
/// point where a native histogram and classic buckets with the same labels (`le` aside) are both
/// present, neither is evaluated. Each field is evaluated on its own. Classic results drop
/// `le`; all results drop `__name__`, `__type__` and `__unit__`. Series that then share a label
/// set are merged as [`merge_same_labelset`] does.
pub fn evaluate_histograms(
function: HistogramFunction,
schema: &Arc<SeriesSchema>,
series: &[Series],
) -> Result<Vec<Series>> {
let mut groups: HashMap<LabelSet, Group<'_>> = HashMap::new();
for series in series {
let range = grid_range(series)?;
if let Some(upper_bound) = series
.metric
.get(BUCKET_LABEL)
.and_then(classic::parse_bound)
{
let mut metric = series.metric.clone();
metric.remove(BUCKET_LABEL);
groups
.entry(label_set(&metric))
.or_insert_with(|| Group::new(metric))
.classic
.push((upper_bound, series, range));
}
if has_histograms(series) {
let metric = series.metric.clone();
groups
.entry(label_set(&metric))
.or_insert_with(|| Group::new(metric))
.native = Some((series, range));
}
}
let mut output = Vec::with_capacity(groups.len());
let mut builder = GridSeriesBuilder::try_new(schema.clone())?;
let mut classic_buckets = Vec::new();
let mut native_buckets = Vec::new();
let mut values = vec![None; schema.fields.len()];
for mut group in groups.into_values() {
let ranges = group
.classic
.iter()
.map(|member| member.2)
.chain(group.native.map(|native| native.1));
let start = ranges.clone().map(|range| range.0).min().unwrap_or(0);
let end = ranges.map(|range| range.1).max().unwrap_or(0);
for point in start..end {
for (field, value) in values.iter_mut().enumerate() {
classic_buckets.clear();
for &(upper_bound, series, range) in &group.classic {
if let Some(SampleRef::Float(count)) = sample(series, range, field, point) {
classic_buckets.push(classic::Bucket { upper_bound, count });
}
}
// Go sorts the buckets present at each point by `<`: +0 and -0 tie and NaN
// ties with everything, so which of them coalescing keeps and where a NaN
// bound lands follow its sort algorithm.
go_sort::sort_by(&mut classic_buckets, |a, b| a.upper_bound < b.upper_bound);
let native = group.native.and_then(|(series, range)| {
match sample(series, range, field, point) {
Some(SampleRef::Histogram(h)) => Some(h),
_ => None,
}
});
// A native histogram and classic buckets of one label set conflict.
*value = match (native, classic_buckets.is_empty()) {
(Some(_), false) | (None, true) => None,
(Some(h), true) => Some(match function {
HistogramFunction::Quantile(q) => {
native::quantile(q, &h, &mut native_buckets)?
}
HistogramFunction::Fraction { lower, upper } => {
native::fraction(lower, upper, &h, &mut native_buckets)?
}
}),
(None, false) => Some(match function {
HistogramFunction::Quantile(q) => {
classic::bucket_quantile(q, &mut classic_buckets)
}
HistogramFunction::Fraction { lower, upper } => {
classic::bucket_fraction(lower, upper, &mut classic_buckets)
}
}),
};
}
builder.push(point, &values)?;
}
group.metric.remove_all(&METADATA_LABELS);
output.extend(builder.finish(group.metric)?);
}
merge_same_labelset(schema, output)
}
/// The series of one label set (`le` aside): classic buckets, and the series of exactly that
/// label set when it holds native histograms.
struct Group<'a> {
metric: MetricName,
classic: Vec<(f64, &'a Series, (usize, usize))>,
native: Option<(&'a Series, (usize, usize))>,
}
impl Group<'_> {
fn new(metric: MetricName) -> Self {
Self {
metric,
classic: Vec::new(),
native: None,
}
}
}
/// The absolute grid range `[start, end)` of `series`.
fn grid_range(series: &Series) -> Result<(usize, usize)> {
let TimeAxis::Grid(range) = &series.time else {
return Err(DataFusionError::Execution(
"histogram functions require evaluated series".into(),
));
};
Ok((range.start(), range.start() + range.len()))
}
fn sample(
series: &Series,
(start, end): (usize, usize),
field: usize,
point: usize,
) -> Option<SampleRef<'_>> {
(start..end)
.contains(&point)
.then(|| series.fields.as_slice()[field].sample(point - start))
}
fn has_histograms(series: &Series) -> bool {
series.fields.iter().any(|samples| {
samples
.segments()
.iter()
.any(|segment| !matches!(segment.dense(), Some(DenseSamplesRef::Float(_))))
})
}
/// A function of one native histogram.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NativeHistogramFunction {
Count,
Sum,
Avg,
StdDev,
StdVar,
}
/// Applies `function` to the native histogram samples of `series`; float samples are dropped and
/// so is `__name__`. The result is `None` when the series has no histogram sample.
///
/// Series that share a label set once their names are dropped must then go through
/// [`merge_same_labelset`].
pub fn map_native_histogram(
function: NativeHistogramFunction,
schema: &Arc<SeriesSchema>,
series: &Series,
) -> Result<Option<Series>> {
let range = grid_range(series)?;
let mut builder = GridSeriesBuilder::try_new(schema.clone())?;
let mut buckets = Vec::new();
let mut values = vec![None; schema.fields.len()];
for point in range.0..range.1 {
for (field, value) in values.iter_mut().enumerate() {
*value = match sample(series, range, field, point) {
Some(SampleRef::Histogram(h)) => Some(function.apply(&h, &mut buckets)?),
_ => None,
};
}
builder.push(point, &values)?;
}
let mut metric = series.metric.clone();
// Prometheus `simpleHistogramFunc` drops only `__name__`, unlike `histogram_quantile`.
metric.remove("__name__");
builder.finish(metric)
}
impl NativeHistogramFunction {
fn apply(self, h: &HistogramView<'_>, buckets: &mut Vec<native::NativeBucket>) -> Result<f64> {
Ok(match self {
Self::Count => h.count,
Self::Sum => h.sum,
Self::Avg => h.sum / h.count,
Self::StdDev => native::variance(h, buckets)?.sqrt(),
Self::StdVar => native::variance(h, buckets)?,
})
}
}
src/promql/src/histogram_eval/classic.rs
+235
View File
@@ -0,0 +1,235 @@
// 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.
//! Prometheus `BucketQuantile` and `BucketFraction` over the buckets of a classic histogram at
//! one point.
/// Relative difference under which neighbouring bucket counts are taken as equal.
const SMALL_DELTA_TOLERANCE: f64 = 1e-12;
/// Go's `strconv.ParseFloat`, which rejects finite literals out of the f64 range and a signed
/// NaN, and accepts `_` between digits. Hexadecimal literals are not supported.
pub(super) fn parse_bound(text: &str) -> Option<f64> {
let unsigned = text.trim_start_matches(['+', '-']);
if unsigned.len() != text.len() && unsigned.eq_ignore_ascii_case("nan") {
return None;
}
let value: f64 = if text.contains('_') {
if !underscores_ok(text) {
return None;
}
text.replace('_', "").parse().ok()?
} else {
text.parse().ok()?
};
if value.is_infinite() {
let unsigned = text.trim_start_matches(['+', '-']);
if !unsigned.eq_ignore_ascii_case("inf") && !unsigned.eq_ignore_ascii_case("infinity") {
return None;
}
}
Some(value)
}
/// Go's `strconv.underscoreOK` for a decimal literal: every `_` sits between two digits.
fn underscores_ok(text: &str) -> bool {
let digits = text.strip_prefix(['+', '-']).unwrap_or(text).as_bytes();
digits.iter().enumerate().all(|(i, &byte)| {
byte != b'_'
|| (i > 0
&& digits[i - 1].is_ascii_digit()
&& digits.get(i + 1).is_some_and(u8::is_ascii_digit))
})
}
#[derive(Debug, Clone, Copy)]
pub(super) struct Bucket {
pub(super) upper_bound: f64,
pub(super) count: f64,
}
/// Prometheus `BucketQuantile` over buckets sorted by upper bound.
pub(super) fn bucket_quantile(q: f64, buckets: &mut Vec<Bucket>) -> f64 {
if q.is_nan() {
return f64::NAN;
}
if q < 0.0 {
return f64::NEG_INFINITY;
}
if q > 1.0 {
return f64::INFINITY;
}
if buckets.last().unwrap().upper_bound != f64::INFINITY {
return f64::NAN;
}
coalesce(buckets);
ensure_monotonic(buckets);
if buckets.len() < 2 {
return f64::NAN;
}
let observations = buckets.last().unwrap().count;
if observations == 0.0 {
return f64::NAN;
}
let mut rank = q * observations;
let b = go_search(buckets.len() - 1, |i| buckets[i].count >= rank);
if b == buckets.len() - 1 {
return buckets[buckets.len() - 2].upper_bound;
}
if b == 0 && buckets[0].upper_bound <= 0.0 {
return buckets[0].upper_bound;
}
let mut bucket_start = 0.0;
let bucket_end = buckets[b].upper_bound;
let mut count = buckets[b].count;
if b > 0 {
bucket_start = buckets[b - 1].upper_bound;
count -= buckets[b - 1].count;
rank -= buckets[b - 1].count;
}
bucket_start + (bucket_end - bucket_start) * (rank / count)
}
/// Prometheus `BucketFraction` over buckets sorted by upper bound.
pub(super) fn bucket_fraction(lower: f64, upper: f64, buckets: &mut Vec<Bucket>) -> f64 {
if buckets.last().unwrap().upper_bound != f64::INFINITY {
return f64::NAN;
}
coalesce(buckets);
let count = buckets.last().unwrap().count;
if count == 0.0 || lower.is_nan() || upper.is_nan() {
return f64::NAN;
}
if lower >= upper {
return 0.0;
}
let (mut rank, mut lower_rank, mut upper_rank) = (0.0, 0.0, 0.0);
let (mut lower_set, mut upper_set) = (false, false);
let mut lower_bound = if buckets[0].upper_bound <= 0.0 {
f64::NEG_INFINITY
} else {
0.0
};
for (i, bucket) in buckets.iter().enumerate() {
if i > 0 {
lower_bound = buckets[i - 1].upper_bound;
}
let upper_bound = bucket.upper_bound;
// An infinite-width bucket contributes nothing to the interpolation.
let interpolate = |value: f64| {
if lower_bound == f64::NEG_INFINITY {
bucket.count
} else {
rank + (bucket.count - rank) * (value - lower_bound) / (upper_bound - lower_bound)
}
};
if !lower_set && lower_bound >= lower {
lower_rank = rank;
lower_set = true;
}
if !upper_set && lower_bound >= upper {
upper_rank = rank;
upper_set = true;
}
if lower_set && upper_set {
break;
}
if !lower_set && lower_bound < lower && upper_bound > lower {
lower_rank = interpolate(lower);
lower_set = true;
}
if !upper_set && lower_bound < upper && upper_bound > upper {
upper_rank = interpolate(upper);
upper_set = true;
}
if lower_set && upper_set {
break;
}
rank = bucket.count;
}
if !lower_set || lower_rank > count {
lower_rank = count;
}
if !upper_set || upper_rank > count {
upper_rank = count;
}
(upper_rank - lower_rank) / count
}
/// Merges buckets with the same upper bound.
fn coalesce(buckets: &mut Vec<Bucket>) {
let mut last = buckets[0];
let mut i = 0;
for j in 1..buckets.len() {
let bucket = buckets[j];
if bucket.upper_bound == last.upper_bound {
last.count += bucket.count;
} else {
buckets[i] = last;
last = bucket;
i += 1;
}
}
buckets[i] = last;
buckets.truncate(i + 1);
}
/// Prometheus `ensureMonotonicAndIgnoreSmallDeltas`: counts that differ from the previous one by
/// a negligible relative amount, or are lower, are raised to it.
fn ensure_monotonic(buckets: &mut [Bucket]) {
let mut prev = buckets[0].count;
for bucket in &mut buckets[1..] {
let curr = bucket.count;
if curr == prev {
continue;
}
if almost_equal(prev, curr, SMALL_DELTA_TOLERANCE) || curr < prev {
bucket.count = prev;
continue;
}
prev = curr;
}
}
/// Prometheus `almost.Equal`.
fn almost_equal(a: f64, b: f64, epsilon: f64) -> bool {
if a.is_nan() && b.is_nan() {
return true;
}
if a == b {
return true;
}
let abs_sum = a.abs() + b.abs();
let diff = (a - b).abs();
if a == 0.0 || b == 0.0 || abs_sum < f64::MIN_POSITIVE {
return diff < epsilon * f64::MIN_POSITIVE;
}
diff / abs_sum.min(f64::MAX) < epsilon
}
/// Go's `sort.Search`: the smallest index in `[0, n)` where `f` holds, or `n`, found by the same
/// bisection, so a predicate that is not monotonic (NaN counts) picks the same index.
fn go_search(n: usize, f: impl Fn(usize) -> bool) -> usize {
let (mut i, mut j) = (0, n);
while i < j {
let h = (i + j) / 2;
if !f(h) {
i = h + 1;
} else {
j = h;
}
}
i
}
src/promql/src/histogram_eval/classic_cases.txt
+3863
View File
File diff suppressed because it is too large Load Diff
src/promql/src/histogram_eval/native.rs
+378
View File
@@ -0,0 +1,378 @@
// 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.
//! Prometheus `HistogramQuantile`, `HistogramFraction` and the `histogramVariance` estimate over
//! one native histogram, with the bucket iteration of `model/histogram`.
use datafusion::error::{DataFusionError, Result};
use datatypes::series::histogram::{BucketSideView, HistogramView};
const CUSTOM_BUCKETS_SCHEMA: i32 = -53;
#[derive(Debug, Clone, Copy)]
pub(super) struct NativeBucket {
lower: f64,
upper: f64,
count: f64,
}
/// The buckets of `h` from the lowest to the highest, or the reverse: negative buckets, the zero
/// bucket when it has observations, then positive buckets (Go `AllBucketIterator` and
/// `AllReverseBucketIterator`). Buckets overlapping the zero bucket are clipped to it.
pub(super) fn all_buckets(
h: &HistogramView<'_>,
reverse: bool,
out: &mut Vec<NativeBucket>,
) -> Result<()> {
out.clear();
let (low, high) = if reverse {
(h.positive, h.negative)
} else {
(h.negative, h.positive)
};
// The first side runs from its highest index down, the second from its lowest index up.
side(h, low, reverse, out)?;
out.reverse();
if h.zero_count > 0.0 {
if h.schema == CUSTOM_BUCKETS_SCHEMA {
return Err(DataFusionError::Execution(
"histograms with custom buckets have no zero bucket".into(),
));
}
out.push(NativeBucket {
lower: -h.zero_threshold,
upper: h.zero_threshold,
count: h.zero_count,
});
}
side(h, high, !reverse, out)?;
for bucket in out.iter_mut() {
if bucket.upper < 0.0 && bucket.upper > -h.zero_threshold {
bucket.upper = -h.zero_threshold;
} else if bucket.lower > 0.0 && bucket.lower < h.zero_threshold {
bucket.lower = h.zero_threshold;
}
}
Ok(())
}
/// Appends the buckets of one side by increasing index.
fn side(
h: &HistogramView<'_>,
side: BucketSideView<'_>,
positive: bool,
out: &mut Vec<NativeBucket>,
) -> Result<()> {
if side.span_offsets.len() != side.span_lengths.len() {
return Err(invalid_layout());
}
let mut counts = side.buckets.iter();
let mut index = 0i64;
for (&offset, &length) in side.span_offsets.iter().zip(side.span_lengths) {
if length < 0 {
return Err(invalid_layout());
}
index += i64::from(offset);
for _ in 0..length {
let count = *counts.next().ok_or_else(invalid_layout)?;
let idx = i32::try_from(index).map_err(|_| invalid_layout())?;
out.push(bucket(h, idx, count, positive)?);
index += 1;
}
}
if counts.next().is_some() {
return Err(invalid_layout());
}
Ok(())
}
fn bucket(h: &HistogramView<'_>, idx: i32, count: f64, positive: bool) -> Result<NativeBucket> {
let bound = |idx: i32| bound(idx, h.schema, h.custom_values);
Ok(if positive {
NativeBucket {
lower: bound(idx - 1)?,
upper: bound(idx)?,
count,
}
} else {
NativeBucket {
lower: -bound(idx)?,
upper: -bound(idx - 1)?,
count,
}
})
}
/// Go `getBound`.
fn bound(idx: i32, schema: i32, custom_values: &[f64]) -> Result<f64> {
if schema == CUSTOM_BUCKETS_SCHEMA {
let len = custom_values.len() as i64;
return match i64::from(idx) {
-1 => Ok(f64::NEG_INFINITY),
i if i == len => Ok(f64::INFINITY),
i if (0..len).contains(&i) => Ok(custom_values[i as usize]),
_ => Err(invalid_layout()),
};
}
if !(-4..=8).contains(&schema) {
return Err(invalid_layout());
}
// The upper bound of bucket `idx` is 2^(idx / 2^schema).
let exponent = f64::from(idx) * 2f64.powi(-schema);
if exponent == 1024.0 {
// The last bucket of finite values; its computed bound would overflow to +Inf, which
// Go reserves for the bucket of infinite observations.
return Ok(f64::MAX);
}
Ok(exponent.exp2())
}
fn invalid_layout() -> DataFusionError {
DataFusionError::Execution("native histogram has an invalid bucket layout".into())
}
/// Prometheus `HistogramQuantile`.
pub(super) fn quantile(
q: f64,
h: &HistogramView<'_>,
buckets: &mut Vec<NativeBucket>,
) -> Result<f64> {
if q < 0.0 {
return Ok(f64::NEG_INFINITY);
}
if q > 1.0 {
return Ok(f64::INFINITY);
}
if h.count == 0.0 || q.is_nan() {
return Ok(f64::NAN);
}
let custom = h.schema == CUSTOM_BUCKETS_SCHEMA;
let forward = h.sum.is_nan() || q < 0.5;
all_buckets(h, !forward, buckets)?;
let mut rank = if forward {
q * h.count
} else {
(1.0 - q) * h.count
};
// Go keeps the last bucket visited, empty or not, and a zero bucket when there is none.
let mut bucket = NativeBucket {
lower: 0.0,
upper: 0.0,
count: 0.0,
};
let mut count = 0.0;
let mut visited = 0;
for &next in buckets.iter() {
visited += 1;
bucket = next;
if bucket.count == 0.0 {
continue;
}
count += bucket.count;
if count >= rank {
break;
}
}
if !custom && bucket.lower < 0.0 && bucket.upper > 0.0 {
if h.negative.buckets.is_empty() && !h.positive.buckets.is_empty() {
bucket.lower = 0.0;
} else if h.positive.buckets.is_empty() && !h.negative.buckets.is_empty() {
bucket.upper = 0.0;
}
} else if custom {
if bucket.lower == f64::NEG_INFINITY {
if bucket.upper <= 0.0 {
return Ok(bucket.upper);
}
bucket.lower = 0.0;
} else if bucket.upper == f64::INFINITY {
return Ok(bucket.lower);
}
}
if count > h.count {
count = h.count;
}
if count < rank {
return Ok(if h.sum.is_nan() {
f64::NAN
} else {
bucket.upper
});
}
if forward {
rank -= count - bucket.count;
} else {
rank = count - rank;
}
if h.sum.is_nan() {
// Prometheus walks the remaining buckets to annotate NaN observations and leaves the last
// one in `bucket`, which the interpolation below then uses.
for &next in &buckets[visited..] {
bucket = next;
}
}
let fraction = rank / bucket.count;
if custom || (bucket.lower <= 0.0 && bucket.upper >= 0.0) {
return Ok(bucket.lower + (bucket.upper - bucket.lower) * fraction);
}
let log_lower = bucket.lower.abs().log2();
let log_upper = bucket.upper.abs().log2();
if bucket.lower > 0.0 {
return Ok((log_lower + (log_upper - log_lower) * fraction).exp2());
}
Ok(-(log_upper + (log_lower - log_upper) * (1.0 - fraction)).exp2())
}
/// Prometheus `HistogramFraction`.
pub(super) fn fraction(
lower: f64,
upper: f64,
h: &HistogramView<'_>,
buckets: &mut Vec<NativeBucket>,
) -> Result<f64> {
if h.count == 0.0 || lower.is_nan() || upper.is_nan() {
return Ok(f64::NAN);
}
if lower >= upper {
return Ok(0.0);
}
let custom = h.schema == CUSTOM_BUCKETS_SCHEMA;
all_buckets(h, false, buckets)?;
let (mut count, mut rank, mut lower_rank, mut upper_rank) = (0.0, 0.0, 0.0, 0.0);
let (mut lower_set, mut upper_set) = (false, false);
let mut visited = 0;
for b in buckets.iter() {
visited += 1;
let mut b = *b;
count += b.count;
let mut zero_bucket = false;
if b.lower <= 0.0 && b.upper >= 0.0 {
zero_bucket = true;
if h.negative.buckets.is_empty() && !h.positive.buckets.is_empty() {
b.lower = 0.0;
} else if h.positive.buckets.is_empty() && !h.negative.buckets.is_empty() {
b.upper = 0.0;
}
}
let interpolate = |v: f64| {
if custom || zero_bucket {
if b.lower == f64::NEG_INFINITY {
b.count
} else {
rank + b.count * ((v - b.lower) / (b.upper - b.lower))
}
} else {
rank + b.count * fraction_below_exponential(b, v)
}
};
if !lower_set && b.lower >= lower {
lower_rank = rank;
lower_set = true;
}
if !upper_set && b.lower >= upper {
upper_rank = rank;
upper_set = true;
}
if lower_set && upper_set {
break;
}
if !lower_set && b.lower < lower && b.upper > lower {
lower_rank = interpolate(lower);
lower_set = true;
}
if !upper_set && b.lower < upper && b.upper > upper {
upper_rank = interpolate(upper);
upper_set = true;
}
if lower_set && upper_set {
break;
}
rank += b.count;
}
if h.sum.is_nan() {
for b in &buckets[visited..] {
count += b.count;
}
} else {
count = h.count;
}
if !lower_set || lower_rank > count {
lower_rank = count;
}
if !upper_set || upper_rank > count {
upper_rank = count;
}
Ok((upper_rank - lower_rank) / h.count)
}
/// Go `Bucket.FractionBelow` with exponential interpolation.
fn fraction_below_exponential(b: NativeBucket, v: f64) -> f64 {
let log_lower = b.lower.abs().log2();
let log_upper = b.upper.abs().log2();
let log_v = v.abs().log2();
if v > 0.0 {
(log_v - log_lower) / (log_upper - log_lower)
} else {
1.0 - ((log_v - log_upper) / (log_lower - log_upper))
}
}
/// Prometheus `histogramVariance`: bucket midpoints weighted by count around the mean.
pub(super) fn variance(h: &HistogramView<'_>, buckets: &mut Vec<NativeBucket>) -> Result<f64> {
let mean = h.sum / h.count;
let custom = h.schema == CUSTOM_BUCKETS_SCHEMA;
all_buckets(h, false, buckets)?;
let (mut variance, mut compensation) = (0.0, 0.0);
for bucket in buckets.iter().filter(|bucket| bucket.count != 0.0) {
let value = if custom {
(bucket.upper + bucket.lower) / 2.0
} else if bucket.lower <= 0.0 && bucket.upper >= 0.0 {
0.0
} else {
let value = (bucket.upper * bucket.lower).sqrt();
if bucket.upper < 0.0 { -value } else { value }
};
let delta = value - mean;
(variance, compensation) = kahan_inc(bucket.count * delta * delta, variance, compensation);
}
Ok((variance + compensation) / h.count)
}
/// Prometheus `kahansum.Inc`.
fn kahan_inc(inc: f64, sum: f64, c: f64) -> (f64, f64) {
let t = sum + inc;
if t.is_infinite() {
return (t, 0.0);
}
let c = if sum.abs() >= inc.abs() {
c + ((sum - t) + inc)
} else {
c + ((inc - t) + sum)
};
(t, c)
}
#[cfg(test)]
mod tests {
/// The last bucket of finite values ends at `f64::MAX`; only the next one, which counts
/// infinite observations, ends at +Inf.
#[test]
fn last_finite_bucket_ends_at_max() {
for schema in [-4, 0, 3, 8] {
let last = 1024i32 << schema.max(0) >> (-schema).max(0);
assert_eq!(super::bound(last, schema, &[]).unwrap(), f64::MAX);
assert_eq!(super::bound(last + 1, schema, &[]).unwrap(), f64::INFINITY);
}
}
}
src/promql/src/histogram_eval/native_cases.txt
+2964
View File
File diff suppressed because it is too large Load Diff
src/promql/src/histogram_eval/tests.rs
+467
View File
@@ -0,0 +1,467 @@
// 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 datafusion::error::DataFusionError;
use datatypes::series::histogram::{
BucketSideView, CounterResetHint, HistogramBatchBuilder, HistogramView,
};
use datatypes::series::samples::{
FieldSpec, HistogramSamples, Samples, Series, SeriesFields, SeriesSchema,
};
use datatypes::series::time::{GridRange, TimeAxis};
use super::{
HistogramFunction, NativeHistogramFunction, evaluate_histograms, map_native_histogram,
};
use crate::grid_series::merge_same_labelset;
use crate::grid_series::reference::{
GoldenCase, Labels, Points, canonical_bits, golden_cases, labels_of, metric_of, points_of,
};
fn schema(case: &GoldenCase) -> Arc<SeriesSchema> {
Arc::new(SeriesSchema {
fields: vec![FieldSpec {
name: Arc::from("value"),
}]
.into(),
eval_grid: Some(case.grid.clone()),
})
}
fn bits_argument(case: &GoldenCase, index: usize) -> f64 {
f64::from_bits(u64::from_str_radix(&case.spec[index], 16).unwrap())
}
/// A native histogram written by the generator as `/`-separated fields.
struct OwnedHistogram {
schema: i32,
zero_threshold: f64,
zero_count: f64,
count: f64,
sum: f64,
reset_hint: i32,
positive: (Vec<i32>, Vec<i32>, Vec<f64>),
negative: (Vec<i32>, Vec<i32>, Vec<f64>),
custom_values: Vec<f64>,
}
impl OwnedHistogram {
fn parse(text: &str) -> Self {
let fields: Vec<&str> = text.split('/').collect();
let float = |text: &str| f64::from_bits(u64::from_str_radix(text, 16).unwrap());
let floats = |text: &str| -> Vec<f64> {
text.split(',')
.filter(|v| !v.is_empty())
.map(float)
.collect()
};
let spans = |text: &str| -> (Vec<i32>, Vec<i32>) {
text.split(',')
.filter(|v| !v.is_empty())
.map(|span| {
let (offset, length) = span.split_once('_').unwrap();
(
offset.parse::<i32>().unwrap(),
length.parse::<i32>().unwrap(),
)
})
.unzip()
};
let (positive_offsets, positive_lengths) = spans(fields[6]);
let (negative_offsets, negative_lengths) = spans(fields[8]);
Self {
schema: fields[0].parse().unwrap(),
zero_threshold: float(fields[1]),
zero_count: float(fields[2]),
count: float(fields[3]),
sum: float(fields[4]),
reset_hint: fields[5].parse().unwrap(),
positive: (positive_offsets, positive_lengths, floats(fields[7])),
negative: (negative_offsets, negative_lengths, floats(fields[9])),
custom_values: floats(fields[10]),
}
}
fn view(&self) -> HistogramView<'_> {
fn side(
(offsets, lengths, buckets): &(Vec<i32>, Vec<i32>, Vec<f64>),
) -> BucketSideView<'_> {
BucketSideView {
span_offsets: offsets,
span_lengths: lengths,
buckets,
}
}
HistogramView {
schema: self.schema,
zero_threshold: self.zero_threshold,
sum: self.sum,
count: self.count,
zero_count: self.zero_count,
reset_hint: CounterResetHint::from(self.reset_hint),
start_timestamp: None,
custom_values: &self.custom_values,
positive: side(&self.positive),
negative: side(&self.negative),
}
}
}
/// A single-field series holding float and native histogram points.
fn mixed_series(
schema: &Arc<SeriesSchema>,
labels: &Labels,
floats: &Points,
histograms: &BTreeMap<usize, String>,
) -> Option<Series> {
let points = floats.keys().chain(histograms.keys());
let start = *points.clone().min()?;
let end = *points.max()? + 1;
let float_values: Float64Array = (start..end)
.map(|point| floats.get(&point).map(|bits| f64::from_bits(*bits)))
.collect();
let samples = if histograms.is_empty() {
Samples::from_float(float_values)
} else {
let mut builder = HistogramBatchBuilder::new();
let indices: Vec<_> = (start..end)
.map(|point| {
let text = histograms.get(&point)?;
Some(builder.push(OwnedHistogram::parse(text).view()).unwrap())
})
.collect();
let histograms =
HistogramSamples::try_new(UInt32Array::from(indices), builder.finish().unwrap())
.unwrap();
Samples::try_from_mixed(float_values, histograms).unwrap()
};
let range = GridRange::try_new(schema.eval_grid.clone().unwrap(), start, end - start).unwrap();
Some(
Series::try_new(
schema,
metric_of(labels),
TimeAxis::Grid(range),
SeriesFields::one(samples),
)
.unwrap(),
)
}
/// Compares a kernel outcome with a fixture outcome; returns whether it was an error.
fn assert_outcome(
case: &GoldenCase,
actual: Result<Vec<Series>, DataFusionError>,
expected: &Result<BTreeMap<Labels, Points>, String>,
) -> bool {
match (actual, expected) {
(Ok(actual), Ok(expected)) => {
let actual: BTreeMap<_, _> = actual
.iter()
.map(|series| (labels_of(&series.metric), points_of(series)))
.collect();
let expected: BTreeMap<_, _> = expected
.iter()
.map(|(labels, points)| {
let points = points
.iter()
.map(|(&index, &bits)| (index, canonical_bits(f64::from_bits(bits))))
.collect::<Points>();
(labels.clone(), points)
})
.collect();
let keys = |series: &BTreeMap<Labels, Points>| {
series
.iter()
.map(|(labels, points)| (labels.clone(), points.keys().copied().collect()))
.collect::<Vec<(Labels, Vec<usize>)>>()
};
assert_eq!(keys(&actual), keys(&expected), "case {:?}", case.spec);
// promql-compat's tolerance for ill-conditioned evaluation covers the variance.
let epsilon = if case.spec[0].contains("stddev") || case.spec[0].contains("stdvar") {
1e-6
} else {
1e-10
};
for (labels, points) in &actual {
for (point, &bits) in points {
let got = f64::from_bits(bits);
let want = f64::from_bits(expected[labels][point]);
assert!(
close(got, want, epsilon),
"case {:?}: {labels:?} at {point}: {got} vs {want}",
case.spec
);
}
}
false
}
(Err(actual), Err(expected)) => {
assert!(actual.to_string().contains(expected), "{actual}");
true
}
(actual, expected) => panic!("case {:?}: {actual:?} is not {expected:?}", case.spec),
}
}
#[test]
fn classic_histograms_match_prometheus() {
let cases = golden_cases(include_str!("classic_cases.txt"));
assert_eq!(cases.len(), 300);
let mut failed = 0;
for case in &cases {
let schema = schema(case);
// The fixture lists buckets by bound; the kernel must not rely on input order.
let input: Vec<_> = case
.inputs
.iter()
.rev()
.filter_map(|(labels, points)| mixed_series(&schema, labels, points, &BTreeMap::new()))
.collect();
let function = match case.spec[0].as_str() {
"quantile" => HistogramFunction::Quantile(bits_argument(case, 1)),
"fraction" => HistogramFunction::Fraction {
lower: bits_argument(case, 1),
upper: bits_argument(case, 2),
},
other => panic!("unknown case {other}"),
};
let actual = evaluate_histograms(function, &schema, &input);
failed += usize::from(assert_outcome(case, actual, &case.outcome));
}
assert!(failed > 20, "{failed} failed");
}
#[test]
fn native_histograms_match_prometheus() {
let cases = golden_cases(include_str!("native_cases.txt"));
assert_eq!(cases.len(), 420);
for case in &cases {
let schema = schema(case);
let input: Vec<_> = case
.inputs
.iter()
.zip(&case.histograms)
.filter_map(|((labels, floats), histograms)| {
mixed_series(&schema, labels, floats, histograms)
})
.collect();
let map = |function| -> Result<Vec<Series>, DataFusionError> {
let mapped = input
.iter()
.filter_map(|series| map_native_histogram(function, &schema, series).transpose())
.collect::<Result<Vec<_>, _>>()?;
merge_same_labelset(&schema, mapped)
};
let actual = match case.spec[0].as_str() {
"histogram_count" => map(NativeHistogramFunction::Count),
"histogram_sum" => map(NativeHistogramFunction::Sum),
"histogram_avg" => map(NativeHistogramFunction::Avg),
"histogram_stddev" => map(NativeHistogramFunction::StdDev),
"histogram_stdvar" => map(NativeHistogramFunction::StdVar),
"histogram_quantile" => evaluate_histograms(
HistogramFunction::Quantile(bits_argument(case, 1)),
&schema,
&input,
),
"histogram_fraction" => evaluate_histograms(
HistogramFunction::Fraction {
lower: bits_argument(case, 1),
upper: bits_argument(case, 2),
},
&schema,
&input,
),
other => panic!("unknown case {other}"),
};
assert_outcome(case, actual, &case.outcome);
}
}
/// `histogram_quantile` drops all metadata labels; the single-histogram functions drop only
/// `__name__`, as Prometheus does.
#[test]
fn histogram_functions_drop_metadata_labels() {
let grid = Arc::new(
datatypes::series::time::EvalGrid::try_new(
datatypes::series::time::GridSpec::try_new(0, 1_000, 1).unwrap(),
)
.unwrap(),
);
let schema = Arc::new(SeriesSchema {
fields: vec![FieldSpec {
name: Arc::from("value"),
}]
.into(),
eval_grid: Some(grid),
});
let labels = |extra: &[(&str, &str)]| -> Labels {
[
("__name__", "m"),
("__type__", "histogram"),
("__unit__", "seconds"),
("job", "x"),
]
.iter()
.chain(extra)
.map(|(name, value)| (name.to_string(), value.to_string()))
.collect()
};
let one = BTreeMap::from([(0, 1f64.to_bits())]);
let buckets = vec![
mixed_series(&schema, &labels(&[("le", "1")]), &one, &BTreeMap::new()).unwrap(),
mixed_series(&schema, &labels(&[("le", "+Inf")]), &one, &BTreeMap::new()).unwrap(),
];
let quantile =
evaluate_histograms(HistogramFunction::Quantile(0.5), &schema, &buckets).unwrap();
assert_eq!(
labels_of(&quantile[0].metric),
Labels::from([("job".to_string(), "x".to_string())])
);
let native = mixed_series(
&schema,
&labels(&[]),
&BTreeMap::new(),
&BTreeMap::from([(
0,
"0/0/0/3ff0000000000000/3ff0000000000000/0/0_1/3ff0000000000000///".to_string(),
)]),
)
.unwrap();
let count = map_native_histogram(NativeHistogramFunction::Count, &schema, &native)
.unwrap()
.unwrap();
let mut expected = labels(&[]);
expected.remove("__name__");
assert_eq!(labels_of(&count.metric), expected);
}
/// Prometheus `almost.Equal`, treating results below 1e-10 as equal and any NaN as equal to NaN,
/// as promql-compat compares engines.
fn close(a: f64, b: f64, epsilon: f64) -> bool {
if a.is_nan() || b.is_nan() {
return a.is_nan() && b.is_nan();
}
if a == b {
return true;
}
let (diff, sum) = ((a - b).abs(), a.abs() + b.abs());
let equal = if a == 0.0 || b == 0.0 || sum < f64::MIN_POSITIVE {
diff < epsilon * f64::MIN_POSITIVE
} else {
diff / sum.min(f64::MAX) < epsilon
};
equal || (a.abs() < 1e-10 && b.abs() < 1e-10)
}
/// Bounds +0 and -0 compare equal and keep input order, so coalescing keeps the first one:
/// `1 / histogram_quantile(0, sort_desc(h))` over `le` "+Inf", "0", "-0" is +Inf.
#[test]
fn classic_bounds_keep_the_first_signed_zero() {
let grid = Arc::new(
datatypes::series::time::EvalGrid::try_new(
datatypes::series::time::GridSpec::try_new(0, 1_000, 1).unwrap(),
)
.unwrap(),
);
let schema = Arc::new(SeriesSchema {
fields: vec![FieldSpec {
name: Arc::from("value"),
}]
.into(),
eval_grid: Some(grid),
});
let buckets: Vec<_> = [("+Inf", 3.0), ("0", 2.0), ("-0", 1.0)]
.iter()
.map(|(le, count)| {
let labels = Labels::from([
("__name__".to_string(), "h".to_string()),
("le".to_string(), le.to_string()),
]);
let points = BTreeMap::from([(0, f64::to_bits(*count))]);
mixed_series(&schema, &labels, &points, &BTreeMap::new()).unwrap()
})
.collect();
let result = evaluate_histograms(HistogramFunction::Quantile(0.0), &schema, &buckets).unwrap();
let points = points_of(&result[0]);
assert_eq!(1.0 / f64::from_bits(points[&0]), f64::INFINITY);
}
/// Classic histograms over `le` values in input order, at one point.
fn classic_result(function: HistogramFunction, buckets: &[(&str, f64)]) -> f64 {
let grid = Arc::new(
datatypes::series::time::EvalGrid::try_new(
datatypes::series::time::GridSpec::try_new(0, 1_000, 1).unwrap(),
)
.unwrap(),
);
let schema = Arc::new(SeriesSchema {
fields: vec![FieldSpec {
name: Arc::from("value"),
}]
.into(),
eval_grid: Some(grid),
});
let series: Vec<_> = buckets
.iter()
.map(|(le, count)| {
let labels = Labels::from([
("__name__".to_string(), "h".to_string()),
("le".to_string(), le.to_string()),
]);
let points = BTreeMap::from([(0, f64::to_bits(*count))]);
mixed_series(&schema, &labels, &points, &BTreeMap::new()).unwrap()
})
.collect();
let result = evaluate_histograms(function, &schema, &series).unwrap();
f64::from_bits(points_of(&result[0])[&0])
}
/// Go's insertion sort cannot move a bucket across a NaN bound, so a NaN bucket stays where it
/// is and `+Inf` can still end the list. Values are from Prometheus v3.15.0.
#[test]
fn nan_bounds_stay_in_place() {
let buckets = [("1", 10.0), ("NaN", 20.0), ("inf", 30.0)];
let quantile = classic_result(HistogramFunction::Quantile(0.1), &buckets);
assert!((quantile - 0.3).abs() < 1e-12, "{quantile}");
let fraction = classic_result(
HistogramFunction::Fraction {
lower: 0.0,
upper: 0.5,
},
&buckets,
);
assert!((fraction - 1.0 / 6.0).abs() < 1e-12, "{fraction}");
}
/// `le` parses like Go's `strconv.ParseFloat`: a signed NaN is not a bound, and `_` may separate
/// digits. Values are from Prometheus v3.15.0.
#[test]
fn bounds_parse_like_go() {
let signed_nan = [("1", 10.0), ("+nan", 20.0), ("+Inf", 30.0)];
for (q, expected) in [(0.1, 0.3), (0.5, 1.0)] {
let quantile = classic_result(HistogramFunction::Quantile(q), &signed_nan);
assert!((quantile - expected).abs() < 1e-12, "{q}: {quantile}");
}
let underscore = [("1", 10.0), ("1_000", 20.0), ("+Inf", 30.0)];
let quantile = classic_result(HistogramFunction::Quantile(0.5), &underscore);
assert!((quantile - 500.5).abs() < 1e-9, "{quantile}");
for rejected in ["-nan", "_1", "1__0", "1_", "1_.5", "1e_5"] {
assert_eq!(super::classic::parse_bound(rejected), None, "{rejected}");
}
}
src/promql/src/lib.rs
+1
View File
@@ -17,6 +17,7 @@ pub mod extension_plan;
pub mod functions;
mod go_sort;
pub mod grid_series;
pub mod histogram_eval;
pub mod map_label;
pub mod map_time;
mod metrics;