Add histogram_quantile() PromQL query for percentiles

The PromQL layer only had a fixed-width histogram() (per-bucket
counts), with no way to ask for p50/p95/p99. Adds
histogram_quantile(p, metric[range]), interpolating over each
series's retained raw sample values (linear interpolation between
closest ranks) rather than building a separate pre-aggregated
bucketed-histogram representation — exact for the data actually kept.
An optional bracketed range narrows the computation to a trailing
window, mirroring rate()'s range selector.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ

src/engine/promql.rs

@@ -8,6 +8,7 @@ pub enum AggOp {
     Rate,
     Histogram,
     Gauge,
+    Quantile,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -23,6 +24,9 @@ pub struct PromqlQuery {
     pub matchers: Vec<LabelMatcher>,
     pub range: Option<Duration>,
     pub bucket_width: Option<f64>,
+    /// The `p` in `histogram_quantile(p, metric[range])`, e.g. `0.95` for p95. Only set when
+    /// `agg` is `Some(AggOp::Quantile)`.
+    pub quantile: Option<f64>,
     pub inner_rate: bool,
     pub by_labels: Vec<String>,
 }
@@ -51,6 +55,13 @@ impl PromqlQuery {
                 "histogram() requires a bucket width, e.g. histogram(my_metric[10])".to_string(),
             );
         }
+        if query.agg == Some(AggOp::Quantile) && query.quantile.is_none() {
+            return Err(
+                "histogram_quantile() requires a quantile between 0 and 1 and a metric, e.g. \
+                 histogram_quantile(0.95, latency_ms)"
+                    .to_string(),
+            );
+        }
         Ok(query)
     }
 }
@@ -225,11 +236,42 @@ impl<'a> Parser<'a> {
             matchers,
             range,
             bucket_width,
+            quantile: None,
             inner_rate: false,
             by_labels: Vec::new(),
         })
     }
 
