From f2673ce10b3cf928b6090a2b8aaaf39b8bfd6edc Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Wed, 21 Oct 2026 17:46:56 +0800 Subject: [PATCH] perf(promql): resolve info signatures per series and cover the review gaps Signed-off-by: Dennis Zhuang --- src/promql/src/info.rs | 205 +++++++++++++++-------- src/promql/src/info/prometheus_cases.txt | 49 +++++- src/promql/src/info/tests.rs | 129 +++++++++++++- 3 files changed, 308 insertions(+), 75 deletions(-) diff --git a/src/promql/src/info.rs b/src/promql/src/info.rs index 69ce9fddf45..af619b9d280 100644 --- a/src/promql/src/info.rs +++ b/src/promql/src/info.rs @@ -56,8 +56,9 @@ pub struct InfoSelector { /// /// `info` holds the info series the selector chooses, one float field each, whose value at a /// point is the original timestamp of the sample selected there. For each info metric a base -/// series takes the info series with the same identifying labels; when several are present at a -/// point the one with the newest timestamp counts, and two equally new ones are an error. Labels +/// series takes the info series with the same identifying labels. Several of them at one point are +/// compared in input order, as Prometheus compares them in the order its storage returns them: a +/// newer one replaces the one kept so far, and one exactly as new is an error. Labels /// the base series already has are kept, and two info metrics giving one label different values /// are an error. A base series without identifying labels matches no info series, since /// Prometheus fetches none for it, and a base series that is itself an info metric is passed @@ -105,9 +106,7 @@ pub fn info( } } - // Errors as `(point, rank, message)`: Prometheus fails at the first failing step, and within - // a step checks the info series, then each base series, then the result label sets. - let mut errors: Vec<(usize, u8, String)> = Vec::new(); + let mut error = FirstError::default(); // Prometheus fetches info series per presence pattern of the identifying labels among the // base series it enriches: a present label matches any value the pattern's base series have, @@ -131,7 +130,7 @@ pub fn info( } } - // The newest info series of each signature at each point. + // The info series kept for each signature at each point. let mut by_signature: HashMap> = HashMap::new(); let mut info_names: BTreeSet> = BTreeSet::new(); for (index, series) in info.iter().enumerate() { @@ -147,21 +146,19 @@ pub fn info( } let name: Arc = Arc::from(series.metric.name().unwrap_or("")); info_names.insert(name.clone()); - let newest = by_signature - .entry(signature(name, &series.metric)) - .or_default(); + let kept = by_signature.entry((name, values)).or_default(); let mut cursor = series.fields.as_slice()[0].cursor(); for (point, row) in series.time.positions() { let timestamp = match cursor.at(row).expect("the axis and the fields are aligned") { SampleRef::Missing => continue, SampleRef::Float(timestamp) => timestamp, SampleRef::Histogram(_) => { - return Err(DataFusionError::Execution( - "info sample should be float".into(), + return Err(DataFusionError::Internal( + "info series must hold the original sample timestamps as floats".into(), )); } }; - match newest.entry(point) { + match kept.entry(point) { Entry::Vacant(entry) => { entry.insert((timestamp, index)); } @@ -170,18 +167,16 @@ pub fn info( if timestamp > existing { entry.insert((timestamp, index)); } else if timestamp == existing && base_present[point] { - errors.push(( - point, - 0, + error.record(point, 0, || { format!( - "found duplicate series for info metric: existing {:?} @ {}, \ - new {:?} @ {}", - label_set(&info[existing_index].metric), + "found duplicate series for info metric: existing {} @ {}, new \ + {} @ {}", + render(&label_set(&info[existing_index].metric)), existing as i64, - label_set(&series.metric), + render(&label_set(&series.metric)), timestamp as i64 - ), - )); + ) + }); } } } @@ -189,10 +184,25 @@ pub fn info( } let width = schema.fields.len(); - let mut results: HashMap = HashMap::new(); + let mut result_index: HashMap = HashMap::new(); + let mut results: Vec<(MetricName, Vec)> = Vec::new(); let mut added: Vec<(Arc, Arc)> = Vec::new(); + let mut chosen: Vec = Vec::new(); for series in base { let ignored = is_info_metric(series.metric.name().unwrap_or("")); + // The signature of each info metric is fixed for a series, so its kept series are looked + // up once; a point then only reads which of them is kept there. + let candidates: Vec<&BTreeMap> = if ignored { + Vec::new() + } else { + let values = identifying_values(&series.metric); + info_names + .iter() + .filter_map(|name| by_signature.get(&(name.clone(), values.clone()))) + .collect() + }; + // The result of the previous point, reused while the same info series are kept. + let mut previous: Option = None; let mut cursors: Vec<_> = series.fields.iter().map(|field| field.cursor()).collect(); for (point, row) in series.time.positions() { let mut values = Vec::with_capacity(width); @@ -204,72 +214,92 @@ pub fn info( if values.iter().all(Option::is_none) { continue; } - let metric = if ignored { - series.metric.clone() - } else { + chosen.clear(); + chosen.extend( + candidates + .iter() + .filter_map(|kept| kept.get(&point).map(|&(_, index)| index)), + ); + let reuse = previous + .as_ref() + .is_some_and(|previous| previous.info == chosen); + if !reuse { added.clear(); - let mut seen = 0; - for name in &info_names { - let Some(&(_, index)) = by_signature - .get(&signature(name.clone(), &series.metric)) - .and_then(|newest| newest.get(&point)) - else { - continue; - }; - seen += 1; + let mut conflict = None; + for &index in &chosen { if let Err(label) = add_data_labels(&info[index].metric, &series.metric, selector, &mut added) { - errors.push((point, 1, format!("conflicting label: {label}"))); + conflict = Some(label); + break; } } - if seen == 0 && !selector.keep_unmatched { - continue; - } - let mut metric = series.metric.clone(); - for (name, value) in &added { - metric.set(name, value.clone()).map_err(external)?; - } - metric - }; - let (_, rows) = results - .entry(label_set(&metric)) - .or_insert_with(|| (metric.clone(), BTreeMap::new())); - match rows.entry(point) { - Entry::Vacant(entry) => { - entry.insert(values); - } - Entry::Occupied(mut entry) => { - for (slot, value) in entry.get_mut().iter_mut().zip(values) { + let result = if chosen.is_empty() && !selector.keep_unmatched && !ignored { + None + } else { + let mut metric = series.metric.clone(); + for (name, value) in &added { + metric.set(name, value.clone()).map_err(external)?; + } + Some(match result_index.entry(label_set(&metric)) { + std::collections::hash_map::Entry::Occupied(entry) => *entry.get(), + std::collections::hash_map::Entry::Vacant(entry) => { + results.push((metric, Vec::new())); + *entry.insert(results.len() - 1) + } + }) + }; + previous = Some(Choice { + info: chosen.clone(), + result, + conflict, + }); + } + let choice = previous.as_ref().expect("set above"); + if let Some(label) = &choice.conflict { + error.record(point, 1, || format!("conflicting label: {label}")); + } + if let Some(result) = choice.result { + results[result].1.push((point, values)); + } + } + } + + // Rows of one label set arrive per base series; sorted by point, a field held twice at one + // point is a collision. + for (metric, rows) in &mut results { + rows.sort_by_key(|(point, _)| *point); + let mut merged: Vec = Vec::with_capacity(rows.len()); + for (point, values) in rows.drain(..) { + match merged.last_mut() { + Some((last, slots)) if *last == point => { + for (slot, value) in slots.iter_mut().zip(values) { if value.is_none() { continue; } if slot.is_some() { - errors.push(( - point, - 2, + error.record(point, 2, || { format!( - "vector cannot contain metrics with the same labelset {:?}", - label_set(&metric) - ), - )); + "vector cannot contain metrics with the same labelset {}", + render(&label_set(metric)) + ) + }); } *slot = value; } } + _ => merged.push((point, values)), } } + *rows = merged; } - if let Some((_, _, message)) = errors - .into_iter() - .min_by_key(|(point, rank, _)| (*point, *rank)) - { + if let Some((_, _, message)) = error.0 { return Err(DataFusionError::Execution(message)); } let mut builder = GridSeriesBuilder::try_new(schema.clone())?; let mut output = Vec::with_capacity(results.len()); - for (metric, rows) in results.into_values() { + for (metric, rows) in results { builder.begin_series(); for (point, row) in rows { builder.push_samples(point, &row)?; @@ -279,16 +309,49 @@ pub fn info( Ok(output) } -/// The samples of one result series by grid point, one per field. -type Rows = BTreeMap>>; +/// The samples of one result series at one grid point, one per field. +type Row = (usize, Vec>); + +/// What the info series kept at a point make of one base series. +struct Choice { + /// The info series, one per info metric that has one. + info: Vec, + /// The result series the point goes to, `None` when the point is dropped. + result: Option, + /// A label two of the info series set to different values. + conflict: Option>, +} + +/// The error Prometheus reports: the one of the first failing step, and within a step the first +/// of its checks (the info series, then each base series, then the result label sets). +#[derive(Default)] +struct FirstError(Option<(usize, u8, String)>); + +impl FirstError { + /// Records an error of check `rank` at `point`, formatting it only if it comes first. + fn record(&mut self, point: usize, rank: u8, message: impl FnOnce() -> String) { + if self + .0 + .as_ref() + .is_none_or(|(first, first_rank, _)| (point, rank) < (*first, *first_rank)) + { + self.0 = Some((point, rank, message())); + } + } +} + +/// A label set as Prometheus prints one. +fn render(labels: &LabelSet) -> String { + let labels: Vec = labels + .iter() + .map(|(name, value)| format!("{name}=\"{value}\"")) + .collect(); + format!("{{{}}}", labels.join(", ")) +} /// An info metric name and the identifying label values present on a series. type Signature = (Arc, Vec>>); -fn signature(name: Arc, metric: &MetricName) -> Signature { - (name, identifying_values(metric)) -} - /// The value of each identifying label, `None` where the series does not have it. fn identifying_values(metric: &MetricName) -> Vec>> { IDENTIFYING_LABELS diff --git a/src/promql/src/info/prometheus_cases.txt b/src/promql/src/info/prometheus_cases.txt index 073e82173ba..84b0cba30ad 100644 --- a/src/promql/src/info/prometheus_cases.txt +++ b/src/promql/src/info/prometheus_cases.txt @@ -1,7 +1,8 @@ # Generated by promql-compat cmd/narthex-golden-info built for amd64 (Prometheus v3.15.0 engine, # seed 1, lookback 25000ms). `case `, `base` the -# evaluated base selector, `info` the original sample timestamps of the info series Prometheus -# fetches, then the result series or the error. Points are `grid index:hex value bits`. +# evaluated base expression, `info` the original sample timestamps of the info series matching the +# name and data matchers (a superset of what Prometheus fetches: the identifying-label filter is +# the kernel's), then the result series or the error. Points are `grid index:hex value bits`. case - true - expr info({__name__="a"}) grid 100000 10000 2 @@ -4835,3 +4836,47 @@ info __name__=target_info,job=x,k8s_cluster=c1 ; 3:40ffbd0000000000 info __name__=target_info,job=x,k8s_cluster=c2 ; 3:40ffbd0000000000 result __name__=a,job=x ; 0:3ff0000000000000 1:3ff0000000000000 2:3ff0000000000000 end +case - true - +expr info({__name__="a"}) +grid 100000 10000 2 +base __name__=a,job=x ; 1:4000000000000000 +info __name__=target_info,job=x,k8s_cluster=a ; 0:40f86a0000000000 1:40f86a0000000000 +info __name__=target_info,job=x,k8s_cluster=b ; 0:40f86a0000000000 1:40f86a0000000000 +info __name__=target_info,job=x,k8s_cluster=c ; 1:40fadb0000000000 +error found duplicate series for info metric: existing {__name__="target_info", job="x", k8s_cluster="a"} @ 100000, new {__name__="target_info", job="x", k8s_cluster="b"} @ 100000 +end +case - true - +expr info({__name__="a"}) +grid 100000 10000 2 +base __name__=a,job=x ; 1:4000000000000000 +info __name__=target_info,job=x,k8s_cluster=c ; 1:40fadb0000000000 +info __name__=target_info,job=x,k8s_cluster=a ; 0:40f86a0000000000 1:40f86a0000000000 +info __name__=target_info,job=x,k8s_cluster=b ; 0:40f86a0000000000 1:40f86a0000000000 +result __name__=a,job=x,k8s_cluster=c ; 1:4000000000000000 +end +case - true __name__=~"target_info|build_info" +expr info({__name__="a"}, {__name__=~"target_info|build_info"}) +grid 100000 10000 4 +base __name__=a,job=x ; 1:4000000000000000 2:4000000000000000 3:4010000000000000 +base __name__=a,job=x,k8s_cluster=c1 ; 1:4000000000000000 2:4000000000000000 3:4000000000000000 +info __name__=target_info,job=x,k8s_cluster=c1 ; 1:40fadb0000000000 2:40fadb0000000000 3:40fadb0000000000 +info __name__=build_info,job=x,k8s_cluster=c2 ; 3:40ffbd0000000000 +error vector cannot contain metrics with the same labelset +end +case - true __name__!~"build_.*" +expr info(a * 2, {__name__!~"build_.*"}) +grid 100000 10000 3 +base instance=i1,job=x ; 0:4000000000000000 1:4010000000000000 2:4018000000000000 +base job=y ; 0:4000000000000000 1:4000000000000000 2:4018000000000000 +info __name__=target_info,instance=i1,job=x,k8s_cluster=c1 ; 0:40f86a0000000000 1:40f86a0000000000 2:40f86a0000000000 +result instance=i1,job=x,k8s_cluster=c1 ; 0:4000000000000000 1:4010000000000000 2:4018000000000000 +result job=y ; 0:4000000000000000 1:4000000000000000 2:4018000000000000 +end +case - true __name__=~"build_.*" +expr info(a * 2, {__name__=~"build_.*"}) +grid 100000 10000 3 +base job=y ; 0:4000000000000000 1:4000000000000000 2:4018000000000000 +info __name__=build_info,job=y,version=v1 ; 2:40fd4c0000000000 +result job=y ; 0:4000000000000000 1:4000000000000000 +result job=y,version=v1 ; 2:4018000000000000 +end diff --git a/src/promql/src/info/tests.rs b/src/promql/src/info/tests.rs index 3885ee0b438..0215fceaecb 100644 --- a/src/promql/src/info/tests.rs +++ b/src/promql/src/info/tests.rs @@ -15,10 +15,14 @@ use std::collections::BTreeMap; use std::sync::Arc; -use datatypes::series::samples::{FieldSpec, SeriesSchema}; +use datatypes::series::samples::{FieldSpec, SampleRef, Series, SeriesSchema}; use datatypes::series::time::{EvalGrid, GridSpec}; -use crate::grid_series::reference::{Labels, Points, labels_of, points_of, sparse_series}; +use crate::grid_series::reference::{ + Labels, OwnedHistogram, Points, labels_of, metric_of, points_of, sparse_series, +}; +use crate::grid_series::{GridSample, GridSeriesBuilder}; +use crate::histogram_eval::float_histogram::FloatHistogram; use crate::info::{InfoSelector, info}; use crate::select::MatchOp; @@ -140,3 +144,124 @@ fn info_matches_prometheus() { "{matched} matched, {failed} failed" ); } + +fn grid_schema(fields: usize) -> Arc { + Arc::new(SeriesSchema { + fields: (0..fields) + .map(|field| FieldSpec { + name: Arc::from(format!("f{field}")), + }) + .collect::>() + .into(), + eval_grid: Some(Arc::new( + EvalGrid::try_new(GridSpec::try_new(0, 10_000, 1).unwrap()).unwrap(), + )), + }) +} + +fn one_point( + schema: &Arc, + labels: &[(&str, &str)], + row: Vec>, +) -> Series { + let labels: Labels = labels + .iter() + .map(|(name, value)| (name.to_string(), value.to_string())) + .collect(); + let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap(); + builder.push_samples(0, &row).unwrap(); + builder.finish(metric_of(&labels)).unwrap().unwrap() +} + +fn target_info() -> Series { + one_point( + &grid_schema(1), + &[ + ("__name__", "target_info"), + ("job", "x"), + ("k8s_cluster", "c1"), + ], + vec![Some(GridSample::Float(0.0))], + ) +} + +fn all_labels() -> InfoSelector { + InfoSelector { + name_matchers: Vec::new(), + data_labels: Vec::new(), + keep_unmatched: true, + } +} + +/// Each field is a series of its own: two base series that end up with one label set merge where +/// they hold different fields, and collide only where both hold the same field. +#[test] +fn results_with_one_label_set_merge_by_field() { + let schema = grid_schema(2); + let float = |value| Some(GridSample::Float(value)); + let enriched = one_point( + &schema, + &[("__name__", "a"), ("job", "x")], + vec![float(1.0), None], + ); + let labelled = |row| { + one_point( + &schema, + &[("__name__", "a"), ("job", "x"), ("k8s_cluster", "c1")], + row, + ) + }; + let merged = info( + &schema, + &[enriched.clone(), labelled(vec![None, float(2.0)])], + &[target_info()], + &all_labels(), + ) + .unwrap(); + assert_eq!(merged.len(), 1); + let fields: Vec<_> = merged[0] + .fields + .iter() + .map(|field| match field.sample(0) { + SampleRef::Float(value) => Some(value), + _ => None, + }) + .collect(); + assert_eq!(fields, [Some(1.0), Some(2.0)]); + + let error = info( + &schema, + &[enriched, labelled(vec![float(3.0), None])], + &[target_info()], + &all_labels(), + ) + .unwrap_err(); + assert!(error.to_string().contains("same labelset"), "{error}"); +} + +/// A native histogram base sample is enriched like a float one. +#[test] +fn histogram_base_samples_are_enriched() { + let schema = grid_schema(1); + let histogram = + OwnedHistogram::parse("0/0/0/3ff0000000000000/3ff0000000000000/0/1_1/3ff0000000000000///"); + let base = one_point( + &schema, + &[("__name__", "a"), ("job", "x")], + vec![Some(GridSample::Histogram(Box::new( + FloatHistogram::try_from_view(&histogram.view()).unwrap(), + )))], + ); + let output = info(&schema, &[base], &[target_info()], &all_labels()).unwrap(); + assert_eq!(output.len(), 1); + assert_eq!( + labels_of(&output[0].metric) + .get("k8s_cluster") + .map(String::as_str), + Some("c1") + ); + assert!(matches!( + output[0].fields.as_slice()[0].sample(0), + SampleRef::Histogram(_) + )); +}