chore: merge narthex into feat/narthex-select-compile

src/promql/src/histogram_eval.rs

@@ -97,7 +97,17 @@ pub fn evaluate_histograms(
     schema: &Arc<SeriesSchema>,
     series: &[Series],
 ) -> Result<Vec<Series>> {
-    let mut groups: HashMap<LabelSet, Group<'_>> = 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<LabelSet, usize> = HashMap::new();
+    let mut groups: Vec<Group<'_>> = 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<f64>, &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

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!(
             "{} {} [{}] {}",

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<SeriesSchema>,
     series: Vec<Series>,
 ) -> Result<Vec<Series>> {
     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::<f64>()));
         }
+        BoundVectorOp::Histogram(_) => {
+            let rows: usize = series.iter().map(|item| item.time.len()).sum();
+            let label = std::mem::size_of::<(Arc<str>, Arc<str>)>();
+            let row = std::mem::size_of::<(f64, f64)>() + fields * std::mem::size_of::<f64>();
+            let per_series = std::mem::size_of::<(f64, &Series)>()
+                + fields * std::mem::size_of::<RowCursor<'_>>();
+            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 => {

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<usize> = decided(&chunks)
+        .into_iter()
+        .filter_map(|(position, value)| value.map(|_| position))
+        .collect();
+    assert_eq!(positions, [0, 1]);
+}

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.

src/query/src/promql/native_compile.rs

@@ -29,6 +29,7 @@
 //!   <math function>(<expression>[, <constant>]*)   e.g. abs(rate(x[5m])), clamp_min(x, 0)
 //!   <sort>(<expression>[, <string>]*)              e.g. sort_desc(x), sort_by_label(x, "job")
 //!   absent(<expression>)                           e.g. absent(x{job="a"})
+//!   <histogram>(<constant>+, <expression>)         e.g. histogram_quantile(0.9, rate(x[5m]))
 //!   <expression> <operator> <constant>             e.g. x * 100, x > bool 0
 //!   <constant> <operator> <expression>             e.g. 1 / x, 0 < x
 //!   -<expression>                                  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(<constant>, <expression>)` — 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<NativeNode> {
     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<PromExpr>],
@@ -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!(

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 {