From e49bb094806cd142c444ffa53718affcb68e6260 Mon Sep 17 00:00:00 2001 From: discord9 Date: Wed, 11 Nov 2026 11:40:40 +0800 Subject: [PATCH] feat(promql): support metric-name rewrites in native labels (#10176) Signed-off-by: discord9 --- src/promql/src/map_label.rs | 14 + src/promql/src/native/plan.rs | 446 +++++++++++++-------- src/promql/src/native/plan/tests.rs | 237 ++++++++++- src/query/src/promql/planner/native.rs | 198 ++++++++- src/query/src/promql/planner/test.rs | 96 +++++ tests-integration/src/tests/promql_test.rs | 268 ++++++++++++- 6 files changed, 1058 insertions(+), 201 deletions(-) diff --git a/src/promql/src/map_label.rs b/src/promql/src/map_label.rs index f65265592d0..9a3f0e4db98 100644 --- a/src/promql/src/map_label.rs +++ b/src/promql/src/map_label.rs @@ -255,6 +255,20 @@ impl LabelRewrite { } } + /// Whether this change can write a nonempty value. + /// + /// An empty value removes the destination instead ([`set_label`]), so a change whose value is + /// necessarily empty only deletes a label and publishes none that its input did not carry. + /// `label_replace`'s template is judged as written: `$1` can expand to nothing for one series and + /// to a value for another, so it may write. A `label_join` with no source labels joins nothing + /// and is always empty. + pub fn writes_value(&self) -> bool { + match &self.0 { + Change::Replace { replacement, .. } => !replacement.is_empty(), + Change::Join { src, .. } => !src.is_empty(), + } + } + /// `label_replace`'s change, failing on an invalid expression or an empty destination. pub fn replace(dst: &str, replacement: &str, src: &str, regex: &str) -> Result { let compiled = Regex::new(&format!("^(?s:{regex})$")).map_err(|_| { diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs index 754f066bffe..b2cdb9f91b5 100644 --- a/src/promql/src/native/plan.rs +++ b/src/promql/src/native/plan.rs @@ -55,11 +55,24 @@ use crate::vector_match::{ /// its result does not report. const LE_LABEL: &str = "le"; +/// The label a metric name is stored under, which a scan binds beside its tag columns rather than +/// among them. +const METRIC_NAME: &str = "__name__"; + /// Metadata labels the runtime removes from histogram result label sets. const PROMQL_METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"]; -/// The label columns `node` reports, from the tags of the scan each input reads and the labels the -/// nodes above it write. +/// The complete label set `node` reports at run time, from the labels each input reports and the +/// ones the nodes above it write. +/// +/// This is the identity a [`NativeExpr::Merge`] groups by and a vector match reads, so the metric +/// name is part of it whenever the series carries one. The caller's `source_tags` are the complete +/// label sets of the sources, indexed by the `source_index` of the leaf that reads each: a scan +/// bound to a single metric names every series it hands over even though the name is not one of its +/// tag columns, so the canonical pass includes `__name__` in that source's entry. The query planner +/// derives the *output columns* with its own walk, which reads the same algebra from the tag columns +/// each source declares; the two have to agree on where a kernel drops the name, which is why this +/// walk drops it exactly where the runtime does. /// /// They follow Prometheus' result labels: a one-to-one match keeps the left side's labels, narrowed /// by `on` or `ignoring`; `and` and `unless` keep the left side's series and `or` adds the right @@ -67,41 +80,76 @@ const PROMQL_METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"]; /// of each value keeps its operand's; a scalar and `vector(s)` have none; `count_values` keeps its /// grouping and the label it writes; a fill that answers a missing many-side sample files the one /// side's matching labels, so they join the set, while a grouped `include` label joins them only when -/// the one side carries it or the matching keeps it in the key a fill reads. `None` for a node whose -/// labels this walk does not derive - a label rewrite of the metric name, an `absent` naming a label -/// no input column carries, and a `count_values` whose destination is the metric name or no label at -/// all, none of which is a column here - so a tree holding one reports no label set here. `None` is -/// not the query's answer: the caller asks the existing planner for the columns the node is named -/// after instead of declining the query. +/// the one side carries it or the matching keeps it in the key a fill reads. A label rewrite writes +/// its destination, and a destination that names the metric name writes that name. +/// +/// A node that reports a value of its own drops the name where its kernel does - `timestamp`, a +/// range function whose result is computed (`rate`, `increase`, ...), a math or histogram map, +/// `histogram_quantile`, every non-filtering binary - so a name rewrite below one of them cannot +/// make the name reappear above it. +/// +/// `None` for a node whose labels this walk does not derive - a `count_values` whose destination is +/// empty. `None` proves nothing, so the caller treats it as "may alias" rather than as "distinct". pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Option> { match &node.expr { NativeExpr::Input(input) => source_tags.get(input.source_index).cloned(), NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } | NativeExpr::VectorOf { .. } => { Some(Vec::new()) } - NativeExpr::Instant { input, .. } - | NativeExpr::Window { input, .. } - | NativeExpr::Timestamp { input } => fused_output_tags(input, source_tags), + // These report a value of their own rather than the sample they read, and the runtime drops + // the name for each of them. `source_tags` is the authority on what an input carries, so a + // discovery input arrives carrying `__name__` and these arms are what decides it does not + // survive. + NativeExpr::Instant { input, op } => { + let tags = fused_output_tags(input, source_tags)?; + Some(match op.value { + InstantValue::Timestamp => without_metric_name(tags), + InstantValue::Sample | InstantValue::InfoTimestamp => tags, + }) + } + NativeExpr::Window { input, op } => { + let tags = fused_output_tags(input, source_tags)?; + Some(if op.function.drops_metric_name() { + without_metric_name(tags) + } else { + tags + }) + } + NativeExpr::Timestamp { input } => { + Some(without_metric_name(fused_output_tags(input, source_tags)?)) + } // A label rewrite changes the label set itself: `label_replace` adds its destination label - // and an empty replacement removes it, so the input's labels are not the output's. A map of - // math alone keeps them. + // and an empty replacement removes it, so the input's labels are not the output's. A math or + // histogram kernel drops the metric name where it runs, and the ops apply in written order, + // so what follows a rewrite sees the set the rewrite left. NativeExpr::Map { input, op } => { let mut tags = fused_output_tags(input, source_tags)?; for operation in op.ops.iter() { - if let MapOp::Label(rewrite) = operation { - let destination = rewrite.destination(); - if destination == "__name__" { - return None; + match operation { + MapOp::Label(rewrite) => { + let destination = rewrite.destination(); + if tags.iter().any(|tag| tag == destination) { + // The input carries the label, so the rewrite changes its value - and an + // empty replacement may remove it, which the column still represents. + tags.retain(|tag| tag != destination); + tags.push(destination.to_string()); + } else if rewrite.writes_value() { + // A fresh destination is a label the output only carries when the + // rewrite can write a value; a delete-only rewrite of a label no input + // carried publishes nothing. + tags.push(destination.to_string()); + } } - tags.retain(|tag| tag != destination); - tags.push(destination.to_string()); + MapOp::Math(_) | MapOp::Histogram(_) => tags = without_metric_name(tags), } } Some(tags) } - // Its parameters are scalars, which carry no labels, so what the node reports are its - // input's labels. - NativeExpr::ParameterizedMap { input, .. } => fused_output_tags(input, source_tags), + // Its parameters are scalars, which carry no labels; the map computes a value of its own + // with the input, and its kernel drops the metric name. + NativeExpr::ParameterizedMap { input, .. } => { + Some(without_metric_name(fused_output_tags(input, source_tags)?)) + } // `scalar()` reports one value and no labels, like the scalar it produces. NativeExpr::ScalarOf { .. } => Some(Vec::new()), // Its parameter is a scalar, which carries no labels, and ranking keeps each selected @@ -117,13 +165,9 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Opti } // An aggregate reports the grouping labels its input carries: `by(...)` keeps exactly the // ones it names, and `without(...)` keeps the rest and drops the metric name, as Prometheus - // does. - // - // What this derives is the set of *output columns*, which is what the node's schema needs. - // It is not the complete runtime label set: a chunk's labels also carry the metric name, - // which is bound separately from these tag columns rather than being one of them. A - // grouping by `__name__` keeps that name in the labels while contributing no column here, - // and `sum by (__name__)` over one metric is still one column set. + // does. A grouping by `__name__` keeps the name where the input reports one; a named scan + // only reports it among the runtime labels, so the aggregate keeps it there and publishes no + // column for it, which is what `sum by (__name__)` over one metric reports. NativeExpr::Aggregate { input, op } => { let input_tags = fused_output_tags(input, source_tags)?; Some(match &op.group { @@ -134,7 +178,7 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Opti Grouping::Without(dropped) => input_tags .into_iter() .filter(|tag| { - tag != "__name__" + tag != METRIC_NAME && !dropped.iter().any(|name| name.as_ref() == tag.as_str()) }) .collect(), @@ -148,63 +192,51 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Opti | BoundVectorOp::LimitK { .. } | BoundVectorOp::LimitRatio { .. } | BoundVectorOp::Sort(_) => fused_output_tags(input, source_tags), - BoundVectorOp::Absent { labels } => { - let input_tags = fused_output_tags(input, source_tags)?; - let named = labels + // `absent` reports the labels its own matchers name, not the input's set and not the + // table's columns: it answers one series carrying exactly those, so a label the input + // does not carry is still a label of the result. + BoundVectorOp::Absent { labels } => Some( + labels .tags() .iter() - .map(|label| label.name.as_ref()) - .collect::>(); - // `absent` reports matcher-named labels, not all input labels or the table's columns; - // only named labels present as input columns can be represented as frame columns. - if named - .iter() - .any(|name| !input_tags.iter().any(|tag| tag == name)) - { - return None; - } - Some( - input_tags - .into_iter() - .filter(|tag| named.contains(&tag.as_str())) - .collect(), - ) - } + .map(|label| label.name.to_string()) + .collect(), + ), // `count_values` writes the value it counts into its own label and then keeps its // grouping: `by(...)` keeps the labels named plus that destination, `without(...)` // keeps the rest minus the labels named and the metric name. What the node reports is // therefore the grouping applied to the labels its input carries, with the destination // among them - not the input's set, and not the scan's, since the grouping narrows both. // - // The destination is a column the node creates, so a rewrite of it above this node is a - // rewrite of an existing label. A destination naming the metric name, or no label at - // all, writes something that is not a column here, which leaves the set underivable for - // the same reason a map's `__name__` destination does. This walk derives the labels; it - // does not decide which groupings the kernel accepts, and those refusals stay where they - // are. - BoundVectorOp::CountValues { label, group } - if !label.is_empty() && label.as_ref() != "__name__" => - { + // The destination is a label the node creates, so a rewrite of it above this node is a + // rewrite of an existing label. A destination naming the metric name writes the name + // the node groups by, which is the one label its result carries. This walk derives the + // labels; it does not decide which groupings the kernel accepts, and those refusals stay + // where they are. + BoundVectorOp::CountValues { label, group } if !label.is_empty() => { let mut tags = fused_output_tags(input, source_tags)?; - let keeps_destination = match group { - Grouping::By(kept) => { - tags.retain(|tag| kept.iter().any(|name| name.as_ref() == tag.as_str())); - // `by` keeps the label the node writes whether the input carried it or not. - true - } - Grouping::Without(dropped) => { - tags.retain(|tag| { - tag != "__name__" - && !dropped.iter().any(|name| name.as_ref() == tag.as_str()) - }); - // `without` removes it only when the grouping names it. - !dropped.iter().any(|name| name.as_ref() == label.as_ref()) - } - }; - if keeps_destination && !tags.iter().any(|tag| tag.as_str() == label.as_ref()) { + if !tags.contains(&label.to_string()) { tags.push(label.to_string()); } - Some(tags) + Some(match group { + Grouping::By(names) => { + let mut output = tags + .into_iter() + .filter(|tag| names.iter().any(|name| name.as_ref() == tag)) + .collect::>(); + // `by` keeps the label the node writes whether the input carried it or not. + if !output.contains(&label.to_string()) { + output.push(label.to_string()); + } + output + } + Grouping::Without(names) => tags + .into_iter() + .filter(|tag| { + tag != METRIC_NAME && !names.iter().any(|name| name.as_ref() == tag) + }) + .collect(), + }) } BoundVectorOp::CountValues { .. } => None, // A classic histogram function folds the bucket series that share the other labels into @@ -223,105 +255,133 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Opti left, right, matching, - } => match (&left.output, &right.output) { - (NativeType::Scalar(_), _) => fused_output_tags(right, source_tags), - (_, NativeType::Scalar(_)) => fused_output_tags(left, source_tags), - _ => { - let kept = |tags: Vec| -> Vec { - match &matching.labels { - MatchingLabels::On(names) => tags - .into_iter() - .filter(|tag| names.iter().any(|name| name.as_ref() == tag.as_str())) - .collect(), - MatchingLabels::Ignoring(names) => tags - .into_iter() - .filter(|tag| !names.iter().any(|name| name.as_ref() == tag.as_str())) - .collect(), - } - }; - let left_tags = fused_output_tags(left, source_tags)?; - let right_tags = fused_output_tags(right, source_tags)?; - // One-to-one matching narrows the left side's labels with `on`/`ignoring`. - // A group retains the many side's labels and copies the named labels from the one - // side: `group_left(container)` keeps the left labels and adds `container` from - // the right. - let mut tags = match &matching.cardinality { - Cardinality::OneToOne => kept(left_tags.clone()), - Cardinality::ManyToOne(_) => left_tags.clone(), - Cardinality::OneToMany(_) => right_tags.clone(), - }; - let (includes, include_tags): (&[Arc], &Vec) = - match &matching.cardinality { - Cardinality::OneToOne => (&[], &left_tags), - Cardinality::ManyToOne(labels) => (labels, &right_tags), - Cardinality::OneToMany(labels) => (labels, &left_tags), - }; - for tag in include_tags { - if includes.iter().any(|label| label.as_ref() == tag.as_str()) - && !tags.contains(tag) - { - tags.push(tag.clone()); - } - } - // A grouped result carries an include label only when a row can give it one. - // `result_metric` removes the label from the many side's labels unless the one side - // has it, so a paired row and a row filed from a one-side series take it from the - // one series, which is what the loop above declares. A row filed from a many-side - // series takes it from that series' match key instead, so it carries the label only - // when the matching keeps the label in the key. Everything else the many side - // carries is the base above, so a label neither of those can provide has to leave - // the set: nothing would ever fill its column. - if !includes.is_empty() { - let many_rows_filled = match &matching.cardinality { - // `group_right` puts the one side on the left, so the fills trade places: - // `fill_left` is the one that answers a missing one-side sample there. - Cardinality::OneToMany(_) => matching.fill.lhs, - Cardinality::ManyToOne(_) | Cardinality::OneToOne => matching.fill.rhs, - } - .is_some(); - let in_match_key = |label: &str| match &matching.labels { - MatchingLabels::On(names) => { - names.iter().any(|name| name.as_ref() == label) - } - // A match key is built from a series' tags, which never hold the metric name. - MatchingLabels::Ignoring(names) => { - label != "__name__" && !names.iter().any(|name| name.as_ref() == label) + } => { + let keeps_name = keeps_metric_name(matching); + let scalar_side = matches!(&left.output, NativeType::Scalar(_)) + || matches!(&right.output, NativeType::Scalar(_)); + let tags = match (&left.output, &right.output) { + (NativeType::Scalar(_), _) => fused_output_tags(right, source_tags)?, + (_, NativeType::Scalar(_)) => fused_output_tags(left, source_tags)?, + _ => { + let kept = |tags: Vec| -> Vec { + match &matching.labels { + MatchingLabels::On(names) => tags + .into_iter() + .filter(|tag| { + names.iter().any(|name| name.as_ref() == tag.as_str()) + }) + .collect(), + MatchingLabels::Ignoring(names) => tags + .into_iter() + .filter(|tag| { + !names.iter().any(|name| name.as_ref() == tag.as_str()) + }) + .collect(), } }; - tags.retain(|tag| { - // The name is not an ordinary label here: this walk reports it by - // `__name__` in the set, which the nodes above decide. - tag == "__name__" - || !includes.iter().any(|label| label.as_ref() == tag.as_str()) - || include_tags.contains(tag) - || (many_rows_filled && in_match_key(tag)) - }); - } - // A fill answers a missing sample from the series of the other side, and the kernel - // files that row at the label set the series it has gives it - // (`vector_match::binary_field`): a many-side series keeps all of its labels, which - // `tags` already holds, while a one-side series keeps only its matching labels plus - // the group labels the modifier names, which the loop above adds. So what a fill can - // add here is the one side's matching labels, and only a fill that can file a - // one-side row has any to add: one where the many side is the missing side. Which - // fill that is follows the kernel, which trades the two for `group_right`, whose one - // side is the left - there it is `fill_right` that leaves the many side missing. - let (one_tags, fills_many_side) = match &matching.cardinality { - Cardinality::OneToMany(_) => (&left_tags, matching.fill.rhs), - Cardinality::ManyToOne(_) | Cardinality::OneToOne => { - (&right_tags, matching.fill.lhs) + let left_tags = fused_output_tags(left, source_tags)?; + let right_tags = fused_output_tags(right, source_tags)?; + // One-to-one matching narrows the left side's labels with `on`/`ignoring`. + // A group retains the many side's labels and copies the named labels from the one + // side: `group_left(container)` keeps the left labels and adds `container` from + // the right. + let mut tags = match &matching.cardinality { + Cardinality::OneToOne => kept(left_tags.clone()), + Cardinality::ManyToOne(_) => left_tags.clone(), + Cardinality::OneToMany(_) => right_tags.clone(), + }; + // `result_metric` drops the metadata labels before it matches or copies an + // included label, so an arithmetic result starts without the name. An include can + // still put it back from the one side, which is why the final rule below does not + // run on this branch. + if !keeps_name { + tags.retain(|tag| tag != METRIC_NAME); } - }; - if fills_many_side.is_some() { - for tag in kept(one_tags.clone()) { - if !tags.contains(&tag) { - tags.push(tag); + let (includes, include_tags): (&[Arc], &Vec) = + match &matching.cardinality { + Cardinality::OneToOne => (&[], &left_tags), + Cardinality::ManyToOne(labels) => (labels, &right_tags), + Cardinality::OneToMany(labels) => (labels, &left_tags), + }; + for tag in include_tags { + if includes.iter().any(|label| label.as_ref() == tag.as_str()) + && !tags.contains(tag) + { + tags.push(tag.clone()); } } + // A grouped result carries an include label only when a row can give it one. + // `result_metric` removes the label from the many side's labels unless the one side + // has it, so a paired row and a row filed from a one-side series take it from the + // one series, which is what the loop above declares. A row filed from a many-side + // series takes it from that series' match key instead, so it carries the label only + // when the matching keeps the label in the key. Everything else the many side + // carries is the base above, so a label neither of those can provide has to leave + // the set: nothing would ever fill its column. + if !includes.is_empty() { + let many_rows_filled = match &matching.cardinality { + // `group_right` puts the one side on the left, so the fills trade places: + // `fill_left` is the one that answers a missing one-side sample there. + Cardinality::OneToMany(_) => matching.fill.lhs, + Cardinality::ManyToOne(_) | Cardinality::OneToOne => matching.fill.rhs, + } + .is_some(); + let in_match_key = |label: &str| match &matching.labels { + MatchingLabels::On(names) => { + names.iter().any(|name| name.as_ref() == label) + } + // A match key is built from a series' tags, which never hold the metric name. + MatchingLabels::Ignoring(names) => { + label != "__name__" + && !names.iter().any(|name| name.as_ref() == label) + } + }; + tags.retain(|tag| { + !includes.iter().any(|label| label.as_ref() == tag.as_str()) + || include_tags.contains(tag) + || (many_rows_filled && in_match_key(tag)) + }); + } + // A fill answers a missing sample from the series of the other side, and the kernel + // files that row at the label set the series it has gives it + // (`vector_match::binary_field`): a many-side series keeps all of its labels, which + // `tags` already holds, while a one-side series keeps only its matching labels plus + // the group labels the modifier names, which the loop above adds. So what a fill can + // add here is the one side's matching labels, and only a fill that can file a + // one-side row has any to add: one where the many side is the missing side. Which + // fill that is follows the kernel, which trades the two for `group_right`, whose one + // side is the left - there it is `fill_right` that leaves the many side missing. + let (one_tags, fills_many_side) = match &matching.cardinality { + Cardinality::OneToMany(_) => (&left_tags, matching.fill.rhs), + Cardinality::ManyToOne(_) | Cardinality::OneToOne => { + (&right_tags, matching.fill.lhs) + } + }; + // A filled row keeps only the one side's matching labels, so the name can reach + // it only through an `on` key that names the metric name, and only while the + // operation reports its samples' names. + let fill_carries_name = keeps_name + && matches!(&matching.labels, MatchingLabels::On(names) + if names.iter().any(|name| name.as_ref() == METRIC_NAME)); + if fills_many_side.is_some() { + for tag in kept(one_tags.clone()) { + if tag == METRIC_NAME && !fill_carries_name { + continue; + } + if !tags.contains(&tag) { + tags.push(tag); + } + } + } + tags } - Some(tags) - } - }, + }; + Some(if scalar_side && !keeps_name { + without_metric_name(tags) + } else { + tags + }) + } NativeExpr::Set { left, right, op } => { let mut tags = fused_output_tags(left, source_tags)?; if op.op == SetOp::Or { @@ -355,6 +415,23 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Opti } } } +/// Whether a binary operation reports the samples of the side it keeps, name included. +/// +/// A filtering comparison reports those samples, so the name survives it. Every other binary +/// operation computes new values, and `bool` reports the comparison's own outcome rather than +/// either operand's samples, so both drop the name. +fn keeps_metric_name(matching: &BinaryMatching) -> bool { + matches!( + matching.op, + BinaryOp::Eq | BinaryOp::Ne | BinaryOp::Gt | BinaryOp::Lt | BinaryOp::Ge | BinaryOp::Le + ) && !matching.return_bool +} + +/// The labels a node reports after the metric name is dropped, which is what a kernel that computes +/// a value of its own does to the samples it reports. +fn without_metric_name(tags: Vec) -> Vec { + tags.into_iter().filter(|tag| tag != METRIC_NAME).collect() +} /// What one node produces. /// @@ -1012,10 +1089,21 @@ pub struct NativeNode { /// /// The rewrite is judged by the labels it writes and the ones its input carries, which is why the /// caller supplies what it knows about each source's tags, indexed by the `source_index` the leaves -/// were bound to: a destination that names no label of the input writes a fresh one, and a fresh -/// label cannot make two series equal. That is the difference between `label_replace(x, "dst", "$1", -/// "host", "(.*)")` - which no merge has to collect - and `label_replace(x, "host", "same", "host", -/// ".*")`, which folds every series of `x` into one label set. +/// were bound to: a destination that names no ordinary label of the input writes a fresh one, and a +/// fresh ordinary label cannot make two series equal. That is the difference between +/// `label_replace(x, "dst", "$1", "host", "(.*)")` - which no merge has to collect - and +/// `label_replace(x, "host", "same", "host", ".*")`, which folds every series of `x` into one label +/// set. +/// +/// A source's entry is its complete runtime label set, so `__name__` is one of its labels whenever +/// its series carry a name - a scan bound to one metric names every series from its binding. That is +/// what makes `label_replace(x, "__name__", "same", "__name__", ".*")` a rewrite of an existing +/// label: two metrics whose ordinary labels are equal can be told apart by name alone, and rewriting +/// that name folds them into one identity. A name write is collected even when the derived set no +/// longer carries `__name__` - after `abs`, a `timestamp` or a range function dropped it - because +/// the scan below those kernels can still hold several metrics whose ordinary labels are equal, and +/// writing one name puts them under one identity; the walk has no proof that the input reads a +/// single metric, so the conservative answer is the one that collects. /// /// Everything else either keeps the set it read or answers a set of its own: /// @@ -1104,6 +1192,12 @@ pub fn duplicates_canonical_labels(node: &NativeNode, source_tags: &[Vec /// /// The walk is over the ops in written order, starting from the labels the input reports. A /// destination the input already carries can fold two series into one; a fresh destination cannot. +/// A `__name__` destination is always treated as carried, even when the derived set has lost it: a +/// kernel such as `abs`, `timestamp` or a range function drops the name, but the scan below it can +/// still hold several metrics whose ordinary labels are equal, and writing one name puts those +/// identities under one label set. Nothing here proves the input reads a single metric, so the +/// conservative answer is the one that collects. +/// /// `None` when the input's label set is not derivable, which is treated as "it may carry the /// destination" by the caller: an unknown set proves nothing, so the conservative answer is the one /// that collects. @@ -1117,10 +1211,12 @@ fn map_writes_existing_label( for operation in op.ops.iter() { if let MapOp::Label(rewrite) = operation { let destination = rewrite.destination(); - if destination == "__name__" || tags.iter().any(|tag| tag == destination) { + if destination == METRIC_NAME || tags.iter().any(|tag| tag == destination) { existing = true; } - if !tags.iter().any(|tag| tag == destination) { + // A label the input did not carry is one the rewrite publishes only when it can write a + // value; a delete-only rewrite adds nothing a later op could read as carried. + if rewrite.writes_value() && !tags.iter().any(|tag| tag == destination) { tags.push(destination.to_string()); } } diff --git a/src/promql/src/native/plan/tests.rs b/src/promql/src/native/plan/tests.rs index 18e6e71b773..6d75e2be9b8 100644 --- a/src/promql/src/native/plan/tests.rs +++ b/src/promql/src/native/plan/tests.rs @@ -26,6 +26,7 @@ use super::{ NativeType, duplicates_canonical_labels, fused_output_tags, reads_across_series, }; use crate::extension_plan::{FusedSeries, FusedSeriesBinding}; +use crate::histogram_eval::NativeHistogramFunction; use crate::info::InfoSelector; use crate::map::MathOp; use crate::range_eval::{RangeFunction, RangeWindow}; @@ -1924,13 +1925,18 @@ fn info_reads_the_sample_times_of_one_field() { /// `label_replace(x, dst, "$1", src, "(.*)")` over `inner`, a rewrite that writes `dst`. fn rewritten(inner: NativeNode, dst: &str, src: &str) -> NativeNode { + replaced(inner, dst, "$1", src) +} + +/// `label_replace(x, dst, replacement, src, "(.*)")` over `inner`. +fn replaced(inner: NativeNode, dst: &str, replacement: &str, src: &str) -> NativeNode { let output = inner.output.clone(); NativeNode { expr: NativeExpr::Map { input: Box::new(inner), op: BoundMap { ops: Arc::new([MapOp::Label(Arc::new( - crate::map_label::LabelRewrite::replace(dst, "$1", src, "(.*)") + crate::map_label::LabelRewrite::replace(dst, replacement, src, "(.*)") .expect("the rewrite compiles"), ))]), }, @@ -1979,6 +1985,225 @@ fn a_rewrite_of_an_existing_label_aliases() { )); } +/// A scan bound to one metric names every series it hands over, so `__name__` is one of its runtime +/// labels even though it is not a tag column. A rewrite of that name writes an existing label and +/// can fold two metrics whose ordinary labels are equal into one identity. +#[test] +fn a_rewrite_of_the_metric_name_aliases_a_named_source() { + let out = grid(0, 60_000, 4); + let inner = input_on(Arc::clone(&out)); + let named = vec![vec!["host".to_string(), "__name__".to_string()]]; + assert!(duplicates_canonical_labels( + &rewritten(inner, "__name__", "__name__"), + &named + )); + // The delete-only form can fold the same two identities, so it is collected too. + let inner = input_on(Arc::clone(&out)); + assert!(duplicates_canonical_labels( + &replaced(inner, "__name__", "", "__name__"), + &named + )); +} + +/// A name write is collected even above a kernel that dropped the name: the scan below `abs`, a +/// `timestamp` or a range function can still hold several metrics whose ordinary labels are equal, +/// so writing one name puts those identities under one label set. A fresh ordinary destination is +/// still uncollected, because no scan can carry an ordinary label it does not declare. +#[test] +fn a_name_rewrite_above_a_dropped_name_is_collected() { + let out = grid(0, 60_000, 4); + let named = vec![vec!["host".to_string(), "__name__".to_string()]]; + let selected = || input_on(Arc::clone(&out)); + let timestamp = |inner: NativeNode| NativeNode { + expr: NativeExpr::Timestamp { + input: Box::new(inner), + }, + output: evaluated(fields(1), Arc::clone(&out)), + }; + for (name, dropped) in [ + ("abs", map_over(selected())), + ("timestamp", timestamp(selected())), + // `rate` computes a value of its own, so it drops the name too. + ( + "rate", + window_fields(input_with_fields(1), Arc::clone(&out), 1), + ), + ] { + assert!( + duplicates_canonical_labels( + &replaced(dropped.clone(), "__name__", "same", "missing"), + &named + ), + "{name}: a name write may alias metrics the kernel left indistinguishable" + ); + assert!( + !duplicates_canonical_labels(&rewritten(dropped, "dst", "host"), &named), + "{name}: a fresh ordinary destination cannot alias" + ); + } +} + +/// The label walk keeps a name a rewrite writes and drops it wherever the kernel does, so the +/// query's output columns never claim a name the runtime does not report. +#[test] +fn a_written_name_survives_only_where_the_tree_keeps_it() { + let out = grid(0, 60_000, 4); + let selected = || input_on(Arc::clone(&out)); + // A scan bound to one metric: its runtime labels carry the name, its columns do not. + let runtime = vec![vec!["host".to_string(), "__name__".to_string()]]; + let columns = vec![vec!["host".to_string()]]; + let written = || rewritten(selected(), "__name__", "__name__"); + let name_kept = |tags: Option>| { + tags.is_some_and(|tags| tags.iter().any(|tag| tag == "__name__")) + }; + + assert_eq!( + Some(vec!["host".to_string(), "__name__".to_string()]), + fused_output_tags(&written(), &runtime) + ); + // A column set gains the name because the rewrite writes a value into it. + assert_eq!( + Some(vec!["host".to_string(), "__name__".to_string()]), + fused_output_tags(&written(), &columns) + ); + // A delete-only rewrite over a scan that publishes no name column publishes none. + assert_eq!( + Some(vec!["host".to_string()]), + fused_output_tags(&replaced(selected(), "__name__", "", "__name__"), &columns) + ); + // The delete-only rewrite over a set that carries the name keeps it: whether a series keeps its + // value depends on the regex, so the label is still part of the set. + assert_eq!( + Some(vec!["host".to_string(), "__name__".to_string()]), + fused_output_tags(&replaced(selected(), "__name__", "", "__name__"), &runtime) + ); + + let histogram = |inner: NativeNode| NativeNode { + expr: NativeExpr::Map { + input: Box::new(inner), + op: BoundMap { + ops: Arc::new([MapOp::Histogram(NativeHistogramFunction::Count)]), + }, + }, + output: evaluated(fields(1), Arc::clone(&out)), + }; + let timestamp = |inner: NativeNode| NativeNode { + expr: NativeExpr::Timestamp { + input: Box::new(inner), + }, + output: evaluated(fields(1), Arc::clone(&out)), + }; + for (name, dropped) in [ + ("abs", map_over(written())), + ("histogram_count", histogram(written())), + ("timestamp", timestamp(written())), + ("rate", window_fields(written(), Arc::clone(&out), 1)), + ] { + assert!(!name_kept(fused_output_tags(&dropped, &runtime)), "{name}"); + assert!(!name_kept(fused_output_tags(&dropped, &columns)), "{name}"); + } +} + +/// `group_left(__name__)` copies the one side's name onto an arithmetic result: the kernel drops the +/// metadata before it copies an included label, so the name the one side carries is what the result +/// reports, while a one side without the name removes it from the result. +#[test] +fn an_included_metric_name_survives_an_arithmetic_match() { + let out = grid(0, 60_000, 4); + let input = |source_index: usize| { + let output = NativeType::Samples(Arc::new(SeriesSchema { + fields: fields(1), + eval_grid: None, + })); + NativeNode { + expr: NativeExpr::Input(BoundInput { + output: output.clone(), + source_index, + ordering: SeriesOrdering::KeyThenTime, + }), + output, + } + }; + let binary = |left: NativeNode, right: NativeNode, matching: BinaryMatching| NativeNode { + expr: NativeExpr::Binary { + left: Box::new(left), + right: Box::new(right), + matching, + }, + output: evaluated(fields(1), Arc::clone(&out)), + }; + let group_left = || BinaryMatching { + op: BinaryOp::Mul, + return_bool: false, + labels: MatchingLabels::On(vec![Arc::from("host")]), + cardinality: Cardinality::ManyToOne(vec![Arc::from("__name__")]), + fill: FillValues::default(), + }; + let source_tags = vec![ + vec!["host".to_string(), "__name__".to_string()], + vec!["host".to_string(), "__name__".to_string()], + ]; + let tagged = |source_index: usize| input(source_index); + + // The one side (right) carries a rewritten name, so the include puts it on the result. + let renamed = replaced(tagged(1), "__name__", "renamed", "__name__"); + assert_eq!( + Some(vec!["host".to_string(), "__name__".to_string()]), + fused_output_tags(&binary(tagged(0), renamed, group_left()), &source_tags) + ); + // A one side without the name cannot give the include one, so the result drops it. + let unnamed_tags = vec![ + vec!["host".to_string(), "__name__".to_string()], + vec!["host".to_string()], + ]; + assert_eq!( + Some(vec!["host".to_string()]), + fused_output_tags(&binary(tagged(0), tagged(1), group_left()), &unnamed_tags) + ); + // `group_right` trades the sides: the include is read from the left (one) side there. + let group_right = || BinaryMatching { + op: BinaryOp::Mul, + return_bool: false, + labels: MatchingLabels::On(vec![Arc::from("host")]), + cardinality: Cardinality::OneToMany(vec![Arc::from("__name__")]), + fill: FillValues::default(), + }; + let renamed = replaced(tagged(0), "__name__", "renamed", "__name__"); + assert_eq!( + Some(vec!["host".to_string(), "__name__".to_string()]), + fused_output_tags(&binary(renamed, tagged(1), group_right()), &source_tags) + ); + // A `bool` comparison drops the name like arithmetic does, and the include still copies the one + // side's name afterwards: the drop happens before the include. + let bool_comparison = || BinaryMatching { + op: BinaryOp::Eq, + return_bool: true, + labels: MatchingLabels::On(vec![Arc::from("host")]), + cardinality: Cardinality::ManyToOne(vec![Arc::from("__name__")]), + fill: FillValues::default(), + }; + let renamed = replaced(tagged(1), "__name__", "renamed", "__name__"); + assert_eq!( + Some(vec!["host".to_string(), "__name__".to_string()]), + fused_output_tags(&binary(tagged(0), renamed, bool_comparison()), &source_tags) + ); + + // A filtering comparison keeps the name of the side it reports, but an included name the one + // side does not carry still leaves the set: `result_metric` removes an included label it cannot + // copy, and the base name is not exempt. + let filtering = || BinaryMatching { + op: BinaryOp::Eq, + return_bool: false, + labels: MatchingLabels::On(vec![Arc::from("host")]), + cardinality: Cardinality::ManyToOne(vec![Arc::from("__name__")]), + fill: FillValues::default(), + }; + assert_eq!( + Some(vec!["host".to_string()]), + fused_output_tags(&binary(tagged(0), tagged(1), filtering()), &unnamed_tags) + ); +} + /// A source the caller says nothing about proves nothing: the pass has to assume the destination /// was one of its labels, because two series sharing one is exactly the wrong answer to ship. #[test] @@ -2121,14 +2346,20 @@ fn count_values_reports_its_grouping_and_the_label_it_writes() { &source_tags ) ); - // A destination naming the metric name writes no column this walk can name. + // A destination naming the metric name writes the name, which is the one label the node's + // result carries when its grouping names nothing else. assert_eq!( - None, + tags(&["__name__"]), fused_output_tags( &counted("__name__", Grouping::By(Arc::new([]))), &source_tags ) ); + // An empty destination cannot be a label, so the walk declines rather than name a column. + assert_eq!( + None, + fused_output_tags(&counted("", Grouping::By(Arc::new([]))), &source_tags) + ); } /// A consumer that files its rows by the label set a sample ends under answers one series per set, diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index 396e2234d0c..f292293c19c 100644 --- a/src/query/src/promql/planner/native.rs +++ b/src/query/src/promql/planner/native.rs @@ -63,8 +63,8 @@ use promql::extension_plan::{ use promql::native::execute::NativeExecutionState; use promql::native::plan::{ AggregateKind, BoundAggregate, BoundInput, BoundVectorOp, Grouping, InstantValue, MapOp, - NativeExpr, NativeNode, NativeTree, NativeType, binary_output_field_kind, fused_output_tags, - reads_across_series, uniform_field_kind, + NativeExpr, NativeNode, NativeTree, NativeType, binary_output_field_kind, reads_across_series, + uniform_field_kind, }; use promql::vector_match::{Cardinality, MatchingLabels, SetOp}; use promql_parser::label::{METRIC_NAME, MatchOp, Matchers}; @@ -1321,6 +1321,23 @@ impl PromPlanner { .iter() .map(|(_, _, _, _, tags, _, _)| tags.clone()) .collect::>(); + // The complete runtime label set of each input: a scan bound to one metric names every + // series it hands over, so `__name__` is one of its labels even though it is not a tag + // column. The canonical pass judges a rewrite by the labels it writes and the ones its input + // carries, so rewriting the name of a named scan is a rewrite of an existing label - and a + // discovered source declares the name among its tags itself. The output-column walks keep + // `source_tags`: the name becomes a column only when a rewrite writes one, or when the + // source declares it. + let runtime_source_tags = prepared + .iter() + .map(|(_, _, metric, _, tags, _, _)| { + let mut tags = tags.clone(); + if metric.is_some() && !tags.iter().any(|tag| tag == METRIC_NAME) { + tags.push(METRIC_NAME.to_string()); + } + tags + }) + .collect::>(); // Whether any input's table is split into regions along a partition column. The canonical // boundary is the only part of this route that claims a whole-query label-set answer, and a // region-local producer cannot see the pieces another region holds, so this fact decides @@ -1334,7 +1351,7 @@ impl PromPlanner { // matters and nowhere else, so a tree with no collapsing rewrite keeps every partition. let tree = match promql::native::canonicalize::merge_label_rewrites( &compiled.tree, - &source_tags, + &runtime_source_tags, ) { Ok(tree) => tree, Err(_) => return Ok(None), @@ -2375,7 +2392,7 @@ fn canonical_boundary_stages( .iter() .map(|field| field.name.to_string()) .collect::>(); - let output_tags = fused_output_tags(tree.root(), source_tags)?; + let output_tags = native_output_tags(tree.root(), source_tags)?; let output_schema = fused_output_schema(&fields, &output_tags, time_index, root_kind).ok()?; // The producer: the statement's own tree, run per partition, exporting a frame. @@ -2764,7 +2781,7 @@ fn segment_tree( )?; match plan { Some(plan) => { - let Some(cut_tags) = fused_output_tags(cut, ¤t_tags) else { + let Some(cut_tags) = native_output_tags(cut, ¤t_tags) else { return Ok(None); }; Some((plan, cut_tags)) @@ -3386,11 +3403,18 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Option< for operation in op.ops.iter() { if let MapOp::Label(rewrite) = operation { let destination = rewrite.destination(); - if destination == "__name__" { - return None; + if tags.iter().any(|tag| tag == destination) { + // The input carries the label - a discovered source declares `__name__` + // among its tags - so the rewrite changes its value, and an empty + // replacement may remove it, which the nullable column still represents. + tags.retain(|tag| tag != destination); + tags.push(destination.to_string()); + } else if rewrite.writes_value() { + // A fresh destination is a column only when the rewrite can write a + // value: that is how a `__name__` destination publishes a name column, + // and how a delete-only rewrite publishes none. + tags.push(destination.to_string()); } - tags.retain(|tag| tag != destination); - tags.push(destination.to_string()); } } return Some(tags); @@ -3486,6 +3510,8 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Option< matching, } => { let keeps_name = keeps_metric_name(matching); + let scalar_side = matches!(&left.output, NativeType::Scalar(_)) + || matches!(&right.output, NativeType::Scalar(_)); let tags = match (&left.output, &right.output) { (NativeType::Scalar(_), _) => native_output_tags(right, source_tags)?, (_, NativeType::Scalar(_)) => native_output_tags(left, source_tags)?, @@ -3518,6 +3544,13 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Option< Cardinality::ManyToOne(_) => left_tags.clone(), Cardinality::OneToMany(_) => right_tags.clone(), }; + // `result_metric` drops the metadata labels before it matches or copies an + // included label, so an arithmetic result starts without the name. An include can + // still put it back from the one side, which is why the final rule below does not + // run on this branch. + if !keeps_name { + tags.retain(|tag| tag != METRIC_NAME); + } let (includes, include_tags) = match &matching.cardinality { Cardinality::OneToOne => (&[][..], &[][..]), Cardinality::ManyToOne(labels) => { @@ -3561,10 +3594,7 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Option< } }; tags.retain(|tag| { - // The name is not an ordinary label here: `keeps_name` above decides - // whether the walk reports it. - tag == METRIC_NAME - || !includes.iter().any(|label| label.as_ref() == tag.as_str()) + !includes.iter().any(|label| label.as_ref() == tag.as_str()) || include_tags.contains(tag) || (many_rows_filled && in_match_key(tag)) }); @@ -3585,8 +3615,17 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Option< (right_tags.as_slice(), matching.fill.lhs) } }; + // A filled row keeps only the one side's matching labels, so the name can reach + // it only through an `on` key that names the metric name, and only while the + // operation reports its samples' names. + let fill_carries_name = keeps_name + && matches!(&matching.labels, MatchingLabels::On(names) + if names.iter().any(|name| name.as_ref() == METRIC_NAME)); if fills_many_side.is_some() { for tag in kept(one_tags.to_vec()) { + if tag == METRIC_NAME && !fill_carries_name { + continue; + } if !tags.contains(&tag) { tags.push(tag); } @@ -3595,10 +3634,10 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec]) -> Option< tags } }; - Some(if keeps_name { - tags - } else { + Some(if scalar_side && !keeps_name { without_metric_name(tags) + } else { + tags }) } NativeExpr::Set { left, right, op } => { @@ -4108,15 +4147,15 @@ mod tests { use datafusion::physical_plan::ExecutionPlan; use datatypes::series::SeriesOrdering; use datatypes::series::samples::{FieldSpec, SeriesSchema}; - use promql::native::plan::{NativeExpr, NativeNode, NativeTree}; + use promql::native::plan::{NativeExpr, NativeNode, NativeTree, fused_output_tags}; use session::context::QueryContextBuilder; use super::{ ArrowDataType, ArrowTimeUnit, FieldKind, METRIC_NAME, NATIVE_PROMQL_EXTENSION_KEY, NativeType, aggregate_reads_histogram, collect_selectors_with_set_operations, cut_segments, - fused_output_schema, fused_output_tags, independent_scan_pair, native_output_tags, - native_promql_requested, native_raw_plan_bounds_region_reads, node_children, - plans_over_remote_regions, scan_time_filter, + fused_output_schema, independent_scan_pair, native_output_tags, native_promql_requested, + native_raw_plan_bounds_region_reads, node_children, plans_over_remote_regions, + scan_time_filter, }; #[test] @@ -4696,6 +4735,99 @@ mod tests { ); } + /// A `__name__` rewrite publishes a name column, canonicalizes the complete runtime label set, + /// and leaves the name dropped wherever a kernel computes a value of its own. + #[test] + fn name_rewrite_publishes_a_column_and_is_canonicalized() { + let columns = [vec!["host".to_string(), "kind".to_string()]]; + let runtime = [vec![ + "host".to_string(), + "kind".to_string(), + METRIC_NAME.to_string(), + ]]; + let tree = + compiled_test_tree(r#"label_replace(x, "__name__", "same", "__name__", "(.*)")"#); + assert_eq!( + Some(vec!["host".into(), "kind".into(), METRIC_NAME.into()]), + native_output_tags(tree.root(), &columns), + "a written name is a column the output publishes" + ); + let collected = promql::native::canonicalize::merge_label_rewrites(&tree, &runtime) + .expect("the pass produces a legal tree"); + assert!( + matches!(collected.root().expr, NativeExpr::Merge { .. }), + "a name rewrite writes an existing runtime label: {collected}" + ); + assert_eq!( + native_output_tags(tree.root(), &columns), + native_output_tags(collected.root(), &columns), + "collecting the pieces does not change the columns" + ); + + // A delete-only rewrite over a scan whose columns carry no name publishes none. + let deleted = compiled_test_tree(r#"label_replace(x, "__name__", "", "__name__", "(.*)")"#); + assert_eq!( + Some(vec!["host".into(), "kind".into()]), + native_output_tags(deleted.root(), &columns) + ); + + // A kernel that computes a value of its own drops the name a rewrite below it wrote. + for query in [ + r#"abs(label_replace(x, "__name__", "same", "__name__", "(.*)"))"#, + r#"histogram_count(label_replace(x, "__name__", "same", "__name__", "(.*)"))"#, + r#"timestamp(label_replace(x, "__name__", "same", "__name__", "(.*)"))"#, + r#"sum(label_replace(x, "__name__", "same", "__name__", "(.*)"))"#, + ] { + let tree = compiled_test_tree(query); + let tags = native_output_tags(tree.root(), &columns) + .unwrap_or_else(|| panic!("{query} derives its columns")); + assert!( + !tags.iter().any(|tag| tag == METRIC_NAME), + "{query}: {tags:?}" + ); + } + } + + /// A name write is collected even above a kernel that dropped the name: the scan below it can + /// still hold several metrics whose ordinary labels are equal, so writing one name can fold them + /// into one identity. The output still publishes the written name. + #[test] + fn a_name_rewrite_above_a_dropped_name_is_collected() { + let runtime = [vec![ + "host".to_string(), + "kind".to_string(), + METRIC_NAME.to_string(), + ]]; + let tree = + compiled_test_tree(r#"label_replace(abs(x), "__name__", "same", "missing", "(.*)")"#); + let collected = promql::native::canonicalize::merge_label_rewrites(&tree, &runtime) + .expect("the pass produces a legal tree"); + assert!( + matches!(collected.root().expr, NativeExpr::Merge { .. }), + "the merge sits above the name rewrite: {collected}" + ); + assert_eq!( + 1, + collected.to_string().matches("Merge").count(), + "exactly one merge: {collected}" + ); + // The rewrite writes a name over a dropped one, so the output publishes a name column. + assert_eq!( + Some(vec!["host".into(), "kind".into(), METRIC_NAME.into()]), + native_output_tags(collected.root(), &[vec!["host".into(), "kind".into()]]) + ); + + // A fresh ordinary destination above the same kernel is still uncollected: no scan can carry + // an ordinary label it does not declare. + let fresh = compiled_test_tree(r#"label_replace(abs(x), "dst", "$1", "host", "(.*)")"#); + let collected = promql::native::canonicalize::merge_label_rewrites(&fresh, &runtime) + .expect("the pass produces a legal tree"); + assert!( + !collected.to_string().contains("Merge"), + "a fresh ordinary destination does not alias: {collected}" + ); + } + #[test] fn group_right_copies_included_labels_into_output_tags() { let sources = [ @@ -4709,6 +4841,32 @@ mod tests { ); } + /// `group_left(__name__)` copies the one side's name onto an arithmetic result: the operation + /// drops the name before it copies an included label, so the copied name is a column. A one side + /// without a name cannot provide one, so the column leaves the set. + #[test] + fn an_included_metric_name_is_an_output_column() { + let sources = [ + vec!["host".to_string()], + vec!["host".to_string(), METRIC_NAME.to_string()], + ]; + let included = compiled_test_tree( + r#"x * on(host) group_left(__name__) label_replace(y, "__name__", "renamed", "__name__", "(.*)")"#, + ); + assert_eq!( + Some(vec!["host".to_string(), METRIC_NAME.to_string()]), + native_output_tags(included.root(), &sources) + ); + let unnamed = compiled_test_tree(r#"x * on(host) group_left(__name__) y"#); + assert_eq!( + Some(vec!["host".to_string()]), + native_output_tags( + unnamed.root(), + &[vec!["host".to_string()], vec!["host".to_string()]] + ) + ); + } + #[test] fn numeric_aggregate_projection_keeps_grouping_tags_and_value_schema() { let sources = [vec![ diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index 9d3128d47b4..d6725db4981 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -5559,6 +5559,10 @@ async fn native_fused_aggregate_splits_into_a_final_over_a_partial_producer() { "stdvar(some_metric)", "sum by (tag_0) (rate(some_metric[5m]))", "sum without (tag_0) (some_metric)", + // The output columns are the whole tree's, not the Final half's state input's: `dst` is + // created by the prefix rewrite, so a grouping that keeps or names it publishes it. + r#"sum without (tag_0) (label_replace(some_metric, "dst", "$1", "tag_0", "(.*)"))"#, + r#"sum by (dst) (label_replace(some_metric, "dst", "$1", "tag_0", "(.*)"))"#, "sum(rate(some_metric[5m]))", "ceil(sum(some_metric))", // A literal scalar on either side of the fold. Below the aggregate it is part of the prefix, @@ -5671,6 +5675,12 @@ async fn native_fused_aggregate_splits_into_a_final_over_a_partial_producer() { "stdvar(some_metric)" => &["field_0", "timestamp"], "sum by (tag_0) (rate(some_metric[5m]))" => &["field_0", "tag_0", "timestamp"], "sum without (tag_0) (some_metric)" => &["field_0", "timestamp"], + r#"sum without (tag_0) (label_replace(some_metric, "dst", "$1", "tag_0", "(.*)"))"# => { + &["dst", "field_0", "timestamp"] + } + r#"sum by (dst) (label_replace(some_metric, "dst", "$1", "tag_0", "(.*)"))"# => { + &["dst", "field_0", "timestamp"] + } "sum(rate(some_metric[5m]))" => &["field_0", "timestamp"], "ceil(sum(some_metric))" => &["field_0", "timestamp"], "sum(avg(some_metric) by (tag_0))" => &["field_0", "timestamp"], @@ -6754,6 +6764,92 @@ async fn native_fused_label_rewrites_keep_legacy_output_labels() { } } +/// A `__name__` destination is native-only: the existing planner refuses the destination outright, +/// so the only plan that can answer it is this one. What it must get right is which rewrites publish +/// the name as a column and which leave the output without one. +#[tokio::test] +async fn native_fused_metric_name_rewrites_publish_or_drop_the_name() { + for (query, keeps_name) in [ + // The rewrite at the statement root, and a join that writes a value into the name. + ( + r#"label_replace(some_metric, "__name__", "same", "__name__", "(.*)")"#, + true, + ), + (r#"label_join(some_metric, "__name__", "-", "tag_0")"#, true), + // A delete-only rewrite of a scan that publishes no name column publishes none. + ( + r#"label_replace(some_metric, "__name__", "", "__name__", "(.*)")"#, + false, + ), + (r#"label_join(some_metric, "__name__", "-")"#, false), + // A kernel above the rewrite computes a value of its own and drops the name with it. + ( + r#"abs(label_replace(some_metric, "__name__", "same", "__name__", "(.*)"))"#, + false, + ), + ( + r#"histogram_count(label_replace(some_metric, "__name__", "same", "__name__", "(.*)"))"#, + false, + ), + ( + r#"timestamp(label_replace(some_metric, "__name__", "same", "__name__", "(.*)"))"#, + false, + ), + // A fresh name above a dropped one is a new label, and still becomes a column. + ( + r#"label_replace(abs(some_metric), "__name__", "same", "__name__", "(.*)")"#, + true, + ), + ] { + let plan = build_native_fused_plan(query, true).await; + let display = plan.display_indent().to_string(); + assert!( + find_fused_series(&plan).is_some(), + "`{query}` must run natively: {display}" + ); + assert_eq!( + plan.schema() + .fields() + .iter() + .any(|field| field.name() == METRIC_NAME), + keeps_name, + "`{query}`: {display}" + ); + } +} + +/// A rewrite of the metric name over a discovered selector writes a label the source declares, so +/// two metrics whose ordinary labels are equal are collected by the canonical boundary. +#[tokio::test] +async fn native_metric_discovery_name_rewrite_canonicalizes_by_name() { + let plan = build_native_fused_plan_with_provider( + r#"label_replace({tag_0="x"}, "__name__", "same", "__name__", "(.*)")"#, + true, + metric_discovery_provider(&["alpha_total", "beta_total"]).await, + ) + .await; + let display = plan.display_indent().to_string(); + assert!( + plan.schema() + .fields() + .iter() + .any(|field| field.name() == METRIC_NAME), + "the rewritten name is a column of the answer: {display}" + ); + let fused = find_fused_series(&plan).expect("fused plan"); + assert!( + fused.children().iter().any(|child| matches!( + child, + LogicalPlan::Extension(Extension { node }) + if node + .as_any() + .downcast_ref::() + .is_some_and(FusedSeries::is_canonical_producer) + )), + "the name rewrite runs through a canonical boundary: {display}" + ); +} + #[tokio::test] async fn native_fused_preparation_decline_preserves_legacy_plan() { let query = "absent(vector(1)) + on(tag_0) some_metric{tag_0=\"x\",__field__=\"field_0\"}"; diff --git a/tests-integration/src/tests/promql_test.rs b/tests-integration/src/tests/promql_test.rs index 616e4bd45da..b8a1794ce23 100644 --- a/tests-integration/src/tests/promql_test.rs +++ b/tests-integration/src/tests/promql_test.rs @@ -5780,9 +5780,8 @@ async fn native_promql_label_rewrites_reach_the_native_route_and_keep_prometheus ); } - // A `__name__` destination is not covered here on purpose: the existing planner refuses that - // destination outright ("Invalid destination label name in label_replace()"), so no route can - // answer it and admitting the shape natively would only turn a refusal into a different one. + // A `__name__` destination is native-only - the existing planner refuses it - and is covered by + // `native_promql_rewrites_the_metric_name` below. // Two stored series rewritten to one label set, holding values at different evaluation points, // are one series carrying both points rather than two rows. @@ -5800,6 +5799,269 @@ async fn native_promql_label_rewrites_reach_the_native_route_and_keep_prometheus ); } +/// A `__name__` destination is native-only: the existing planner refuses that destination, so this +/// test reads the native route's own answer and the plan that produced it, and never compares +/// against a legacy run. +/// +/// Three logical metrics share one physical table. `name_left` and `name_right` hold the same two +/// ordinary label sets ten minutes apart, so rewriting both names to one value puts their samples +/// under one canonical identity; `name_clash` holds a sample where `name_left` already has one under +/// `host="h"`, so the same rewrite has a genuine same-point conflict to report. +#[tokio::test(flavor = "multi_thread")] +async fn native_promql_rewrites_the_metric_name() { + common_telemetry::init_default_ut_logging(); + let mito_config = mito2::config::MitoConfig { + experimental_series_scan_v2: true, + ..Default::default() + }; + let (standalone, observer) = native_test_instance( + GreptimeDbStandaloneBuilder::new("native_promql_metric_name_rewrite") + .with_mito_config(mito_config), + ) + .await; + let instance = standalone.fe_instance().clone(); + execute_all( + &instance, + "CREATE TABLE name_phy (ts TIMESTAMP TIME INDEX, val DOUBLE) ENGINE=metric WITH ('physical_metric_table'='true'); \ + CREATE TABLE name_left (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(host)) ENGINE=metric WITH ('on_physical_table'='name_phy'); \ + CREATE TABLE name_right (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(host)) ENGINE=metric WITH ('on_physical_table'='name_phy'); \ + CREATE TABLE name_clash (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(host)) ENGINE=metric WITH ('on_physical_table'='name_phy'); \ + INSERT INTO name_left (host, ts, val) VALUES ('h', 0, 1), ('i', 0, 2); \ + INSERT INTO name_right (host, ts, val) VALUES ('h', 600000, 10), ('i', 600000, 20); \ + INSERT INTO name_clash (host, ts, val) VALUES ('h', 0, 100)", + QueryContext::arc(), + ) + .await; + + let run = |expr: &str, range: &str| { + let sql = format!("TQL EVAL {range} {expr}"); + let instance = instance.clone(); + async move { + instance + .do_query(&sql, native_query_context(true, 4)) + .await + .remove(0) + } + }; + // A nameless selector is answered by enumerating the schema's metric tables, which the + // frontend's PromQL entry point bounds by the caller's grants before it plans. A SQL + // `TQL EVAL` statement prepares no such scope, so the test states the unrestricted one a + // caller with every grant would get. + let discovery_ctx = Arc::new({ + let mut ctx = QueryContext::with_db_name(None); + ctx.set_extension(QUERY_PARALLELISM_HINT, 4.to_string()); + ctx.set_extension(NATIVE_PROMQL_EXTENSION_KEY, "true"); + ctx.set_metric_discovery_scope(MetricDiscoveryScope::Unrestricted); + ctx + }); + let run_discovery = |expr: &str, range: &str| { + let sql = format!("TQL EVAL {range} {expr}"); + let instance = instance.clone(); + let ctx = Arc::clone(&discovery_ctx); + async move { instance.do_query(&sql, ctx).await.remove(0) } + }; + let labels = |pairs: &[(&str, &str)]| { + pairs + .iter() + .map(|(name, value)| ((*name).to_string(), (*value).to_string())) + .collect::>() + }; + // The boundary is a partition-local producer that merges what its own scan partition held and + // the canonical router that collects every partition's pieces of one label set. The router is + // only built over a canonical producer, so finding it by name is what proves the plan answered + // through that boundary rather than some other way. + let expect_canonical_boundary = |context: &str| { + assert!( + observer.fused().is_some(), + "{context}: a native fused node answered" + ); + assert!( + canonical_router(&observer).is_some(), + "{context}: the canonical router collects each label set: {:?}", + observer + .all_plans() + .iter() + .map(display_plan_tree) + .collect::>() + ); + }; + + // Renaming the name keeps series whose other labels differ distinct, and publishes the new name. + observer.clear(); + let renamed = run( + r#"label_replace(name_left, "__name__", "renamed", "__name__", "(.*)")"#, + "(0, 0, '1s')", + ) + .await + .unwrap(); + let (columns, rows) = native_labelled_rows(renamed).await; + assert!( + columns.iter().any(|column| column == "__name__"), + "the rewritten name is a column: {columns:?}" + ); + assert_eq!( + rows, + vec![ + (0, labels(&[("__name__", "renamed"), ("host", "h")]), 1.0), + (0, labels(&[("__name__", "renamed"), ("host", "i")]), 2.0), + ], + "a name rewrite reports the new name and keeps distinct series distinct" + ); + expect_canonical_boundary("a name rewrite of a named scan"); + + // Deleting the name reports no name, and a delete-only rewrite of a scan that publishes no name + // column publishes none: whether a series keeps its value depends on the regex, and a deleted + // value is the only one this rewrite can write. + observer.clear(); + let deleted = run( + r#"label_replace(name_left, "__name__", "", "__name__", "(.*)")"#, + "(0, 0, '1s')", + ) + .await + .unwrap(); + let (columns, rows) = native_labelled_rows(deleted).await; + assert!( + !columns.iter().any(|column| column == "__name__"), + "a deleted name publishes no column: {columns:?}" + ); + assert_eq!( + rows, + vec![ + (0, labels(&[("host", "h")]), 1.0), + (0, labels(&[("host", "i")]), 2.0), + ], + "a deleted name is absent from the answer" + ); + expect_canonical_boundary("a name deletion"); + + // `label_join` writes the name as well, and an empty source list deletes it. + for (expr, expected) in [ + ( + r#"label_join(name_left, "__name__", "-", "host")"#, + vec![ + (0, labels(&[("__name__", "h"), ("host", "h")]), 1.0), + (0, labels(&[("__name__", "i"), ("host", "i")]), 2.0), + ], + ), + ( + r#"label_join(name_left, "__name__", "-")"#, + vec![ + (0, labels(&[("host", "h")]), 1.0), + (0, labels(&[("host", "i")]), 2.0), + ], + ), + ] { + observer.clear(); + let joined = run(expr, "(0, 0, '1s')").await.unwrap(); + let (_, rows) = native_labelled_rows(joined).await; + assert_eq!(rows, expected, "{expr}"); + expect_canonical_boundary(expr); + } + + // Two metrics told apart only by name: rewriting both names to one value merges the points each + // of them holds, because the canonical identity is the complete label set with the new name. + observer.clear(); + let complementary = run_discovery( + r#"label_replace({__name__=~"name_left|name_right", host=~"h|i"}, "__name__", "same", "__name__", "(.*)")"#, + "(0, 600, '10m')", + ) + .await + .unwrap(); + let (_, rows) = native_labelled_rows(complementary).await; + assert_eq!( + rows, + vec![ + (0, labels(&[("__name__", "same"), ("host", "h")]), 1.0), + ( + 600_000, + labels(&[("__name__", "same"), ("host", "h")]), + 10.0 + ), + (0, labels(&[("__name__", "same"), ("host", "i")]), 2.0), + ( + 600_000, + labels(&[("__name__", "same"), ("host", "i")]), + 20.0 + ), + ], + "one identity carries both metrics' points" + ); + expect_canonical_boundary("two metrics renamed to one name"); + + // A name written above `abs` is collected even though `abs` dropped the name: the two metrics are + // still told apart by name alone below it, so writing one name folds their samples into one + // identity. Without the collection each metric would answer as its own series under one label set. + observer.clear(); + let rewritten_above_abs = run_discovery( + r#"label_replace(abs({__name__=~"name_left|name_right", host=~"h|i"}), "__name__", "same", "", ".*")"#, + "(0, 600, '10m')", + ) + .await + .unwrap(); + let (_, rows) = native_labelled_rows(rewritten_above_abs).await; + assert_eq!( + rows, + vec![ + (0, labels(&[("__name__", "same"), ("host", "h")]), 1.0), + ( + 600_000, + labels(&[("__name__", "same"), ("host", "h")]), + 10.0 + ), + (0, labels(&[("__name__", "same"), ("host", "i")]), 2.0), + ( + 600_000, + labels(&[("__name__", "same"), ("host", "i")]), + 20.0 + ), + ], + "one identity carries both metrics' points after `abs` dropped their names" + ); + expect_canonical_boundary("a name rewrite above a dropped name"); + + // The same shape at one point is that identity's duplicate-label-set error, and it is reported + // even though the two series came from different metrics and a math kernel erased their names. + observer.clear(); + let renamed_above_abs_clash = run_discovery( + r#"label_replace(abs({__name__=~"name_left|name_clash", host="h"}), "__name__", "same", "", ".*")"#, + "(0, 0, '1s')", + ) + .await; + let error = match renamed_above_abs_clash { + Ok(output) => collect_error(output).await, + Err(error) => error.to_string(), + }; + assert!( + error.contains("cannot contain metrics with the same labelset"), + "a name written above `abs` reports the duplicate label set: {error}" + ); + assert!( + observer.fused().is_some(), + "the collision after a dropped name is reported by a native plan" + ); + + // A point where the other metric already holds a sample is the duplicate-label-set error, and it + // is reported even though the two series came from different metrics. + observer.clear(); + let clash = run_discovery( + r#"label_replace({__name__=~"name_left|name_clash", host="h"}, "__name__", "same", "__name__", "(.*)")"#, + "(0, 0, '1s')", + ) + .await; + let error = match clash { + Ok(output) => collect_error(output).await, + Err(error) => error.to_string(), + }; + assert!( + error.contains("cannot contain metrics with the same labelset"), + "two metrics renamed to one identity report the duplicate label set: {error}" + ); + assert!( + observer.fused().is_some(), + "the collision is reported by a native plan" + ); +} + #[tokio::test(flavor = "multi_thread")] async fn native_promql_executes_fused_node_and_matches_legacy() { common_telemetry::init_default_ut_logging();