From 8bc2c8aa55520846ae5a12713b88611a58f0289a Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Thu, 22 Oct 2026 05:26:27 +0800 Subject: [PATCH 1/2] feat(promql): run histogram_quantile and histogram_fraction in the native select node Signed-off-by: Dennis Zhuang --- src/promql/src/histogram_eval.rs | 52 +++++++---- src/promql/src/native/display.rs | 7 ++ src/promql/src/native/execute.rs | 39 ++++++-- src/promql/src/native/execute/tests.rs | 97 ++++++++++++++++++++ src/promql/src/native/plan.rs | 9 +- src/query/src/promql/native_compile.rs | 40 ++++++-- src/query/src/promql/native_compile/tests.rs | 10 +- 7 files changed, 217 insertions(+), 37 deletions(-) diff --git a/src/promql/src/histogram_eval.rs b/src/promql/src/histogram_eval.rs index 4b7fdba6c54..c0df39c012c 100644 --- a/src/promql/src/histogram_eval.rs +++ b/src/promql/src/histogram_eval.rs @@ -69,7 +69,18 @@ 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| { + let slot = *index.entry(label_set(&metric)).or_insert_with(|| { + groups.push(Group::new(metric)); + groups.len() - 1 + }); + slot + }; + let mut members: Vec<(usize, Option, &Series)> = Vec::new(); for series in series { grid_axis(series)?; if let Some(upper_bound) = series @@ -79,26 +90,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); + } } } @@ -107,7 +119,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 2124f692aec..9fd4d98fa41 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 { } }, BoundSelect::Absent { labels } => format!("absent {{{}}}", label_pairs(labels)), + BoundSelect::Histogram(HistogramFunction::Quantile(quantile)) => { + format!("histogram_quantile {quantile}") + } + BoundSelect::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 66b17c63f97..9616c248d33 100644 --- a/src/promql/src/native/execute.rs +++ b/src/promql/src/native/execute.rs @@ -94,6 +94,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::{MathOp, drop_metric_name, map_math}; use crate::native::batch::EvaluatedChunk; @@ -1861,7 +1862,7 @@ fn select_overflow(path: &NodePath) -> DataFusionError { /// Runs the kernel `op` binds over the whole input. fn select_series( op: &BoundSelect, - schema: &SeriesSchema, + schema: &Arc, series: Vec, ) -> Result> { let grouping = |group: &Grouping| match group { @@ -1877,6 +1878,12 @@ fn select_series( // A sort hands back the series it was given, rebuilding only those `sort` and `sort_desc` // drop histogram samples from. BoundSelect::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. + BoundSelect::Histogram(function) => { + evaluate_histograms(*function, schema, &merge_same_labelset(schema, series)?) + } BoundSelect::Absent { labels } => Ok(select::absent(schema, &series, labels.clone())? .into_iter() .collect()), @@ -1895,6 +1902,9 @@ fn select_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 selection_workspace_bytes(op: &BoundSelect, schema: &SeriesSchema, series: &[Series]) -> usize { let fields = schema.fields.len(); match op { @@ -1923,6 +1933,22 @@ fn selection_workspace_bytes(op: &BoundSelect, schema: &SeriesSchema, series: &[ .map_or(0, |grid| grid.timestamps().len()); return points.saturating_mul(fields.saturating_mul(1 + std::mem::size_of::())); } + BoundSelect::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)) + }, + ); + } BoundSelect::TopK { .. } | BoundSelect::LimitK { .. } | BoundSelect::LimitRatio { .. } => {} } let row = std::mem::size_of::<(usize, usize)>() @@ -2946,16 +2972,17 @@ impl NativeExecutionState { // A selection keeps samples of the series it read, so pieces of one label // 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` is the exception: it answers with one series - // under the labels it was bound to. - let absent = matches!( + // it treats a map's. `absent` answers with one series under the labels it + // was bound to, and the histogram functions merge the label sets they + // answer with, so neither hands over pieces. + let whole = matches!( expr, NativeExpr::Select { - op: BoundSelect::Absent { .. }, + op: BoundSelect::Absent { .. } | BoundSelect::Histogram(_), .. } ); - holds_shared[index] = input_holds_shared && !absent; + holds_shared[index] = input_holds_shared && !whole; } NodeTag::Window => { let (drops_metric_name, input_schema) = match expr { diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index b77e35e5685..792ced48d15 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}; @@ -6708,3 +6709,99 @@ fn a_sorted_series_is_queued_at_what_its_slices_hold() { assert_eq!(decided(&chunks).len(), points as usize); assert_eq!(pool.reserved(), 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(select( + evaluated_input(Arc::clone(&grid)), + BoundSelect::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(select( + evaluated_input(Arc::clone(&grid)), + BoundSelect::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 8e827b78819..92a816dc748 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; @@ -223,8 +224,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. @@ -244,6 +246,9 @@ pub enum BoundSelect { /// `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 f731c571427..64cf5ca1b11 100644 --- a/src/query/src/promql/native_compile.rs +++ b/src/query/src/promql/native_compile.rs @@ -25,12 +25,14 @@ //! ([, ]*) 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])) //! ``` //! //! which become `Input` (raw samples) → `Instant` (the last sample within the statement's lookback), //! `Input` (raw samples) → `Window` (the range function over the statement's grid) and //! `` → `Map` (a pointwise math function over what its argument evaluates to), and -//! `` → `Select` (a sort or `absent`, which read the whole vector). A +//! `` → `Select` (a sort, `absent` or a classic histogram function, which read the +//! whole vector). A //! selector may carry `offset` in either selection shape and, in the window shape, an `@` anchor. //! Everything else is a [`DataFusionError::NotImplemented`] that names what was found and quotes the //! shapes above, because rejecting a query the compiler does not understand is the only honest @@ -59,6 +61,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, MathOp}; use promql::native::plan::{ BoundInput, BoundInstant, BoundMap, BoundSelect, BoundWindow, NativeExpr, NativeNode, @@ -80,9 +83,9 @@ use crate::promql::label_values::signed_millis_since_epoch; const SUPPORTED: &str = "a selector, a range function over a range selector — \ `([])`, with its constants in front of \ the selector as in `quantile_over_time(0.9, x[5m])` — or a math function, \ - a sort or `absent` over any of them, as in `abs(rate(x[5m]))`: for \ - example `x`, `rate(x[5m])`, `quantile_over_time(0.9, x[5m])` or \ - `abs(rate(x[5m]))`"; + a sort, `absent` or a histogram function over any of them, as in \ + `abs(rate(x[5m]))`: for example `x`, `rate(x[5m])`, \ + `quantile_over_time(0.9, x[5m])` or `abs(rate(x[5m]))`"; /// Compiles `stmt` into a tree that reads `input` and evaluates the statement's expression over it. /// @@ -266,7 +269,7 @@ fn select_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() ))); @@ -282,8 +285,8 @@ fn select_call( }) } -/// The selection a sort or `absent` binds, and the vector it reads, or `None` when `name` is -/// neither. +/// The selection a sort, `absent` or a classic histogram function binds, and the vector it reads, +/// or `None` when `name` is none of them. fn select_op<'a>( name: &str, arguments: &'a [Box], @@ -292,6 +295,29 @@ fn select_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((BoundSelect::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 f4d4b990d58..f6991e29988 100644 --- a/src/query/src/promql/native_compile/tests.rs +++ b/src/query/src/promql/native_compile/tests.rs @@ -1160,13 +1160,14 @@ fn a_math_function_over_a_non_vector_argument_is_refused_and_names_it() { } } -/// The sorts and `absent` bind a selection over the vector they are given. `absent` derives its +/// The sorts, `absent` and the classic histogram functions bind a selection 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_selection() { +fn whole_vector_functions_bind_a_selection() { let cases = [ ("sort(x)", "sort"), ("sort_desc(rate(x[5m]))", "sort_desc"), @@ -1185,6 +1186,11 @@ fn sorts_and_absent_bind_a_selection() { // `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 { From 7e0325780eebaaf20fed13b3d5a403628a31cb86 Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Thu, 22 Oct 2026 05:40:15 +0800 Subject: [PATCH 2/2] chore(promql): return the group slot directly Signed-off-by: Dennis Zhuang --- src/promql/src/histogram_eval.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/promql/src/histogram_eval.rs b/src/promql/src/histogram_eval.rs index c0df39c012c..464fc0251cd 100644 --- a/src/promql/src/histogram_eval.rs +++ b/src/promql/src/histogram_eval.rs @@ -74,11 +74,10 @@ pub fn evaluate_histograms( let mut index: HashMap = HashMap::new(); let mut groups: Vec> = Vec::new(); let mut group_of = |metric: MetricName| { - let slot = *index.entry(label_set(&metric)).or_insert_with(|| { + *index.entry(label_set(&metric)).or_insert_with(|| { groups.push(Group::new(metric)); groups.len() - 1 - }); - slot + }) }; let mut members: Vec<(usize, Option, &Series)> = Vec::new(); for series in series {