From 8c6ae77fa8323a41b5ace26d568cbb9617ecd7fa Mon Sep 17 00:00:00 2001 From: Dennis Zhuang Date: Sat, 17 Oct 2026 22:39:54 +0800 Subject: [PATCH] fix(promql): drop metadata labels from binary operator results Arithmetic and bool comparisons drop __type__ and __unit__ with __name__, as Prometheus does. __type__ stays part of the match group, so results of different groups can now share a label set; that fails as duplicate label sets after the operator, while two results within one group still fail the group_x uniqueness check. Signed-off-by: Dennis Zhuang --- src/promql/src/vector_match.rs | 157 +++++++++++++++++++-------- src/promql/src/vector_match/tests.rs | 61 ++++++++++- 2 files changed, 172 insertions(+), 46 deletions(-) diff --git a/src/promql/src/vector_match.rs b/src/promql/src/vector_match.rs index 14a0a818d79..a01ab145014 100644 --- a/src/promql/src/vector_match.rs +++ b/src/promql/src/vector_match.rs @@ -175,6 +175,11 @@ pub fn binary_vectors( let fields = output.fields.len(); let drop_name = !matching.op.is_comparison() || matching.return_bool; let mut results = Results::default(); + let group_ids: HashMap<&GroupKey, usize> = one_groups + .keys() + .enumerate() + .map(|(id, key)| (key, id)) + .collect(); let mut matched_groups: HashMap> = HashMap::new(); let mut values = vec![None; fields]; for (many_index, many_series) in many.series.iter().enumerate() { @@ -182,6 +187,7 @@ pub fn binary_vectors( let Some(candidates) = one_groups.get(&key) else { continue; }; + let group_id = group_ids[&key]; if matches!(matching.cardinality, Cardinality::OneToOne) { matched_groups.entry(key).or_default().push(many_index); } @@ -224,7 +230,7 @@ pub fn binary_vectors( _ => None, }; } - results.insert(&metric, point, &values); + results.insert(&metric, group_id, point, &values); } } } @@ -249,36 +255,60 @@ pub fn binary_vectors( implicit_many_to_one = Some((point, key)); } } - let grouping_duplicate = results.sort_rows(); - // Prometheus fails at the first failing step, and checks the one side before matching. - let error = match (one_duplicate, implicit_many_to_one, grouping_duplicate) { - (Some((point, key)), many_to_one, grouping) - if many_to_one.is_none_or(|(first, _)| point <= first) - && grouping.as_ref().is_none_or(|(first, _)| point <= *first) => - { - let side = if one_is_left { "left" } else { "right" }; - Some(format!( + // Prometheus fails at the first failing step. Within a step it checks the one side, then + // matches, and only after the operator checks that result label sets are unique. + let mut errors: Vec<(usize, u8, String)> = Vec::new(); + if let Some((point, key)) = one_duplicate { + let side = if one_is_left { "left" } else { "right" }; + errors.push(( + point, + 0, + format!( "found duplicate series for the match group {} on the {side} hand-side of the \ operation at {} ms; many-to-many matching not allowed: matching labels must be \ unique on one side", render(key), grid.timestamps()[point] - )) - } - (_, Some((point, key)), _) => Some(format!( - "multiple matches for labels {} at {} ms: many-to-one matching must be explicit \ - (group_left/group_right)", - render(key), - grid.timestamps()[point] - )), - (_, _, Some((point, key))) => Some(format!( - "multiple matches for labels {} at {} ms: grouping labels must ensure unique matches", - render(&key), - grid.timestamps()[point] - )), - _ => None, - }; - if let Some(error) = error { + ), + )); + } + if let Some((point, key)) = implicit_many_to_one { + errors.push(( + point, + 1, + format!( + "multiple matches for labels {} at {} ms: many-to-one matching must be explicit \ + (group_left/group_right)", + render(key), + grid.timestamps()[point] + ), + )); + } + if let Some(duplicate) = results.sort_rows() { + let grouping = duplicate.within_group && matching.cardinality != Cardinality::OneToOne; + errors.push(if grouping { + ( + duplicate.point, + 1, + format!( + "multiple matches for labels {} at {} ms: grouping labels must ensure \ + unique matches", + render(&duplicate.key), + grid.timestamps()[duplicate.point] + ), + ) + } else { + ( + duplicate.point, + 2, + same_labelset(&duplicate.key, grid.timestamps()[duplicate.point]), + ) + }); + } + if let Some((_, _, error)) = errors + .into_iter() + .min_by_key(|(point, rank, _)| (*point, *rank)) + { return Err(DataFusionError::Execution(error)); } results.build(output) @@ -319,7 +349,7 @@ pub fn set_vectors( for point in lhs.range(index) { if lhs.present(index, point)? { lhs.row(index, point, &mut values)?; - results.insert(&lhs.series[index].metric, point, &values); + results.insert(&lhs.series[index].metric, 0, point, &values); } } } @@ -339,15 +369,14 @@ pub fn set_vectors( } if matched == keep_matched { kept.row(index, point, &mut values)?; - results.insert(&series.metric, point, &values); + results.insert(&series.metric, 0, point, &values); } } } - if let Some((point, key)) = results.sort_rows() { - return Err(DataFusionError::Execution(format!( - "vector cannot contain metrics with the same labelset {} at {} ms", - render(&key), - grid.timestamps()[point] + if let Some(duplicate) = results.sort_rows() { + return Err(DataFusionError::Execution(same_labelset( + &duplicate.key, + grid.timestamps()[duplicate.point], ))); } results.build(output) @@ -444,6 +473,9 @@ fn overlap( Ok(first) } +/// Prometheus `schema.IsMetadataLabel`: the labels removed with the metric name. +const METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"]; + fn result_metric( many: &MetricName, one: &MetricName, @@ -453,7 +485,7 @@ fn result_metric( ) -> Result { let mut metric = many.clone(); if drop_name { - metric.remove("__name__"); + metric.remove_all(&METADATA_LABELS); } if matching.cardinality == Cardinality::OneToOne { match &matching.labels { @@ -586,37 +618,58 @@ struct Results { series: HashMap, } -/// Rows in arrival order: their points, and their field values laid out row after row. +/// Rows in arrival order: their points, the match group each came from, and their field +/// values laid out row after row. struct ResultRows { metric: MetricName, points: Vec, + groups: Vec, values: Vec>, } +/// Two rows of one result label set at one point. +struct Duplicate { + point: usize, + key: GroupKey, + /// Whether two of the rows came from the same match group. + within_group: bool, +} + +fn same_labelset(key: &GroupKey, timestamp: i64) -> String { + format!( + "vector cannot contain metrics with the same labelset {} at {timestamp} ms", + render(key) + ) +} + impl Results { - fn insert(&mut self, metric: &MetricName, point: usize, values: &[Option]) { + fn insert(&mut self, metric: &MetricName, group: usize, point: usize, values: &[Option]) { let rows = match self.series.entry(key_of(metric)) { Entry::Occupied(entry) => entry.into_mut(), Entry::Vacant(entry) => entry.insert(ResultRows { metric: metric.clone(), points: Vec::new(), + groups: Vec::new(), values: Vec::new(), }), }; rows.points.push(point); + rows.groups.push(group); rows.values.extend_from_slice(values); } /// Sorts the rows of each label set by point and returns the earliest point where a label - /// set has two rows. Prometheus evaluates step by step and fails at the first such step. - fn sort_rows(&mut self) -> Option<(usize, GroupKey)> { - let mut first: Option<(usize, &GroupKey)> = None; + /// set has two rows, preferring one within a match group at that point. Prometheus + /// evaluates step by step and fails at the first such step. + fn sort_rows(&mut self) -> Option { + let mut first: Option<(usize, bool, &GroupKey)> = None; for (key, rows) in &mut self.series { if !rows.points.is_sorted() { let fields = rows.values.len() / rows.points.len(); let mut order: Vec = (0..rows.points.len()).collect(); order.sort_by_key(|&row| rows.points[row]); rows.points = order.iter().map(|&row| rows.points[row]).collect(); + rows.groups = order.iter().map(|&row| rows.groups[row]).collect(); rows.values = order .iter() .flat_map(|&row| { @@ -626,13 +679,29 @@ impl Results { }) .collect(); } - if let Some(pair) = rows.points.windows(2).find(|pair| pair[0] == pair[1]) - && first.is_none_or(|(point, _)| pair[0] < point) - { - first = Some((pair[0], key)); + let mut start = 0; + while start < rows.points.len() { + let point = rows.points[start]; + let end = start + rows.points[start..].partition_point(|&p| p == point); + if end - start > 1 { + let mut groups = rows.groups[start..end].to_vec(); + groups.sort_unstable(); + let within_group = groups.windows(2).any(|pair| pair[0] == pair[1]); + if first.is_none_or(|(first_point, first_within, _)| { + (point, !within_group) < (first_point, !first_within) + }) { + first = Some((point, within_group, key)); + } + break; + } + start = end; } } - first.map(|(point, key)| (point, key.clone())) + first.map(|(point, within_group, key)| Duplicate { + point, + key: key.clone(), + within_group, + }) } /// Builds the result series from rows sorted by [`Results::sort_rows`] without duplicates. diff --git a/src/promql/src/vector_match/tests.rs b/src/promql/src/vector_match/tests.rs index cf9ace2d919..0170783a3b4 100644 --- a/src/promql/src/vector_match/tests.rs +++ b/src/promql/src/vector_match/tests.rs @@ -109,6 +109,7 @@ fn random_side(rng: &mut StdRng, grid: &Arc, fields: usize) -> Side { ("host", ["a", "b", ""]), ("job", ["x", "y", "y"]), ("zone", ["z1", "", ""]), + ("__type__", ["counter", "", ""]), ] { let value = values[rng.random_range(0..values.len())]; if !value.is_empty() { @@ -171,7 +172,9 @@ fn signature(labels: &Labels, matching: &MatchingLabels) -> Labels { fn result_labels(many: &Labels, one: &Labels, matching: &BinaryMatching) -> Labels { let mut labels = many.clone(); if matching.return_bool || !matching.op.is_comparison() { - labels.remove("__name__"); + for name in ["__name__", "__type__", "__unit__"] { + labels.remove(name); + } } let include = match &matching.cardinality { Cardinality::OneToOne => { @@ -303,7 +306,11 @@ fn reference_binary( let dense = output .entry(result) .or_insert_with(|| vec![vec![None; fields]; points]); - assert!(!present(&dense[point]), "result label sets collide"); + // Prometheus rejects the step once the operator is done. + if present(&dense[point]) { + violations.insert(Violation::DuplicateLabelSet); + continue; + } dense[point] = row; } } @@ -755,3 +762,53 @@ fn pow_and_atan2_match_go() { } assert_eq!(checked, 452); } + +/// Results that share a label set once `__type__` is dropped fail as a grouping violation within +/// one match group, and as duplicate label sets across match groups, which Prometheus detects +/// only after the operator. +#[test] +fn colliding_results_fail_by_where_they_collide() { + let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 10_000, 1).unwrap()).unwrap()); + let schema = schema(1, &grid); + let mut builder = GridSeriesBuilder::try_new(Arc::new(schema.clone())).unwrap(); + let mut series = |labels: &[(&str, &str)]| { + builder.push(0, &[Some(1.0)]).unwrap(); + let labels = labels + .iter() + .map(|(name, value)| (name.to_string(), value.to_string())) + .collect(); + builder.finish(metric(&labels)).unwrap().unwrap() + }; + let lhs = vec![ + series(&[("__name__", "a"), ("__type__", "counter"), ("job", "x")]), + series(&[("__name__", "a"), ("__type__", "gauge"), ("job", "x")]), + ]; + let rhs = [ + series(&[("__name__", "b"), ("__type__", "counter"), ("job", "x")]), + series(&[("__name__", "b"), ("__type__", "gauge"), ("job", "x")]), + ]; + for (labels, expected) in [ + // `__type__` is part of the match group. + (MatchingLabels::Ignoring(Vec::new()), "same labelset"), + // Both left series fall into the group of each right series. + ( + MatchingLabels::On(vec![Arc::from("job")]), + "grouping labels must ensure unique matches", + ), + ] { + let rhs = match &labels { + MatchingLabels::On(_) => &rhs[..1], + MatchingLabels::Ignoring(_) => &rhs[..], + }; + let matching = BinaryMatching { + op: BinaryOp::Add, + return_bool: false, + labels, + cardinality: Cardinality::ManyToOne(Vec::new()), + }; + let error = binary_vectors(&matching, &schema, &lhs, &schema, rhs) + .unwrap_err() + .to_string(); + assert!(error.contains(expected), "{error}"); + } +}