+    /// Parses a decimal number (e.g. `0`, `1`, `0.95`) — used for `histogram_quantile(p, ...)`'s
+    /// `p` argument, which isn't an integer like the range-selector widths elsewhere in this grammar.
+    fn parse_float(&mut self) -> Result<f64, String> {
+        self.skip_whitespace();
+        let rest = self.remaining();
+        let mut end = 0;
+        let mut seen_digit = false;
+        let mut seen_dot = false;
+        for (idx, c) in rest.char_indices() {
+            if c.is_ascii_digit() {
+                seen_digit = true;
+                end = idx + c.len_utf8();
+            } else if c == '.' && !seen_dot {
+                seen_dot = true;
+                end = idx + c.len_utf8();
+            } else {
+                break;
+            }
+        }
+        if !seen_digit {
+            return Err("expected a number".to_string());
+        }
+        let text = &rest[..end];
+        let value: f64 = text
+            .parse()
+            .map_err(|_| format!("'{text}' is not a valid number"))?;
+        self.pos += end;
+        Ok(value)
+    }
+
     fn parse_by_clause(&mut self) -> Result<Vec<String>, String> {
         self.skip_whitespace();
         let checkpoint = self.pos;
@@ -276,6 +318,7 @@ impl<'a> Parser<'a> {
             "rate" => Some(AggOp::Rate),
             "histogram" => Some(AggOp::Histogram),
             "gauge" => Some(AggOp::Gauge),
+            "histogram_quantile" => Some(AggOp::Quantile),
             _ => None,
         };
         self.skip_whitespace();
@@ -288,7 +331,20 @@ impl<'a> Parser<'a> {
         if agg.is_some() && self.peek() == Some('(') {
             self.expect_char('(')?;
             self.skip_whitespace();
-            let mut inner = if matches!(agg, Some(AggOp::Sum) | Some(AggOp::Avg)) {
+            let mut inner = if agg == Some(AggOp::Quantile) {
+                let quantile = self.parse_float()?;
+                if !(0.0..=1.0).contains(&quantile) {
+                    return Err(format!(
+                        "histogram_quantile() quantile must be between 0 and 1, got {quantile}"
+                    ));
+                }
+                self.skip_whitespace();
+                self.expect_char(',')?;
+                self.skip_whitespace();
+                let mut vector = self.parse_vector_selector()?;
+                vector.quantile = Some(quantile);
+                vector
+            } else if matches!(agg, Some(AggOp::Sum) | Some(AggOp::Avg)) {
                 let inner_checkpoint = self.pos;
                 let took_inner_rate = self.parse_identifier().is_ok_and(|id| id == "rate");
                 if took_inner_rate {
@@ -591,4 +647,50 @@ mod tests {
         let err = PromqlQuery::parse("sum by (path) (requests) by (method)").unwrap_err();
         assert!(err.contains("only appear once"));
     }
+
+    #[test]
+    fn parses_histogram_quantile() {
+        let query = PromqlQuery::parse("histogram_quantile(0.95, latency_ms)").unwrap();
+        assert_eq!(query.agg, Some(AggOp::Quantile));
+        assert_eq!(query.metric_name, "latency_ms");
+        assert_eq!(query.quantile, Some(0.95));
+        assert_eq!(query.range, None);
+    }
+
+    #[test]
+    fn parses_histogram_quantile_with_range_and_matchers() {
+        let query =
+            PromqlQuery::parse(r#"histogram_quantile(0.5, latency_ms{path="/a"}[5m])"#).unwrap();
+        assert_eq!(query.quantile, Some(0.5));
+        assert_eq!(query.range, Some(Duration::from_secs(300)));
+        assert_eq!(
+            query.matchers,
+            vec![LabelMatcher {
+                name: "path".to_string(),
+                value: "/a".to_string()
+            }]
+        );
+    }
+
+    #[test]
+    fn parses_histogram_quantile_with_by_clause() {
+        let query = PromqlQuery::parse("histogram_quantile(0.99, latency_ms) by (path)").unwrap();
+        assert_eq!(query.by_labels, vec!["path".to_string()]);
+    }
+
+    #[test]
+    fn rejects_histogram_quantile_missing_comma() {
+        assert!(PromqlQuery::parse("histogram_quantile(0.95 latency_ms)").is_err());
+    }
+
+    #[test]
+    fn rejects_histogram_quantile_out_of_range() {
+        let err = PromqlQuery::parse("histogram_quantile(1.5, latency_ms)").unwrap_err();
+        assert!(err.contains("between 0 and 1"));
+    }
+
+    #[test]
+    fn rejects_histogram_quantile_without_number() {
+        assert!(PromqlQuery::parse("histogram_quantile(latency_ms)").is_err());
+    }
 }

src/engine/snapshot/mod.rs

@@ -1287,10 +1287,76 @@ impl EngineSnapshot {
                     })
                     .collect()
             }
+            Some(AggOp::Quantile) => {
+                let quantile = parsed
+                    .quantile
+                    .expect("histogram_quantile() always carries a quantile");
+                // The bracketed range, if any, narrows each series to its trailing window
+                // (mirroring rate()'s range selector) rather than the full panel window —
+                // interpolating over whatever raw values are retained for that series, since
+                // this engine doesn't keep a separate pre-aggregated bucketed-histogram
+                // representation to compute quantiles from.
+                let range_ms = parsed.range.map(|d| d.as_millis() as i64);
+                let last_ts = series_points
+                    .values()
+                    .flat_map(|points| points.iter().map(|(ts, _)| *ts))
+                    .max()
+                    .unwrap_or(0);
+
+                let mut groups: BTreeMap<BTreeMap<String, String>, Vec<f64>> = BTreeMap::new();
+                for (labels, points) in &series_points {
+                    let group_key = project_labels(labels, &parsed.by_labels);
+                    let window_start = match range_ms {
+                        Some(range_ms) => {
+                            let window_end = points.last().map(|(ts, _)| *ts).unwrap_or(0);
+                            Some(window_end - range_ms)
+                        }
+                        None => None,
+                    };
+                    groups.entry(group_key).or_default().extend(
+                        points
+                            .iter()
+                            .filter(|(ts, _)| window_start.is_none_or(|start| *ts >= start))
+                            .map(|(_, value)| *value),
+                    );
+                }
+
+                groups
+                    .into_iter()
+                    .filter_map(|(labels, mut values)| {
+                        let quantile_value = interpolated_quantile(&mut values, quantile)?;
+                        Some(PromqlSeriesDto {
+                            labels,
+                            points: vec![(last_ts, quantile_value)],
+                        })
+                    })
+                    .collect()
+            }
         })
     }
 }
 
