diff --git a/src/promql/src/histogram_eval.rs b/src/promql/src/histogram_eval.rs index d38b860c99d..7b8668b5163 100644 --- a/src/promql/src/histogram_eval.rs +++ b/src/promql/src/histogram_eval.rs @@ -97,7 +97,17 @@ pub fn evaluate_histograms( schema: &Arc, series: &[Series], ) -> Result> { - let mut groups: HashMap> = HashMap::new(); + // The groups in the order their first member arrived, so the result is in one order on every + // run, as `merge_same_labelset` keeps it. + let mut index: HashMap = HashMap::new(); + let mut groups: Vec> = Vec::new(); + let mut group_of = |metric: MetricName| { + *index.entry(label_set(&metric)).or_insert_with(|| { + groups.push(Group::new(metric)); + groups.len() - 1 + }) + }; + let mut members: Vec<(usize, Option, &Series)> = Vec::new(); for series in series { grid_axis(series)?; if let Some(upper_bound) = series @@ -107,26 +117,27 @@ pub fn evaluate_histograms( { 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)); + members.push((group_of(metric), Some(upper_bound), series)); } if has_histograms(series) { - let metric = series.metric.clone(); - let group = groups - .entry(label_set(&metric)) - .or_insert_with(|| Group::new(metric)); - // A label set carries at most one native histogram series: a second one would - // overwrite the first and drop it from the result. - debug_assert!( - group.native.is_none(), - "one label set carries at most one native histogram series; a second one for \ - the same label set means the input handed one label set over as several \ - series, which a gap in the axis is there to avoid" - ); - group.native = Some(series); + members.push((group_of(series.metric.clone()), None, series)); + } + } + for (slot, upper_bound, series) in members { + let group = &mut groups[slot]; + match upper_bound { + Some(upper_bound) => group.classic.push((upper_bound, series)), + None => { + // A label set carries at most one native histogram series: a second one would + // overwrite the first and drop it from the result. + debug_assert!( + group.native.is_none(), + "one label set carries at most one native histogram series; a second one for \ + the same label set means the input handed one label set over as several \ + series, which a gap in the axis is there to avoid" + ); + group.native = Some(series); + } } } @@ -135,7 +146,7 @@ pub fn evaluate_histograms( 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() { + for mut group in groups { // One builder serves every group, so each result series starts explicitly. builder.begin_series(); // The positions any member carries, merged once: the holes no member holds are never diff --git a/src/promql/src/native/display.rs b/src/promql/src/native/display.rs index 40768f8bed1..0e0349bb79d 100644 --- a/src/promql/src/native/display.rs +++ b/src/promql/src/native/display.rs @@ -25,6 +25,7 @@ use std::sync::Arc; use datatypes::series::time::EvalGrid; use datatypes::series::{MetricName, SeriesOrdering}; +use crate::histogram_eval::HistogramFunction; use crate::map::MathOp; use crate::native::metrics::{self, NodePath}; use crate::native::plan::{ @@ -120,6 +121,12 @@ fn describe(node: &NativeNode) -> String { } }, BoundVectorOp::Absent { labels } => format!("absent {{{}}}", label_pairs(labels)), + BoundVectorOp::Histogram(HistogramFunction::Quantile(quantile)) => { + format!("histogram_quantile {quantile}") + } + BoundVectorOp::Histogram(HistogramFunction::Fraction { lower, upper }) => { + format!("histogram_fraction {lower} {upper}") + } }, NativeExpr::Aggregate { op, .. } => format!( "{} {} [{}] {}", diff --git a/src/promql/src/native/execute.rs b/src/promql/src/native/execute.rs index 2a1b6b13a3b..600a9b61d08 100644 --- a/src/promql/src/native/execute.rs +++ b/src/promql/src/native/execute.rs @@ -104,6 +104,7 @@ use datatypes::series::{MetricName, SeriesOrdering}; use crate::grid_series::{ GridSample, GridSeriesBuilder, LabelSet, label_set, merge_same_labelset, owned, }; +use crate::histogram_eval::evaluate_histograms; use crate::instant_eval::select_instants; use crate::map::{ ArithmeticOp, Comparison, ComparisonOp, MathOp, ScalarOperand, drop_metric_name, @@ -1892,7 +1893,7 @@ fn vector_overflow(path: &NodePath) -> DataFusionError { /// Runs the kernel `op` binds over the whole input. fn vector_series( op: &BoundVectorOp, - schema: &SeriesSchema, + schema: &Arc, series: Vec, ) -> Result> { let grouping = |group: &Grouping| match group { @@ -1908,6 +1909,12 @@ fn vector_series( // A sort hands back the series it was given, rebuilding only those `sort` and `sort_desc` // drop histogram samples from. BoundVectorOp::Sort(order) => select::sort_instant(order, schema, series), + // The kernel groups bucket series by label set and takes one native histogram series per + // set, so pieces of one label set -- an empty tag and an absent one, or what a map folded + // -- are merged first, and two of them holding one point fail as they do elsewhere. + BoundVectorOp::Histogram(function) => { + evaluate_histograms(*function, schema, &merge_same_labelset(schema, series)?) + } BoundVectorOp::Absent { labels } => Ok(select::absent(schema, &series, labels.clone())? .into_iter() .collect()), @@ -1926,6 +1933,9 @@ fn vector_series( /// - A sort: the value each series is keyed by, and the floats `sort` and `sort_desc` rebuild a /// series into when it holds histograms; the label sorts move the series they were given. /// - `absent`: a presence flag and a result value per grid point and field. +/// - `histogram_quantile` and `histogram_fraction`: the merge of label-set pieces before the kernel +/// ([`merge_workspace_bytes`]); per series, its group entry and label pairs and a cursor per +/// field; per row, the bucket it contributes and a result value per field. fn vector_workspace_bytes(op: &BoundVectorOp, schema: &SeriesSchema, series: &[Series]) -> usize { let fields = schema.fields.len(); match op { @@ -1953,6 +1963,22 @@ fn vector_workspace_bytes(op: &BoundVectorOp, schema: &SeriesSchema, series: &[S }); return points.saturating_mul(fields.saturating_mul(1 + std::mem::size_of::())); } + BoundVectorOp::Histogram(_) => { + let rows: usize = series.iter().map(|item| item.time.len()).sum(); + let label = std::mem::size_of::<(Arc, Arc)>(); + let row = std::mem::size_of::<(f64, f64)>() + fields * std::mem::size_of::(); + let per_series = std::mem::size_of::<(f64, &Series)>() + + fields * std::mem::size_of::>(); + return series.iter().fold( + merge_workspace_bytes(rows, series.len(), fields), + |total, item| { + total + .saturating_add(per_series) + .saturating_add((item.metric.tags().len() + 1).saturating_mul(label)) + .saturating_add(item.time.len().saturating_mul(row)) + }, + ); + } BoundVectorOp::TopK { .. } | BoundVectorOp::LimitK { .. } | BoundVectorOp::LimitRatio { .. } => {} @@ -3202,8 +3228,9 @@ impl NativeExecutionState { // set stay pieces: they hold disjoint points, so ranking them apart picks // what ranking the merged series would, and what is above treats them as // it treats a map's. `absent` answers with one series under the labels it - // was bound to, and `limitk` counts series in input order, so it merges the - // pieces first and hands over whole series. + // was bound to, the histogram functions merge the label sets they answer + // with, and `limitk` counts series in input order, so it merges the pieces + // first; none of them hands over pieces. let op = match expr { NativeExpr::Vector { op, .. } => Some(op), _ => None, @@ -3215,7 +3242,11 @@ impl NativeExecutionState { { vector.merge_input = true; } - let whole = limitk || matches!(op, Some(BoundVectorOp::Absent { .. })); + let whole = limitk + || matches!( + op, + Some(BoundVectorOp::Absent { .. } | BoundVectorOp::Histogram(_)) + ); holds_shared[index] = input_holds_shared && !whole; } NodeTag::Window => { diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index 088c68f378c..8adc04d3b38 100644 --- a/src/promql/src/native/execute/tests.rs +++ b/src/promql/src/native/execute/tests.rs @@ -40,6 +40,7 @@ use super::{ }; use crate::evaluated::EvaluatedChunk; use crate::grid_series::merge_same_labelset; +use crate::histogram_eval::HistogramFunction; use crate::instant_eval::select_instants; use crate::map::MathOp; use crate::native::metrics::{NodeMetric, NodeMetrics, NodePath}; @@ -7600,3 +7601,99 @@ fn limitk_keeps_the_order_a_sort_below_it_decided() { vec![("z".to_string(), vec![(0, Some(2.0))])] ); } + +/// `histogram_quantile` reads a classic histogram from the bucket series that share its labels, +/// however they arrived, and answers under those labels without `le` and the name. +#[test] +fn histogram_quantile_reads_the_buckets_of_the_whole_vector() { + let grid = grid(0, 60_000, 1); + let schema = evaluated_schema(Arc::clone(&grid)); + let tree = NativeTree::try_new(vector_op( + evaluated_input(Arc::clone(&grid)), + BoundVectorOp::Histogram(HistogramFunction::Quantile(0.5)), + )) + .unwrap(); + let bucket = |le: &str| { + MetricName::try_new( + Some("m".into()), + vec![ + Label { + name: "job".into(), + value: "a".into(), + }, + Label { + name: "le".into(), + value: le.into(), + }, + ], + ) + .unwrap() + }; + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + for (le, count) in [("+Inf", 2.0), ("1", 1.0)] { + execution + .push_evaluated(evaluated_chunk_of( + &schema, + bucket(le), + 0, + vec![Some(count)], + true, + )) + .unwrap(); + } + let chunks = run(&mut execution); + assert_eq!(chunks.len(), 1); + assert_eq!(chunks[0].series.metric.name(), None); + assert_eq!(chunks[0].series.metric.get("le"), None); + assert_eq!(chunks[0].series.metric.get("job"), Some("a")); + assert_eq!(decided(&chunks), vec![(0, Some(1.0))]); +} + +/// An empty tag and an absent one are one label set, so the two native histogram series below are +/// one series to `histogram_quantile`, answered at both of their points rather than one of them +/// dropped. +#[test] +fn histogram_quantile_reads_the_pieces_of_one_label_set_as_one_series() { + let grid = grid(0, 60_000, 2); + let schema = evaluated_schema(Arc::clone(&grid)); + let tree = NativeTree::try_new(vector_op( + evaluated_input(Arc::clone(&grid)), + BoundVectorOp::Histogram(HistogramFunction::Quantile(0.5)), + )) + .unwrap(); + let metric = |tags: &[(&str, &str)]| { + MetricName::try_new( + Some("m".into()), + tags.iter() + .map(|(name, value)| Label { + name: (*name).into(), + value: (*value).into(), + }) + .collect(), + ) + .unwrap() + }; + let table = histogram_table(2, 4); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + for (start, tags, index) in [ + (0, vec![("job", "a"), ("zone", "")], 0), + (1, vec![("job", "a")], 1), + ] { + execution + .push_evaluated(evaluated_histogram_chunk_of( + &schema, + metric(&tags), + start, + vec![Some(index)], + &table, + true, + )) + .unwrap(); + } + let chunks = run(&mut execution); + let positions: Vec = decided(&chunks) + .into_iter() + .filter_map(|(position, value)| value.map(|_| position)) + .collect(); + assert_eq!(positions, [0, 1]); +} diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs index 352dc755996..48ffdf52d3a 100644 --- a/src/promql/src/native/plan.rs +++ b/src/promql/src/native/plan.rs @@ -35,6 +35,7 @@ use datatypes::series::samples::{FieldSpec, SeriesSchema}; use datatypes::series::time::EvalGrid; use datatypes::series::{MetricName, SeriesOrdering}; +use crate::histogram_eval::HistogramFunction; use crate::map::MathOp; use crate::range_eval::{RangeFunction, RangeWindow}; use crate::select::SortOrder; @@ -256,8 +257,9 @@ pub struct BoundAggregate { pub stage: AggregateStage, } -/// An operator that keeps some of the input's samples at each grid point, with their series' labels -/// and values: which samples it keeps depends on the whole vector at that point, not on one series. +/// An operator over the whole input vector at each grid point: one that keeps some of the input's +/// samples with their series' labels and values, orders the series, reports where none is present, +/// or reads a classic histogram from its bucket series. /// /// Its constants are bound here, as a window's are: a `k` or a ratio only known at run time has no /// field. @@ -277,6 +279,9 @@ pub enum BoundVectorOp { /// `absent`: `1` under `labels` at every point where the input holds no sample. The labels are /// what the compiler derives from the argument's matchers. Absent { labels: MetricName }, + /// `histogram_quantile` or `histogram_fraction`: each classic histogram is read from the bucket + /// series that share its labels, so this too reads the whole vector at a point. + Histogram(HistogramFunction), } /// What a node computes. Every variant holds a bound operation, never a name to resolve. diff --git a/src/query/src/promql/native_compile.rs b/src/query/src/promql/native_compile.rs index 1bc667dba3a..53bb3a187a0 100644 --- a/src/query/src/promql/native_compile.rs +++ b/src/query/src/promql/native_compile.rs @@ -29,6 +29,7 @@ //! ([, ]*) e.g. abs(rate(x[5m])), clamp_min(x, 0) //! ([, ]*) e.g. sort_desc(x), sort_by_label(x, "job") //! absent() e.g. absent(x{job="a"}) +//! (+, ) e.g. histogram_quantile(0.9, rate(x[5m])) //! e.g. x * 100, x > bool 0 //! e.g. 1 / x, 0 < x //! - e.g. -x @@ -124,6 +125,7 @@ use datafusion::error::{DataFusionError, Result}; use datatypes::series::samples::SeriesSchema; use datatypes::series::time::{EvalGrid, GridSpec}; use datatypes::series::{MetricName, SeriesOrdering}; +use promql::histogram_eval::HistogramFunction; use promql::map::{ArithmeticOp, ComparisonOp, MathOp}; use promql::map_time::DateFunction; use promql::native::plan::{ @@ -163,8 +165,10 @@ const SUPPORTED: &str = "a selector — `x`, optionally with an `offset` — a r that folds a group's values into one value per grid point — `sum`, `avg`, \ `count`, `min`, `max`, `group`, `stddev`, `stdvar` and \ `quantile(, )` — over any of them, optionally \ - grouped by `by` or `without` as in `avg by(job)(rate(x[5m]) * 100)`; a sort or `absent` over any of them \ - (`sort_desc(x)`, `absent(x{job=\"a\"})`); a range function's own \ + grouped by `by` or `without` as in `avg by(job)(rate(x[5m]) * 100)`; a \ + sort, `absent` or `histogram_quantile` / `histogram_fraction` over any of \ + them (`sort_desc(x)`, `absent(x{job=\"a\"})`, \ + `histogram_quantile(0.9, rate(x[5m]))`); a range function's own \ constants sit where that function takes them, before \ the matrix for `quantile_over_time(0.9, x[5m])` and after it for \ `predict_linear(x[5m], 300)` and \ @@ -690,7 +694,7 @@ fn vector_call( ) -> Result { if vector.value_type() != ValueType::Vector { return Err(not_implemented(&format!( - "a call to `{name}` whose first argument is `{vector}`, which is a {}, where the \ + "a call to `{name}` whose vector argument is `{vector}`, which is a {}, where the \ function takes a vector", vector.value_type() ))); @@ -706,8 +710,8 @@ fn vector_call( }) } -/// The vector operation a sort or `absent` binds, and the vector it reads, or `None` when `name` is -/// neither. +/// The vector operation a sort, `absent` or a classic histogram function binds, and the vector it reads, +/// or `None` when `name` is none of them. fn vector_op<'a>( name: &str, arguments: &'a [Box], @@ -716,6 +720,29 @@ fn vector_op<'a>( "sort" => Some(SortOrder::Asc), "sort_desc" => Some(SortOrder::Desc), "sort_by_label" | "sort_by_label_desc" => None, + "histogram_quantile" | "histogram_fraction" => { + let Some((vector, constants)) = arguments.split_last() else { + return Err(not_implemented(&format!( + "a call to `{name}` with no argument" + ))); + }; + let function = match constants { + [quantile] if name == "histogram_quantile" => { + HistogramFunction::Quantile(constant_argument(name, quantile, 0)?) + } + [lower, upper] if name == "histogram_fraction" => HistogramFunction::Fraction { + lower: constant_argument(name, lower, 0)?, + upper: constant_argument(name, upper, 1)?, + }, + _ => { + return Err(not_implemented(&format!( + "a call to `{name}` with {} argument(s)", + arguments.len() + ))); + } + }; + return Ok(Some((BoundVectorOp::Histogram(function), vector))); + } "absent" => { let [vector] = arguments else { return Err(not_implemented(&format!( diff --git a/src/query/src/promql/native_compile/tests.rs b/src/query/src/promql/native_compile/tests.rs index 97cc3267067..d131af69ba9 100644 --- a/src/query/src/promql/native_compile/tests.rs +++ b/src/query/src/promql/native_compile/tests.rs @@ -3682,13 +3682,14 @@ fn a_subquery_grid_holds_every_sample_its_windows_read() { } } -/// The sorts and `absent` bind a vector operation over the vector they are given. `absent` derives its +/// The sorts, `absent` and the classic histogram functions bind a vector operation over the vector they +/// are given, the histogram functions with their constants folded. `absent` derives its /// labels from the selector's matchers as Prometheus does: the first equality of a name sets the /// label and any other matcher of it removes it, the matchers that accept a missing label read /// last; `__name__` and the table-choosing matchers are not labels, and any expression other than /// a selector answers without labels. #[test] -fn sorts_and_absent_bind_a_vector_operation() { +fn whole_vector_functions_bind_a_vector_operation() { let cases = [ ("sort(x)", "sort"), ("sort_desc(rate(x[5m]))", "sort_desc"), @@ -3707,6 +3708,11 @@ fn sorts_and_absent_bind_a_vector_operation() { // `job!="b"` accepts a missing `job`, so it reads after `job="a"` and removes it. (r#"absent(x{job!="b", job="a"})"#, "absent {}"), (r#"absent({__name__="x", job="a"})"#, r#"absent {job="a"}"#), + ( + "histogram_quantile(1 - 0.1, rate(x[5m]))", + "histogram_quantile 0.9", + ), + ("histogram_fraction(0, 0.5, x)", "histogram_fraction 0 0.5"), ("absent(rate(x[5m]))", "absent {}"), ]; for (query, op) in cases {