From d0f5f97d222a5b7d5fee61419b6541b8edc70f36 Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 30 Sep 2026 09:30:00 +0000 Subject: [PATCH] fix: reject duplicate label sets after dropping the metric name PromQL arithmetic with a literal drops `__name__`; if two series then share a label set, Prometheus errors ("vector cannot contain metrics with the same labelset"). The per-series literal path returned both rows, and the Fallback literal path kept `__name__` altogether. Add `Operator::series_without_name`, a `SeriesLabels` rewrite that errors on equal resulting label sets, and use it for both literal paths. Matching and `without` relabeling still tolerate repeats: `series_binary` already rejects right-hand and matched left-hand duplicates, and aggregation merges them. Co-Authored-By: Claude Opus 5.5 --- .../src/operators/mod.rs | 1 + .../src/operators/series_labels.rs | 42 ++++++++++++++++++- .../src/operators/unchecked.rs | 5 ++- .../src/physical_planner/promql_fallback.rs | 1 + .../src/physical_planner/row_values.rs | 2 +- .../tests/deployment_computation.rs | 31 ++++++++++++++ .../tests/promql_fallback.rs | 29 ++++++++++++- .../develop_docs/physical-compile-coverage.md | 2 +- 8 files changed, 107 insertions(+), 6 deletions(-) diff --git a/crates/asap-physical-operators/src/operators/mod.rs b/crates/asap-physical-operators/src/operators/mod.rs index 196e9e8a6..8c189fd4a 100644 --- a/crates/asap-physical-operators/src/operators/mod.rs +++ b/crates/asap-physical-operators/src/operators/mod.rs @@ -81,6 +81,7 @@ enum Kind { SeriesLabels { kind: planner_types::pre_asap::VectorMatchKind, labels: Vec, + unique: bool, }, SeriesBinary { operator: planner_types::post_asap::BinaryOperator, diff --git a/crates/asap-physical-operators/src/operators/series_labels.rs b/crates/asap-physical-operators/src/operators/series_labels.rs index be2483196..5032f4359 100644 --- a/crates/asap-physical-operators/src/operators/series_labels.rs +++ b/crates/asap-physical-operators/src/operators/series_labels.rs @@ -87,7 +87,27 @@ impl Operator { ) -> Result { layout(&input)?; Ok(Self { - kind: Kind::SeriesLabels { kind, labels }, + kind: Kind::SeriesLabels { + kind, + labels, + unique: false, + }, + inputs: vec![input.clone()], + output: input, + }) + } + + /// Drop the metric name from each row's label set, as PromQL arithmetic + /// with a literal does. Unlike matching labels, the result is itself a + /// vector, so two rows that become equal are an error, as in Prometheus. + pub fn series_without_name(input: Schema) -> Result { + layout(&input)?; + Ok(Self { + kind: Kind::SeriesLabels { + kind: VectorMatchKind::Ignoring, + labels: vec![], + unique: true, + }, inputs: vec![input.clone()], output: input, }) @@ -134,7 +154,15 @@ pub(super) fn execute<'a>( let (rows, _memory) = collect_rows(left, &context).await?; let mut result = Vec::new(); match (&operator.kind, right) { - (Kind::SeriesLabels { kind, labels }, None) => { + ( + Kind::SeriesLabels { + kind, + labels, + unique, + }, + None, + ) => { + let mut seen = std::collections::BTreeSet::new(); for mut row in rows { work.checkpoint().await?; let mut set = left_layout.read(&output, &row)?; @@ -146,6 +174,16 @@ pub(super) fn execute<'a>( } left_layout.write(&output, &mut row, &set)?; workspace.grow(row_bytes(&row))?; + // Matching and grouping sides may repeat a label set; + // their consumers decide whether that is an error. + if *unique { + workspace.grow(set.iter().map(|(k, v)| 64 + k.len() + v.len()).sum())?; + if !seen.insert(set) { + return Err(invalid( + "vector cannot contain metrics with the same labelset", + )); + } + } result.push(row); } } diff --git a/crates/asap-physical-operators/src/operators/unchecked.rs b/crates/asap-physical-operators/src/operators/unchecked.rs index 46bc6a0ca..fdafb630a 100644 --- a/crates/asap-physical-operators/src/operators/unchecked.rs +++ b/crates/asap-physical-operators/src/operators/unchecked.rs @@ -65,7 +65,10 @@ impl TryFrom for Operator { at_ms, steps, )?, - Kind::SeriesLabels { kind, labels } => { + // The rebuilt kind must equal the serialized one, which rejects + // a unique rewrite with other matching labels. + Kind::SeriesLabels { unique: true, .. } => Operator::series_without_name(input(0)?)?, + Kind::SeriesLabels { kind, labels, .. } => { Operator::series_labels(input(0)?, kind, labels)? } Kind::SeriesBinary { operator } => { diff --git a/crates/asap-physical-operators/src/physical_planner/promql_fallback.rs b/crates/asap-physical-operators/src/physical_planner/promql_fallback.rs index 12c60ad1a..b0234acc5 100644 --- a/crates/asap-physical-operators/src/physical_planner/promql_fallback.rs +++ b/crates/asap-physical-operators/src/physical_planner/promql_fallback.rs @@ -222,6 +222,7 @@ impl Lowering { }; let step = self.value(vector)?; let input = self.schema(&step); + let step = self.add(Operator::series_without_name(input.clone())?, vec![step]); let value = named_column(&input, &ColumnRef::SampleValue)?; let literal = Expression::Literal { value: crate::values::Value::Float64(literal), diff --git a/crates/asap-physical-operators/src/physical_planner/row_values.rs b/crates/asap-physical-operators/src/physical_planner/row_values.rs index 615539271..7e2c7c5f0 100644 --- a/crates/asap-physical-operators/src/physical_planner/row_values.rs +++ b/crates/asap-physical-operators/src/physical_planner/row_values.rs @@ -86,7 +86,7 @@ pub(super) fn series_scalar_binary( literal_left: bool, ) -> Result, Error> { arithmetic(operator)?; - let relabel = Operator::series_labels(input.clone(), VectorMatchKind::Ignoring, vec![])?; + let relabel = Operator::series_without_name(input.clone())?; let value = input .fields .iter() diff --git a/crates/asap-physical-operators/tests/deployment_computation.rs b/crates/asap-physical-operators/tests/deployment_computation.rs index d11de58e7..532cd331f 100644 --- a/crates/asap-physical-operators/tests/deployment_computation.rs +++ b/crates/asap-physical-operators/tests/deployment_computation.rs @@ -102,6 +102,17 @@ fn execute( dag: &PostAsapDag, samples: &[Sample], end: i64, +) -> Result>, String> { + execute_relabeled(dag, samples, end, &BTreeMap::new()) +} + +/// [`execute`], supplying samples of each instance in `relabel` under its +/// `(__name__, instance)` instead. +fn execute_relabeled( + dag: &PostAsapDag, + samples: &[Sample], + end: i64, + relabel: &BTreeMap<&str, (&str, &str)>, ) -> Result>, String> { let inputs = raw_inputs(dag); let program = compile( @@ -122,6 +133,8 @@ fn execute( .iter() .filter(|sample| sample.0 == metric) .map(|(name, job, instance, at, value)| { + let (name, instance) = + relabel.get(instance).copied().unwrap_or((*name, *instance)); let labels = BTreeMap::from([ ("__name__".to_string(), name.to_string()), ("job".into(), job.to_string()), @@ -469,6 +482,24 @@ fn per_series_scalar_arithmetic_applies_to_stored_readouts() { } } +// A literal operand drops the metric name; series that then share a label set +// are an error, as in Prometheus, rather than duplicate output series. +#[test] +fn per_series_scalar_arithmetic_rejects_label_sets_equal_without_the_name() { + let dag = exact_dag("sum_over_time(m[5m]) * 2"); + let samples = counter("m", "api", 10., 10.) + .chain(counter("m", "api", 1., 1.).map(|s| (s.0, s.1, "y", s.3, s.4))) + .collect::>(); + let distinct = BTreeMap::from([("y", ("n", "y"))]); + assert!(execute_relabeled(&dag, &samples, 300_000, &distinct).is_ok()); + // Both series are then {instance="x",job="api"}, named m and n. + let equal = BTreeMap::from([("y", ("n", "x"))]); + let Err(error) = execute_relabeled(&dag, &samples, 300_000, &equal) else { + panic!("duplicate label sets were accepted"); + }; + assert!(error.contains("same labelset"), "{error}"); +} + fn with_vector_match( mut dag: PostAsapDag, kind: planner_types::pre_asap::VectorMatchKind, diff --git a/crates/asap-physical-operators/tests/promql_fallback.rs b/crates/asap-physical-operators/tests/promql_fallback.rs index a9decb52a..c609d28f0 100644 --- a/crates/asap-physical-operators/tests/promql_fallback.rs +++ b/crates/asap-physical-operators/tests/promql_fallback.rs @@ -128,7 +128,8 @@ fn evaluate( .flat_map(|(_, samples)| samples.iter()) .map(|(spec, seconds, value)| { let mut labels = labels(spec); - labels.insert("__name__".into(), name.clone()); + // A sample may supply its own `__name__`, as a series of another metric. + labels.entry("__name__".into()).or_insert(name.clone()); promql_rows::series_row(&schema, &labels, seconds * 1000, *value).unwrap() }) .collect(); @@ -604,6 +605,32 @@ fn without_grouping_drops_labels_and_the_name() { vec![(String::new(), 4.)] ); assert!(labeled("sum without (inst) (a)", &[], 60).is_empty()); + // Series equal without the name share a group rather than colliding. + let named: &[Sample] = &[ + ("job=x,inst=1", 50, 1.), + ("__name__=b,job=x,inst=1", 50, 2.), + ]; + assert_eq!( + labeled("sum without (inst) (a)", &[("a", named)], 60), + vec![("job=x".into(), 3.)] + ); +} + +// Arithmetic with a literal drops the metric name; series that then share a +// label set are an error, as in Prometheus. +#[test] +fn literal_arithmetic_drops_the_name_and_rejects_equal_label_sets() { + let a: &[Sample] = &[ + ("job=x,inst=1", 50, 1.), + ("__name__=b,job=x,inst=2", 50, 2.), + ]; + assert_eq!( + labeled("a * 2", &[("a", a)], 60), + vec![("inst=1,job=x".into(), 2.), ("inst=2,job=x".into(), 4.)] + ); + let equal: &[Sample] = &[("job=x", 50, 1.), ("__name__=b,job=x", 50, 2.)]; + let error = evaluate("a * 2", &[("a", equal)], 60).unwrap_err(); + assert!(error.contains("same labelset"), "{error}"); } // An empty label value is an absent label, and an empty side yields an empty diff --git a/docs/develop_docs/physical-compile-coverage.md b/docs/develop_docs/physical-compile-coverage.md index f992412b5..9a090ee2b 100644 --- a/docs/develop_docs/physical-compile-coverage.md +++ b/docs/develop_docs/physical-compile-coverage.md @@ -125,7 +125,7 @@ Totals are unchanged: 19 Supported, 5 Partial, 5 Missing, 2 Backend. | Row | Change | |---|---| | 5 | Query-time `Binary` over rows with a series identity, such as per-series readouts of stored state, uses the Fallback's `series_labels` and `series_binary`. Examples: `avg_over_time` as stored sum/count, and `rate(a) / rate(b)`. Matching drops `__name__` and honors `on`/`ignoring` when the payload carries them. Only one-to-one arithmetic is covered; `group_left`/`group_right` stay rejected and comparisons are row 7. Now Supported. | -| 4, 8 | A literal operand also applies to per-series rows and drops `__name__`. | +| 4, 8 | A literal operand also applies to per-series rows and drops `__name__`, in the Fallback too. Series whose label sets become equal are an error, as in Prometheus. | Grouped `sum`/`avg`, current-series `Sum`/`Average` readouts, and `sum_over_time`/`avg_over_time` use Prometheus' Kahan-Neumaier summation. An