+/// Linear interpolation between closest ranks over the raw retained sample values (the "R-7"
+/// method, matching numpy/Excel's default `PERCENTILE` behavior) — exact for the data actually
+/// kept, unlike a bucketed-histogram approximation.
+fn interpolated_quantile(values: &mut [f64], quantile: f64) -> Option<f64> {
+    if values.is_empty() {
+        return None;
+    }
+    values.sort_by(|a, b| a.partial_cmp(b).unwrap());
+    if values.len() == 1 {
+        return Some(values[0]);
+    }
+    let rank = quantile * (values.len() - 1) as f64;
+    let lower = rank.floor() as usize;
+    let upper = rank.ceil() as usize;
+    if lower == upper {
+        return Some(values[lower]);
+    }
+    let fraction = rank - lower as f64;
+    Some(values[lower] + (values[upper] - values[lower]) * fraction)
+}
+
 // --- sort: sort-helper index maintenance ---
 impl EngineSnapshot {
     pub(super) fn build_sort_helper_index(

src/engine/tests.rs

@@ -1724,6 +1724,67 @@ fn rate_and_sum_over_occurrence_count_metrics_are_unaffected_by_the_semantics_gu
     );
 }
 
+#[test]
+fn histogram_quantile_interpolates_over_retained_sample_values() {
+    let mut index = EngineSnapshot::default();
+    declare_gauge_metric_with_value(&mut index);
+    for (sequence, depth) in [(0u64, 1), (1, 2), (2, 3), (3, 4), (4, 5)] {
+        ingest_log(&mut index, sequence, &format!("queue depth={depth}"));
+    }
+
+    let parsed = PromqlQuery::parse("histogram_quantile(0.5, queue_depth)").unwrap();
+    let series = index.evaluate_promql(&parsed, None, None).unwrap();
+
+    assert_eq!(series.len(), 1);
+    assert_eq!(series[0].points.len(), 1);
+    assert_eq!(series[0].points[0].1, 3.0);
+}
+
+#[test]
+fn histogram_quantile_p0_and_p1_return_the_min_and_max() {
+    let mut index = EngineSnapshot::default();
+    declare_gauge_metric_with_value(&mut index);
+    for (sequence, depth) in [(0u64, 10), (1, 20), (2, 30)] {
+        ingest_log(&mut index, sequence, &format!("queue depth={depth}"));
+    }
+
+    let min_parsed = PromqlQuery::parse("histogram_quantile(0, queue_depth)").unwrap();
+    let max_parsed = PromqlQuery::parse("histogram_quantile(1, queue_depth)").unwrap();
+    assert_eq!(
+        index.evaluate_promql(&min_parsed, None, None).unwrap()[0].points[0].1,
+        10.0
+    );
+    assert_eq!(
+        index.evaluate_promql(&max_parsed, None, None).unwrap()[0].points[0].1,
+        30.0
+    );
+}
+
+#[test]
+fn histogram_quantile_range_narrows_to_the_trailing_window() {
+    let mut index = EngineSnapshot::default();
+    declare_gauge_metric_with_value(&mut index);
+    // Values far outside the trailing 2s window should not affect the result.
+    ingest_log(&mut index, 0, "queue depth=1000");
+    ingest_log(&mut index, 5000, "queue depth=10");
+    ingest_log(&mut index, 6000, "queue depth=20");
+
+    let parsed = PromqlQuery::parse("histogram_quantile(0.5, queue_depth[2s])").unwrap();
+    let series = index.evaluate_promql(&parsed, None, None).unwrap();
+
+    assert_eq!(series[0].points[0].1, 15.0);
+}
+
+#[test]
+fn histogram_quantile_on_metric_with_no_data_returns_no_series() {
+    let mut index = EngineSnapshot::default();
+    declare_gauge_metric_with_value(&mut index);
+
+    let parsed = PromqlQuery::parse("histogram_quantile(0.5, queue_depth)").unwrap();
+    let series = index.evaluate_promql(&parsed, None, None).unwrap();
+    assert!(series.is_empty());
+}
+
 #[test]
 fn bare_query_over_string_kind_metric_is_still_allowed() {
     let mut index = EngineSnapshot::default();