From 6106a6ea085548252bb3b454e18e185cdf403166 Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Wed, 30 Sep 2026 13:15:59 +0800 Subject: [PATCH] refactor(promql): unify semantic label resolution and label rewrites Converge the ordinary-label and __name__ paths that had forked during the rebase: - resolve_label_col() is the single label-to-column entry: __name__ reads the marked identity column, ordinary labels resolve like before but can no longer hit the marked column by its physical name, and a missing label is absent for every caller (label functions, aggregation, sort). - label_replace keeps one algorithm for every destination: read source (missing as ""), match anchored regex, expand captures (empty -> NULL), keep the old destination value when unmatched. Ordinary destinations may overwrite existing labels again (#9385 semantics); collapsing series is refused by the existing runtime label-set check instead of a plan-time InvalidDestinationLabelName. - write_destination_label() is the single destination-column entry for label_replace and label_join, so writing an empty label has one representation and a zero-source label_join on __name__ removes the identity instead of projecting an unmarked physical __name__ column. - drop_metric_name() now asserts a unique label set per evaluation timestamp when dropping the identity of a union that may ignore it (abs(a or on(__name__) b)), reusing assert_unique_match_group; TQL strip_metric_name_column() stays free of PromQL series checks. - sort_by_label(..., "__name__") on input without a name sorts by a missing label instead of failing with field_not_found. Production code is a net reduction (-38 lines); the removed ordinary-label regex branch, the overwrite tri-state and the duplicated destination construction are replaced by the three shared entries. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/query/src/promql/planner.rs | 602 +++++++++++++-------------- src/query/src/promql/planner/test.rs | 421 ++++++++++++++----- 2 files changed, 602 insertions(+), 421 deletions(-) diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index 834ca190c63..27ac010a545 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -54,6 +54,7 @@ use datafusion::logical_expr::{ use datafusion::prelude as df_prelude; use datafusion::prelude::{Column, Expr as DfExpr, JoinType}; use datafusion::scalar::ScalarValue; +use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; use datafusion_common::{DFSchema, NullEquality, TableReference}; use datafusion_expr::expr::WindowFunctionParams; use datafusion_expr::expr_fn::when; @@ -66,7 +67,7 @@ use itertools::Itertools; use once_cell::sync::Lazy; use promql::extension_plan::{ EmptyMetric, InstantManipulate, Millisecond, RangeManipulate, SeriesDivide, SeriesNormalize, - build_special_time_expr, + UnionDistinctOn, build_special_time_expr, }; use promql::functions::{ AbsentOverTime, AvgOverTime, Changes, CountOverTime, Delta, Deriv, DoubleExponentialSmoothing, @@ -1190,16 +1191,81 @@ impl PromPlanner { ) } - fn drop_metric_name(&mut self, input: LogicalPlan) -> Result { + /// Drops the metric-name identity of `input`, the result of every function, binary or unary + /// operator that does not keep it. + /// + /// The label set is what names a series in PromQL, so dropping the identity of a plan that + /// unions several series sets can leave rows of one evaluation step with the same label set - + /// `abs(a or on(__name__) b)` for two metrics that share every ordinary label. A vector cannot + /// hold those rows: every reader - the HTTP layer above all - would have to pick one of them, + /// so the plan refuses them instead, the same way a label rewrite that collapses series does. + fn drop_metric_name(&mut self, mut input: LogicalPlan) -> Result { if let Some(marker) = Self::metric_name_column(input.schema())? { self.ctx.tag_columns.retain(|name| name != &marker.name); self.ctx .aggregation_field_labels .retain(|name| name != &marker.name); + + // The plan below is completed, so its labels are the columns of its schema next to + // the identity; a column of the context that it does not project is not a label of it. + let labels = self + .ctx + .tag_columns + .iter() + .filter(|label| input.schema().has_column_with_unqualified_name(label)) + .cloned() + .collect::>(); + let time_index_column = self + .ctx + .time_index_column + .as_deref() + .filter(|column| input.schema().has_column_with_unqualified_name(column)); + if Self::unions_series_sets(&input) + && let Some(time_index_column) = time_index_column + { + let group_exprs = labels + .iter() + .map(|label| DfExpr::Column(Column::from_name(label))) + .collect(); + input = Self::assert_unique_match_group( + input, + group_exprs, + labels, + DfExpr::Column(Column::from_name(time_index_column)), + MatchGroupViolation::DuplicateLabelSet, + )?; + } } Self::strip_metric_name_column(input) } + /// Whether `plan` holds the union of `or`, whose matching may ignore the identity - `on(__name__)` + /// matches two metrics on it alone. Two rows of such a union that share every ordinary label + /// can still come from different series, and only their identity tells them apart, so only the + /// union needs the label-set check when that identity is dropped. + /// + /// Every other operator either matches on the labels themselves (a join keeps only rows whose + /// labels agree) or asserts its own uniqueness where it can multiply rows, so a duplicate label + /// set cannot reach them without one of those checks having run. + fn unions_series_sets(plan: &LogicalPlan) -> bool { + let mut unioned = false; + // The walk stops at the first union, so the visit never fails. + let _ = plan.apply(|node| { + unioned = match node { + LogicalPlan::Extension(extension) => { + extension.node.as_any().is::() + } + _ => false, + }; + Ok(if unioned { + TreeNodeRecursion::Stop + } else { + TreeNodeRecursion::Continue + }) + }); + unioned + } + fn negate_field_columns(&mut self, input: LogicalPlan) -> Result { let input_schema = input.schema().clone(); self.projection_for_each_field_column(input, |col| { @@ -2143,6 +2209,26 @@ impl PromPlanner { Ok(found) } + /// Resolve the PromQL label `label` to the column of `input_schema` that carries its value. + /// + /// `__name__` is the semantic metric name: it resolves through the marked identity column of + /// the input and never through a physical column of that name. Any other label names a column + /// the input projects, read exactly like the label functions have always read their sources - + /// a label the input does not carry is absent, and `None` leaves it to the caller to read that + /// absence as the empty string, a NULL or a dropped label. The marked identity and the metric + /// engine's internal columns are physical details, not labels: only their physical names reach + /// them, and a column that merely looks like one stays an ordinary label. + fn resolve_label_col(input_schema: &DFSchemaRef, label: &str) -> Result> { + let metric_name_column = Self::metric_name_column(input_schema)?; + if label == METRIC_NAME { + return Ok(metric_name_column); + } + if metric_name_column.is_some_and(|column| column.name == label) { + return Ok(None); + } + Ok(Self::find_case_sensitive_column(input_schema, label).map(Column::from_name)) + } + /// Strips the metric-name identity from `plan`'s output, if it carries one. /// /// The identity is an ordinary column distinguished only by its field metadata; this projects @@ -3255,9 +3341,6 @@ impl PromPlanner { Ok(vec![self.create_time_index_column_expr()?]) } Some(LabelModifier::Include(labels)) => { - // `__name__` is not a physical column: it resolves through the marked identity - // column of the input, never through a physical column of that name. - let metric_name_column = Self::metric_name_column(input_schema)?; if update_ctx { // A `by(...)` label can name a value field of the input instead of a tag. The // aggregate still reports it among its tag columns below, but unlike a tag it @@ -3282,29 +3365,17 @@ impl PromPlanner { } let mut exprs = Vec::with_capacity(labels.labels.len()); for label in &labels.labels { - if is_metric_engine_internal_column(label) { + // A missing label - the identity of an input that dropped it included - is + // ignored, exactly like any other label the input does not carry: the marked + // identity and the metric engine's internal columns are not labels, and only + // `__name__` names the identity. + let Some(column) = Self::resolve_label_col(input_schema, label)? else { continue; - } - // Ignore a missing identity, exactly like any other missing label. - if label.as_str() == METRIC_NAME { - if let Some(column) = &metric_name_column { - exprs.push(DfExpr::Column(column.clone())); - if update_ctx { - // record the resolved physical name, not the semantic label - self.ctx.tag_columns.push(column.name.clone()); - } - } - continue; - } - // nonexistence label will be ignored - if let Some(column_name) = Self::find_case_sensitive_column(input_schema, label) - { - exprs.push(DfExpr::Column(Column::from_name(column_name.clone()))); - - if update_ctx { - // update the tag columns in context - self.ctx.tag_columns.push(column_name); - } + }; + exprs.push(DfExpr::Column(column.clone())); + if update_ctx { + // record the resolved physical name, not the semantic label + self.ctx.tag_columns.push(column.name); } } // add timestamp column @@ -4487,12 +4558,8 @@ impl PromPlanner { } "label_replace" => { self.ctx.use_tsid = false; - if let Some((replace_expr, dst_label, overwrite)) = self - .build_regexp_replace_label_expr( - &mut other_input_exprs, - query_engine_state, - input_schema, - )? + if let Some((replace_expr, dst_label)) = + self.build_regexp_replace_label_expr(&mut other_input_exprs, input_schema)? { // Reserve the current field columns except the `dst_label`. for value in &self.ctx.field_columns { @@ -4502,21 +4569,11 @@ impl PromPlanner { } } - if overwrite { - // The replacement overwrites a label the input already projects in - // place, so it must not be projected a second time under the same name. - self.ctx.tag_columns.retain(|tag| *tag != dst_label); - } else { - // Rewriting a label the input series already have can map several of - // them onto the same label set, which PromQL rejects. A new label - // keeps the series distinct. - ensure!( - !self.ctx.tag_columns.contains(&dst_label), - InvalidDestinationLabelNameSnafu { - label_name: dst_label - } - ); - } + // The replacement overwrites a label the input already projects in place, so + // it must not be projected a second time under the same name. Rewriting a + // label the series already carry can map several of them onto one label set, + // which is refused while the plan runs; a new label keeps them distinct. + self.ctx.tag_columns.retain(|tag| *tag != dst_label); new_tags.push(dst_label); // Add the new label expr to evaluate exprs.push(replace_expr); @@ -4823,16 +4880,21 @@ impl PromPlanner { /// Build expr for `label_replace` function /// - /// Returns the replacement expression, the physical name of the destination column and - /// whether that destination overwrites a column the input already projects: the marked - /// identity of the input, which is replaced in place instead of being refused when it - /// already exists, or an existing ordinary label the semantic name is replaced into. + /// Returns the expression of the destination and the physical name of the column it writes - + /// the marked identity for the semantic `__name__`, the label itself otherwise - or `None` + /// when the replacement leaves every series of the input untouched and the destination is not + /// projected at all. + /// + /// Like Prometheus, the anchored regex is matched against every series, reading the source + /// label the way [`Self::resolve_label_col`] resolves it, and a series whose source does not + /// match keeps the destination it already had. A missing source label reads as the empty + /// string for every series, so its whole result is decided here instead of per row. + /// doc: https://prometheus.io/docs/prometheus/latest/querying/functions/#label_replace fn build_regexp_replace_label_expr( &self, other_input_exprs: &mut VecDeque, - query_engine_state: &QueryEngineState, input_schema: &DFSchemaRef, - ) -> Result> { + ) -> Result> { // label_replace(vector, dst_label, replacement, src_label, regex) let dst_label = match other_input_exprs.pop_front() { Some(DfExpr::Literal(ScalarValue::Utf8(Some(d)), _)) => d, @@ -4872,270 +4934,99 @@ impl PromPlanner { .fail()?, }; - // Like Prometheus, match the whole source value. A series whose source value matches gets - // `dst_label` set to the expanded replacement; any other series is left unchanged. - // doc: https://prometheus.io/docs/prometheus/latest/querying/functions/#label_replace + // Like Prometheus, match the whole source value. let anchored = format!("^(?s:{regex})$"); - regex::Regex::new(&anchored).map_err(|_| { - InvalidRegularExpressionSnafu { - regex: regex.clone(), - } - .build() - })?; - let dst_exists = self.ctx.tag_columns.contains(&dst_label); + let compiled = regex::Regex::new(&anchored) + .map_err(|_| InvalidRegularExpressionSnafu { regex }.build())?; - // `__name__` is not a physical column: the semantic source resolves through the marked - // identity column of the input, never through a physical column of that name. - let metric_name_column = Self::metric_name_column(input_schema)?; - let metric_name_src = src_label.as_str() == METRIC_NAME; - let src_column = if metric_name_src { - metric_name_column.clone() - } else if self.ctx.tag_columns.contains(&src_label) { - Some(Column::from_name(src_label.clone())) + // The destination the input already carries, read back on the rows the regex leaves + // unchanged. The semantic one is the marked identity of the input; an ordinary one is a + // label the input projects, and a destination it does not carry stays NULL, like any + // label a series never had. + let existing_dst = if metric_name_dst { + Self::metric_name_column(input_schema)?.map(DfExpr::Column) + } else if self.ctx.tag_columns.contains(&dst_label) { + Some( + DfExpr::Column(Column::from_name(&dst_label)) + .cast_to(&ArrowDataType::Utf8, input_schema) + .context(DataFusionPlanningSnafu)?, + ) } else { None }; - let session_state = query_engine_state.session_state(); - - if !metric_name_src && !metric_name_dst { - // An ordinary source and an ordinary destination keep the existing handling. - // - // If the src_label exists and regex is empty, keep everything unchanged. - if src_column.is_some() && regex.is_empty() { - return Ok(None); + let value = match Self::resolve_label_col(input_schema, &src_label)? { + Some(src_column) => { + let src = Self::label_value_expr(&src_column.name, input_schema)?; + let matched = DfExpr::ScalarFunction(ScalarFunction { + func: datafusion_functions::regex::regexp_like(), + args: vec![src.clone(), lit(anchored.clone())], + }); + let replaced = Self::empty_label_to_null(DfExpr::ScalarFunction(ScalarFunction { + func: datafusion_functions::regex::regexp_replace(), + args: vec![src, lit(anchored), lit(replacement)], + })); + let unchanged = existing_dst + .clone() + .unwrap_or_else(|| lit(ScalarValue::Utf8(None))); + when(matched, replaced) + .otherwise(unchanged) + .context(DataFusionPlanningSnafu)? } - - // If the src_label doesn't exists, and - let Some(src_column) = &src_column else { - if replacement.is_empty() { - // the replacement is empty, keep everything unchanged. - return Ok(None); - } else { - // the replacement is not empty, always adds dst_label with replacement value. - return Ok(Some(( - // alias literal `replacement` as dst_label - lit(replacement).alias(&dst_label), - dst_label, - false, - ))); + None => { + // A missing source label is the empty string every series matches against, so the + // destination is the same for all of them and only the two outcomes of the regex + // remain. + match compiled.captures("").map(|captures| { + let mut value = String::new(); + captures.expand(&replacement, &mut value); + value + }) { + // The regex does not match the empty source: no series is touched. A + // destination the input does not carry stays unprojected. + None => match existing_dst { + Some(existing) => existing, + None if metric_name_dst => lit(ScalarValue::Utf8(None)), + None => return Ok(None), + }, + // An empty expansion is an absent label: it removes a destination the input + // carries, while one it does not carry is never projected for it. + Some(value) if value.is_empty() => match existing_dst { + Some(_) => lit(ScalarValue::Utf8(None)), + None if metric_name_dst => lit(ScalarValue::Utf8(None)), + None => return Ok(None), + }, + Some(value) => lit(value), } - }; - - // Preprocess the regex: - // https://github.com/prometheus/prometheus/blob/d902abc50d6652ba8fe9a81ff8e5cce936114eba/promql/functions.go#L1575C32-L1575C37 - let regex = format!("^(?s:{regex})$"); - let func = session_state - .scalar_functions() - .get("regexp_replace") - .context(UnsupportedExprSnafu { - name: "regexp_replace", - })?; - - // regexp_replace(src_label, regex, replacement) - let args = vec![ - DfExpr::Column(src_column.clone()), - DfExpr::Literal(ScalarValue::Utf8(Some(regex)), None), - DfExpr::Literal(ScalarValue::Utf8(Some(replacement)), None), - ]; - return Ok(Some(( - DfExpr::ScalarFunction(ScalarFunction { - func: func.clone(), - args, - }) - .alias(&dst_label), - dst_label, - false, - ))); - } - - let src = match &src_column { - Some(column) => DfExpr::ScalarFunction(ScalarFunction { - func: coalesce(), - args: vec![ - DfExpr::Column(column.clone()) - .cast_to(&ArrowDataType::Utf8, input_schema) - .context(DataFusionPlanningSnafu)?, - lit(""), - ], - }), - None => lit(""), - }; - let matched = DfExpr::ScalarFunction(ScalarFunction { - func: datafusion_functions::regex::regexp_like(), - args: vec![src.clone(), lit(anchored.clone())], - }); - let replaced = Self::empty_label_to_null(DfExpr::ScalarFunction(ScalarFunction { - func: datafusion_functions::regex::regexp_replace(), - args: vec![src, lit(anchored), lit(replacement)], - })); - let unchanged = if metric_name_dst { - match &metric_name_column { - Some(column) => DfExpr::Column(column.clone()), - None => lit(ScalarValue::Utf8(None)), } - } else if dst_exists { - DfExpr::Column(Column::from_name(&dst_label)) - .cast_to(&ArrowDataType::Utf8, input_schema) - .context(DataFusionPlanningSnafu)? - } else { - lit(ScalarValue::Utf8(None)) }; - let replace_expr = when(matched, replaced) - .otherwise(unchanged) - .context(DataFusionPlanningSnafu)?; - let overwrite = metric_name_dst || dst_exists; - if !metric_name_dst { - return Ok(Some((replace_expr.alias(&dst_label), dst_label, overwrite))); + let (replace_expr, dst_name) = + Self::write_destination_label(&self.ctx, input_schema, &dst_label, value)?; + Ok(Some((replace_expr, dst_name))) + } + + /// Wrap the rewritten label value `value` into the destination `dst_label` of the result. + /// + /// An ordinary destination is projected under its own name, while the semantic `__name__` + /// overwrites the marked identity of the input in place, or attaches one when the input + /// carries none - named the way a selector that attaches an identity names it, so that a + /// physical column is never mistaken for it. + /// + /// Returns the expression and the physical name of the destination column. + fn write_destination_label( + ctx: &PromPlannerContext, + input_schema: &DFSchemaRef, + dst_label: &str, + value: DfExpr, + ) -> Result<(DfExpr, String)> { + if dst_label != METRIC_NAME { + return Ok((value.alias(dst_label), dst_label.to_string())); } // An input without an identity gets a free internal name, chosen the way a selector that // attaches an identity chooses it, so a physical column is never mistaken for it. - let (dst_name, dst_qualifier) = match &metric_name_column { - Some(column) => (column.name.clone(), column.relation.clone()), - None => { - let occupied = input_schema - .fields() - .iter() - .map(|field| field.name().as_str()) - .collect::>(); - let mut name = PROMQL_METRIC_NAME_COLUMN.to_string(); - while occupied.contains(name.as_str()) { - name.push('_'); - } - // Project the identity like the tag columns it is projected next to. - let qualifier = self - .ctx - .time_index_column - .as_deref() - .and_then(|time_index| { - input_schema - .qualified_field_with_unqualified_name(time_index) - .ok() - }) - .and_then(|(qualifier, _)| qualifier.cloned()); - (name, qualifier) - } - }; - let mut metadata = std::collections::HashMap::new(); - metadata.insert( - PROMQL_FIELD_ROLE_KEY.to_string(), - PROMQL_METRIC_NAME_ROLE.to_string(), - ); - - Ok(Some(( - DfExpr::Alias(Alias { - expr: Box::new(replace_expr), - relation: dst_qualifier, - name: dst_name.clone(), - metadata: Some(metadata.into()), - }), - dst_name, - overwrite, - ))) - } - - /// Build expr for `label_join` function - /// - /// Returns the concatenation expression and the physical name of the destination column, - /// which is the marked identity itself when the destination is the semantic `__name__`. - fn build_concat_labels_expr( - other_input_exprs: &mut VecDeque, - ctx: &PromPlannerContext, - query_engine_state: &QueryEngineState, - input_schema: &DFSchemaRef, - ) -> Result<(DfExpr, String)> { - // label_join(vector, dst_label, separator, src_label_1, src_label_2, ...) - - let dst_label = match other_input_exprs.pop_front() { - Some(DfExpr::Literal(ScalarValue::Utf8(Some(d)), _)) => d, - other => UnexpectedPlanExprSnafu { - desc: format!("expected dst_label string literal, but found {:?}", other), - } - .fail()?, - }; - let separator = match other_input_exprs.pop_front() { - Some(DfExpr::Literal(ScalarValue::Utf8(Some(d)), _)) => d, - other => UnexpectedPlanExprSnafu { - desc: format!("expected separator string literal, but found {:?}", other), - } - .fail()?, - }; - - // Create a set of available columns (tag columns + field columns + time index column) - let available_columns: HashSet<&str> = ctx - .tag_columns - .iter() - .chain(ctx.field_columns.iter()) - .chain(ctx.time_index_column.as_ref()) - .map(|s| s.as_str()) - .collect(); - - // `__name__` is not a physical column: the semantic source resolves through the marked - // identity column of the input, never through a physical column of that name. - let metric_name_column = Self::metric_name_column(input_schema)?; - let src_labels = other_input_exprs - .iter() - .map(|expr| { - // `concat_ws` skips NULL arguments together with their separator, while an - // absent label joins as the empty string. `__name__` resolves through the marked - // identity column, never through a physical column of that name. - match expr { - DfExpr::Literal(ScalarValue::Utf8(Some(label)), None) => { - if label.is_empty() { - FunctionInvalidArgumentSnafu { - fn_name: "label_join", - } - .fail() - } else if label.as_str() == METRIC_NAME { - match &metric_name_column { - Some(column) => Self::label_value_expr(&column.name, input_schema), - None => Ok(lit("")), - } - } else if available_columns.contains(label.as_str()) { - Self::label_value_expr(label, input_schema) - } else { - Ok(lit("")) - } - } - other => UnexpectedPlanExprSnafu { - desc: format!( - "expected source label string literal, but found {:?}", - other - ), - } - .fail(), - } - }) - .collect::>>()?; - // Joining no labels yields the empty string, i.e. removes `dst_label`. - if src_labels.is_empty() { - return Ok((lit(ScalarValue::Utf8(None)).alias(&dst_label), dst_label)); - } - - let session_state = query_engine_state.session_state(); - let func = session_state - .scalar_functions() - .get("concat_ws") - .context(UnsupportedExprSnafu { name: "concat_ws" })?; - - // concat_ws(separator, src_label_1, src_label_2, ...) as dst_label - let mut args = Vec::with_capacity(1 + src_labels.len()); - args.push(DfExpr::Literal(ScalarValue::Utf8(Some(separator)), None)); - args.extend(src_labels); - - let concat_expr = DfExpr::ScalarFunction(ScalarFunction { - func: func.clone(), - args, - }); - if dst_label.as_str() != METRIC_NAME { - return Ok((concat_expr.alias(&dst_label), dst_label)); - } - - // The semantic destination overwrites the marked identity of the input in place, or - // attaches a new one when the input carries none. - let (dst_name, dst_qualifier) = match metric_name_column { + let (dst_name, dst_qualifier) = match Self::metric_name_column(input_schema)? { Some(column) => (column.name, column.relation), None => { let occupied = input_schema @@ -5168,7 +5059,7 @@ impl PromPlanner { Ok(( DfExpr::Alias(Alias { - expr: Box::new(Self::empty_label_to_null(concat_expr)), + expr: Box::new(value), relation: dst_qualifier, name: dst_name.clone(), metadata: Some(metadata.into()), @@ -5177,6 +5068,86 @@ impl PromPlanner { )) } + /// Build expr for `label_join` function + /// + /// Returns the joined expression and the physical name of the destination column, which is + /// the marked identity itself when the destination is the semantic `__name__`. + fn build_concat_labels_expr( + other_input_exprs: &mut VecDeque, + ctx: &PromPlannerContext, + query_engine_state: &QueryEngineState, + input_schema: &DFSchemaRef, + ) -> Result<(DfExpr, String)> { + // label_join(vector, dst_label, separator, src_label_1, src_label_2, ...) + + let dst_label = match other_input_exprs.pop_front() { + Some(DfExpr::Literal(ScalarValue::Utf8(Some(d)), _)) => d, + other => UnexpectedPlanExprSnafu { + desc: format!("expected dst_label string literal, but found {:?}", other), + } + .fail()?, + }; + let separator = match other_input_exprs.pop_front() { + Some(DfExpr::Literal(ScalarValue::Utf8(Some(d)), _)) => d, + other => UnexpectedPlanExprSnafu { + desc: format!("expected separator string literal, but found {:?}", other), + } + .fail()?, + }; + + let src_labels = other_input_exprs + .iter() + .map(|expr| match expr { + // `concat_ws` skips NULL arguments together with their separator, while an absent + // label joins as the empty string; a source resolves like every other label read, + // `__name__` through the marked identity of the input included. + DfExpr::Literal(ScalarValue::Utf8(Some(label)), None) => { + if label.is_empty() { + return FunctionInvalidArgumentSnafu { + fn_name: "label_join", + } + .fail(); + } + match Self::resolve_label_col(input_schema, label)? { + Some(column) => Self::label_value_expr(&column.name, input_schema), + None => Ok(lit("")), + } + } + other => UnexpectedPlanExprSnafu { + desc: format!( + "expected source label string literal, but found {:?}", + other + ), + } + .fail(), + }) + .collect::>>()?; + + // Joining no labels yields the empty string, i.e. the destination label is absent. + let value = if src_labels.is_empty() { + lit(ScalarValue::Utf8(None)) + } else { + let session_state = query_engine_state.session_state(); + let func = session_state + .scalar_functions() + .get("concat_ws") + .context(UnsupportedExprSnafu { name: "concat_ws" })?; + + // concat_ws(separator, src_label_1, src_label_2, ...) as dst_label + let mut args = Vec::with_capacity(1 + src_labels.len()); + args.push(DfExpr::Literal(ScalarValue::Utf8(Some(separator)), None)); + args.extend(src_labels); + + // An empty joined label is an absent label, like the sources that never had one. + Self::empty_label_to_null(DfExpr::ScalarFunction(ScalarFunction { + func: func.clone(), + args, + })) + }; + + Self::write_destination_label(ctx, input_schema, &dst_label, value) + } + /// The value of `label` as a string, where NULL (the series has no such label) reads as the /// empty string, as in PromQL. fn label_value_expr(label: &str, input_schema: &DFSchemaRef) -> Result { @@ -5316,26 +5287,17 @@ impl PromPlanner { FunctionInvalidArgumentSnafu { fn_name: func } ); - let metric_name_column = Self::metric_name_column(input_schema)?; tags.iter() .map(|col| match col { DfExpr::Literal(ScalarValue::Utf8(Some(label)), _) => { - // `__name__` is not a physical column: the metric identity lives in the - // marked column of the input. Without one, the label keeps its ordinary - // missing-label behavior. - let column = match (label.as_str(), &metric_name_column) { - (METRIC_NAME, Some(metric_name_column)) => metric_name_column.clone(), - (METRIC_NAME, None) => { - return Err(datafusion::common::field_not_found( - None::, - label, - input_schema.as_ref(), - )) - .context(DataFusionPlanningSnafu); - } - _ => Column::from_name(label), + // PromQL sorts by the value of the label of a series, where a label the series + // does not carry - the semantic `__name__` of an input that dropped its + // identity included - is the empty value, i.e. NULL among NULLs. + let expr = match Self::resolve_label_col(input_schema, label)? { + Some(column) => DfExpr::Column(column), + None => lit(ScalarValue::Utf8(None)), }; - Ok(DfExpr::Column(column).sort(asc, false)) + Ok(expr.sort(asc, false)) } other => UnexpectedPlanExprSnafu { desc: format!("expected label string literal, but found {:?}", other), diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index d6c6b03f862..4d61b867969 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -7040,7 +7040,7 @@ async fn test_label_join() { let expected = r#" Filter: up.field_0 IS NOT NULL [timestamp:Timestamp(ms), field_0:Float64;N, foo:Utf8;N, tag_0:Utf8, tag_1:Utf8, tag_2:Utf8, tag_3:Utf8, __promql_metric_name:Utf8] - Projection: up.timestamp, up.field_0, concat_ws(Utf8(","), coalesce(up.tag_1, Utf8("")), coalesce(up.tag_2, Utf8("")), coalesce(up.tag_3, Utf8(""))) AS foo, up.tag_0, up.tag_1, up.tag_2, up.tag_3, __promql_metric_name [timestamp:Timestamp(ms), field_0:Float64;N, foo:Utf8;N, tag_0:Utf8, tag_1:Utf8, tag_2:Utf8, tag_3:Utf8, __promql_metric_name:Utf8] + Projection: up.timestamp, up.field_0, nullif(concat_ws(Utf8(","), coalesce(up.tag_1, Utf8("")), coalesce(up.tag_2, Utf8("")), coalesce(up.tag_3, Utf8(""))), Utf8("")) AS foo, up.tag_0, up.tag_1, up.tag_2, up.tag_3, __promql_metric_name [timestamp:Timestamp(ms), field_0:Float64;N, foo:Utf8;N, tag_0:Utf8, tag_1:Utf8, tag_2:Utf8, tag_3:Utf8, __promql_metric_name:Utf8] Projection: up.tag_0, up.tag_1, up.tag_2, up.tag_3, up.timestamp, up.field_0, Utf8("up") AS __promql_metric_name [tag_0:Utf8, tag_1:Utf8, tag_2:Utf8, tag_3:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, __promql_metric_name:Utf8] PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag_0:Utf8, tag_1:Utf8, tag_2:Utf8, tag_3:Utf8, timestamp:Timestamp(ms), field_0:Float64;N] PromSeriesDivide: tags=["tag_0", "tag_1", "tag_2", "tag_3"] [tag_0:Utf8, tag_1:Utf8, tag_2:Utf8, tag_3:Utf8, timestamp:Timestamp(ms), field_0:Float64;N] @@ -7076,7 +7076,7 @@ async fn test_label_replace() { let expected = r#" Filter: up.field_0 IS NOT NULL [timestamp:Timestamp(ms), field_0:Float64;N, foo:Utf8;N, tag_0:Utf8, __promql_metric_name:Utf8] - Projection: up.timestamp, up.field_0, regexp_replace(up.tag_0, Utf8("^(?s:(.*):.*)$"), Utf8("$1")) AS foo, up.tag_0, __promql_metric_name [timestamp:Timestamp(ms), field_0:Float64;N, foo:Utf8;N, tag_0:Utf8, __promql_metric_name:Utf8] + Projection: up.timestamp, up.field_0, CASE WHEN regexp_like(coalesce(up.tag_0, Utf8("")), Utf8("^(?s:(.*):.*)$")) THEN nullif(regexp_replace(coalesce(up.tag_0, Utf8("")), Utf8("^(?s:(.*):.*)$"), Utf8("$1")), Utf8("")) ELSE Utf8(NULL) END AS foo, up.tag_0, __promql_metric_name [timestamp:Timestamp(ms), field_0:Float64;N, foo:Utf8;N, tag_0:Utf8, __promql_metric_name:Utf8] Projection: up.tag_0, up.timestamp, up.field_0, Utf8("up") AS __promql_metric_name [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N, __promql_metric_name:Utf8] PromInstantManipulate: range=[0..100000000], lookback=[1000], interval=[5000], time index=[timestamp] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N] PromSeriesDivide: tags=["tag_0"] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N] @@ -10000,7 +10000,7 @@ async fn test_histogram_quantile_folds_with_identity_then_drops_it() { } #[tokio::test] -async fn test_sort_by_label_semantic_name_without_marker_keeps_missing_label_error() { +async fn test_sort_by_label_semantic_name_without_marker_sorts_as_absent() { let state = build_query_engine_state(); let provider = || async { build_test_table_provider_with_distinct_tags(&[("collision_metric", &["__name__", "job"])]) @@ -10024,18 +10024,26 @@ async fn test_sort_by_label_semantic_name_without_marker_keeps_missing_label_err .is_none() ); for function in ["sort_by_label", "sort_by_label_desc"] { + // The name is gone, so the label the sort reads is absent: it sorts as NULL among the + // labels the series carry instead of failing the query, and the physical `__name__` + // column of the input is never read as the semantic name. let query = format!(r#"{function}(abs(collision_metric), "__name__")"#); - let error = - PromPlanner::stmt_to_plan(provider().await, &operator_eval_stmt(&query), &state) - .await - .unwrap_err(); + let plan = PromPlanner::stmt_to_plan(provider().await, &operator_eval_stmt(&query), &state) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + let plan_text = plan.display_indent().to_string(); assert!( - matches!( - error, - crate::promql::error::Error::DataFusionPlanning { .. } - ), - "{error}" + plan_text.contains(&format!( + "Sort: Utf8(NULL) {} NULLS", + if function.ends_with("desc") { + "DESC" + } else { + "ASC" + } + )), + "{query}: {plan_text}" ); + let query = format!(r#"{function}(collision_metric, "__name__")"#); let plan = PromPlanner::stmt_to_plan(provider().await, &operator_eval_stmt(&query), &state) .await @@ -10917,6 +10925,60 @@ async fn test_name_only_vector_matching_does_not_broadcast() { } } +/// Dropping the metric name can leave several series of one evaluation step with the same label +/// set - `or on(__name__)` keeps both operands, which only the name told apart. A vector cannot +/// hold those rows, so the query is refused instead of the HTTP layer silently picking one of +/// them; a set of series the name did not distinguish still executes. +#[tokio::test] +async fn test_dropped_metric_name_rejects_duplicate_label_sets() { + let state = build_query_engine_state(); + // Both metrics carry the same ordinary label `tag="shared"` at `ts=1000` and differ only in + // their name, so `or on(__name__)` keeps both of them. + for query in [ + "abs(left_metric or on(__name__) right_metric)", + "-(left_metric or on(__name__) right_metric)", + ] { + let plan = PromPlanner::stmt_to_plan( + binary_metric_name_table_provider(), + &operator_eval_stmt(query), + &state, + ) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + assert!( + PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .is_none(), + "{query}: the operands of the function must have dropped their names" + ); + let err = execute_err(plan, &state).await; + assert!( + err.contains("vector cannot contain metrics with the same labelset"), + "{query}: expected the duplicate-label-set refusal, got {err}" + ); + } + + // Matching on the ordinary label alone selects one of the operands, so the result holds one + // series per label set - and dropping the name of an input that only ever had one series per + // label set is always fine. + for (query, expected) in [ + ("abs(left_metric or right_metric)", vec![2.0]), + ("abs(left_metric + right_metric)", vec![5.0]), + ] { + let plan = PromPlanner::stmt_to_plan( + binary_metric_name_table_provider(), + &operator_eval_stmt(query), + &state, + ) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + let sample_column = float_sample_column(&plan); + let (_, batches) = execute(plan, &state).await; + assert_metric_name_not_in_batches(&batches); + assert_eq!(values(&batches, &sample_column), expected, "{query}"); + } +} + /// `or` keeps every chosen row's own identity, including the typed NULL of an operand that /// dropped its name, and never invents an identity when neither operand has one. #[tokio::test] @@ -11240,6 +11302,168 @@ async fn test_label_functions_semantic_name_destination_without_input_identity() } } +/// `label_join` with no source labels joins nothing, which is the empty string: the semantic +/// destination has no name left and the destination resolves like every other one instead of +/// being written under a physical column of that name. +#[tokio::test] +async fn test_label_join_without_source_labels_removes_semantic_name() { + let state = build_query_engine_state(); + for query in [ + r#"label_join(cv_metric, "__name__", "")"#, + r#"label_join(cv_metric, "__name__", "-")"#, + ] { + let plan = PromPlanner::stmt_to_plan( + build_count_values_table_provider().await, + &operator_eval_stmt(query), + &state, + ) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + + // The identity the destination names is still the marked one, holdng the value that + // removes the name - an absent label - and no unmarked column takes its place. + let marker = PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .unwrap_or_else(|| panic!("{query}: the destination must be the marked identity")); + assert_eq!(marker.name, PROMQL_METRIC_NAME_COLUMN, "{query}"); + assert!( + plan.schema() + .field_with_unqualified_name(METRIC_NAME) + .is_err(), + "{query}: no physical column may be projected for the semantic name: {}", + plan.display_indent() + ); + + let sample_column = float_sample_column(&plan); + let (_, batches) = execute(plan, &state).await; + assert_metric_name_in_batches(&batches, &marker.name); + assert_eq!( + string_values(&batches, &marker.name), + vec![None, None, None], + "{query}: the joined name is absent" + ); + assert_eq!( + cv_rows(&batches, &sample_column), + vec![ + ("k0".to_string(), 1_000, 1.0), + ("k1".to_string(), 1_000, 2.0), + ("k2".to_string(), 1_000, 1.0), + ], + "{query}: the samples must be untouched" + ); + } +} + +/// The internal metric-name column and the metric engine's internal columns are physical +/// details, not labels: a `by(...)` label or a `label_join` source that names one is a label the +/// series do not carry, while a user column of that very name stays an ordinary label. +#[tokio::test] +async fn test_internal_metric_name_column_is_not_an_ordinary_label() { + let state = build_query_engine_state(); + + // `sum by (__promql_metric_name)` groups by a label no series carries, exactly like `sum`, + // instead of projecting the internal marker as a group key. + let query = "sum by (__promql_metric_name) (cv_metric)"; + for provider in [ + build_count_values_table_provider().await, + build_count_values_collision_table_provider(PROMQL_METRIC_NAME_COLUMN).await, + ] { + let plan = PromPlanner::stmt_to_plan(provider, &operator_eval_stmt(query), &state) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + assert!( + PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .is_none(), + "{query}: the internal marker must not be a group key: {}", + plan.display_indent() + ); + } + + // The group keys of the two inputs tell the two resolutions apart: the plain metric has no + // such label and aggregates into a single series, while the collision metric groups by its + // user column of that name. + for (collision, expected) in [(false, vec![4.0]), (true, vec![1.0, 1.0, 1.0])] { + let provider = if collision { + build_count_values_collision_table_provider(PROMQL_METRIC_NAME_COLUMN).await + } else { + build_count_values_table_provider().await + }; + let plan = PromPlanner::stmt_to_plan(provider, &operator_eval_stmt(query), &state) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + let sample_column = float_sample_column(&plan); + let (_, batches) = execute(plan, &state).await; + // The identity is gone in both cases: the plain metric has no such label to group by, + // and the collision metric groups by its unmarked user column, which is a label. + if !collision { + assert_metric_name_not_in_batches(&batches); + } else { + assert!( + batches.iter().all(|batch| batch + .schema() + .fields() + .iter() + .all(|field| field.metadata().get(PROMQL_FIELD_ROLE_KEY).is_none())), + "{query}: no field may carry the metric-name role" + ); + } + let mut actual = values(&batches, &sample_column); + actual.sort_by(f64::total_cmp); + assert_eq!(actual, expected, "{query}: collision={collision}"); + } + + // `label_join` reads the same label: the internal marker is absent - it joins as the empty + // string, i.e. the destination stays NULL - while a user column of that name is read. + let query = r#"label_join(cv_metric, "copied", "", "__promql_metric_name")"#; + for (collision, expected) in [ + (false, vec![None, None, None]), + ( + true, + vec![ + Some("k0".to_string()), + Some("k1".to_string()), + Some("k2".to_string()), + ], + ), + ] { + let provider = if collision { + build_count_values_collision_table_provider(PROMQL_METRIC_NAME_COLUMN).await + } else { + build_count_values_table_provider().await + }; + let plan = PromPlanner::stmt_to_plan(provider, &operator_eval_stmt(query), &state) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + + // `label_join` keeps the identity of its input, under whatever name that identity has. + let marker = PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .unwrap_or_else(|| panic!("{query}: the identity must survive")); + assert_eq!( + marker.name, + if collision { + format!("{PROMQL_METRIC_NAME_COLUMN}_") + } else { + PROMQL_METRIC_NAME_COLUMN.to_string() + }, + "{query}" + ); + + let (_, batches) = execute(plan, &state).await; + let mut copied = string_values(&batches, "copied"); + copied.sort(); + assert_eq!(copied, expected, "{query}: collision={collision}"); + assert_eq!( + string_values(&batches, &marker.name) + .iter() + .collect::>(), + vec![&Some("cv_metric".to_string()); 3], + "{query}" + ); + } +} + /// A physical label named `__name__` is an ordinary label, not the semantic metric name: the /// semantic source resolves through the marked identity instead, and the semantic destination /// overwrites the identity, while the physical label keeps its values and its unmarked metadata. @@ -11410,12 +11634,12 @@ async fn test_label_functions_semantic_name_destination_avoids_physical_collisio } /// An ordinary `label_replace`/`label_join` - neither source nor destination names `__name__` - -/// keeps the existing handling: the destination label receives `regexp_replace` output even -/// without a match, a missing component of a join joins as the empty string and keeps the -/// separator it joins with, the historical shortcuts stay in place, and an already present -/// destination is still refused. +/// follows the Prometheus semantics of the rebased planner: the destination is the expanded +/// replacement when the anchored regex matches the source, a series whose source does not match +/// keeps the destination it already had, an empty expansion removes it, and a source the input +/// does not carry is the empty string the regex is matched against. #[tokio::test] -async fn test_label_functions_ordinary_only_keeps_existing_behavior() { +async fn test_label_functions_ordinary_only_follows_prometheus_semantics() { let state = build_query_engine_state(); // (query, destination column, expected `(series tag, destination value)` rows) for (query, column, expected) in [ @@ -11429,19 +11653,23 @@ async fn test_label_functions_ordinary_only_keeps_existing_behavior() { Some("renamed_2".to_string()), ], ), - // The anchored regex matches nothing, so the destination keeps the source it replaced. + // The anchored regex matches nothing, so the destination the input does not carry is NULL + // instead of the source value the replacement left unchanged. ( r#"label_replace(cv_metric, "copy", "renamed", "k", "nomatch")"#, "copy", - vec![ - Some("k0".to_string()), - Some("k1".to_string()), - Some("k2".to_string()), - ], + vec![None, None, None], ), - // A missing source with a non-empty replacement keeps the existing shortcut. + // An empty regex matches the empty source only, which no series of the input carries. ( - r#"label_replace(cv_metric, "copy", "addressed", "missing_label", "nomatch")"#, + r#"label_replace(cv_metric, "copy", "renamed", "k", "")"#, + "copy", + vec![None, None, None], + ), + // A missing source label is the empty string, and a regex matching it writes its + // replacement on every series. + ( + r#"label_replace(cv_metric, "copy", "addressed", "missing_label", "()")"#, "copy", vec![ Some("addressed".to_string()), @@ -11481,8 +11709,9 @@ async fn test_label_functions_ordinary_only_keeps_existing_behavior() { ); } - // A present source and an empty regex still add no destination label at all. - let query = r#"label_replace(cv_metric, "copy", "renamed", "k", "")"#; + // A destination the input already carries is overwritten on a match - never refused as an + // invalid destination - and a series the regex does not match keeps its own value. + let query = r#"label_replace(cv_metric, "k", "fixed", "k", "k1")"#; let plan = PromPlanner::stmt_to_plan( build_count_values_table_provider().await, &operator_eval_stmt(query), @@ -11490,36 +11719,70 @@ async fn test_label_functions_ordinary_only_keeps_existing_behavior() { ) .await .unwrap_or_else(|err| panic!("{query}: {err}")); - assert!( - plan.schema().field_with_unqualified_name("copy").is_err(), - "{query}: {}", - plan.display_indent() - ); - let sample_column = float_sample_column(&plan); let (_, batches) = execute(plan, &state).await; + let mut labels = string_values(&batches, "k"); + labels.sort(); assert_eq!( - cv_rows(&batches, &sample_column), + labels, vec![ - ("k0".to_string(), 1_000, 1.0), - ("k1".to_string(), 1_000, 2.0), - ("k2".to_string(), 1_000, 1.0), + Some("fixed".to_string()), + Some("k0".to_string()), + Some("k2".to_string()), ], "{query}" ); - // An already present destination is still refused on the ordinary path. - let query = r#"label_replace(cv_metric, "k", "x", "k", "(.*)")"#; - let err = PromPlanner::stmt_to_plan( + // The empty source either does not match the regex or expands to nothing, so the input keeps + // every series and gets no destination at all. + for query in [ + r#"label_replace(cv_metric, "x", "new", "missing_label", ".+")"#, + r#"label_replace(cv_metric, "x", "$1", "missing_label", "(.*)")"#, + r#"label_replace(cv_metric, "copy", "addressed", "missing_label", "nomatch")"#, + r#"label_replace(cv_metric, "copy", "", "missing_label", "nomatch")"#, + ] { + let plan = PromPlanner::stmt_to_plan( + build_count_values_table_provider().await, + &operator_eval_stmt(query), + &state, + ) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + let destination = if query.contains("\"x\"") { "x" } else { "copy" }; + assert!( + plan.schema() + .field_with_unqualified_name(destination) + .is_err(), + "{query}: {}", + plan.display_indent() + ); + + let sample_column = float_sample_column(&plan); + let (_, batches) = execute(plan, &state).await; + assert_eq!( + cv_rows(&batches, &sample_column), + vec![ + ("k0".to_string(), 1_000, 1.0), + ("k1".to_string(), 1_000, 2.0), + ("k2".to_string(), 1_000, 1.0), + ], + "{query}" + ); + } + + // Overwriting the only label of three series with one constant collapses them onto a single + // label set, which is refused while the plan runs. + let query = r#"label_replace(cv_metric, "k", "x", "k", ".+")"#; + let plan = PromPlanner::stmt_to_plan( build_count_values_table_provider().await, &operator_eval_stmt(query), &state, ) .await - .expect_err(query) - .to_string(); + .unwrap_or_else(|err| panic!("{query}: {err}")); + let err = execute_err(plan, &state).await; assert!( - err.contains("Invalid destination label name in label_replace(): k"), - "{query}: expected the existing-destination refusal, got {err}" + err.contains("vector cannot contain metrics with the same labelset"), + "{query}: expected the duplicate-label-set refusal, got {err}" ); } @@ -11811,39 +12074,21 @@ async fn test_label_replace_semantic_source_into_existing_ordinary_label() { } } -/// The ordinary `label_replace` destination keeps its historical value when the anchored regex -/// does not match: the un-replaced source label, not a NULL. This is the pre-existing behavior -/// of the function on its ordinary path. +/// The ordinary `label_replace` destination keeps the value of its series when the anchored regex +/// does not match: the series stays unchanged, which is why a destination the input already +/// carries is overwritten on a match instead of being refused. #[tokio::test] -async fn test_label_replace_ordinary_destination_keeps_existing_value() { +async fn test_label_replace_ordinary_destination_keeps_its_value() { let state = build_query_engine_state(); - // (query, expected `(series tag, destination value)` rows) + // (query, expected values of `k`, one per series) for (query, expected) in [ ( - r#"label_replace(cv_metric, "copy", "renamed_$1", "k", "k(.)")"#, - vec![ - Some("renamed_0".to_string()), - Some("renamed_1".to_string()), - Some("renamed_2".to_string()), - ], + r#"label_replace(cv_metric, "k", "renamed_$1", "k", "k(.)")"#, + vec!["renamed_0", "renamed_1", "renamed_2"], ), - // No match: the destination holds the source value the replacement did not change. ( - r#"label_replace(cv_metric, "copy", "renamed", "k", "nomatch")"#, - vec![ - Some("k0".to_string()), - Some("k1".to_string()), - Some("k2".to_string()), - ], - ), - // A source the input does not carry keeps the historical shortcut. - ( - r#"label_replace(cv_metric, "copy", "addressed", "missing_label", "nomatch")"#, - vec![ - Some("addressed".to_string()), - Some("addressed".to_string()), - Some("addressed".to_string()), - ], + r#"label_replace(cv_metric, "k", "renamed", "k", "nomatch")"#, + vec!["k0", "k1", "k2"], ), ] { let plan = PromPlanner::stmt_to_plan( @@ -11855,41 +12100,15 @@ async fn test_label_replace_ordinary_destination_keeps_existing_value() { .unwrap_or_else(|err| panic!("{query}: {err}")); let (_, batches) = execute(plan, &state).await; + let mut labels = string_values(&batches, "k"); + labels.sort(); assert_eq!( - tagged_string_rows(&batches, "copy"), + labels, expected .iter() - .enumerate() - .map(|(index, value)| (format!("k{index}"), value.clone())) + .map(|value| Some((*value).to_string())) .collect::>(), "{query}" ); } - - // An empty replacement with a missing source keeps everything unchanged, so no destination - // label exists at all and the samples stay untouched. - let query = r#"label_replace(cv_metric, "copy", "", "missing_label", "nomatch")"#; - let plan = PromPlanner::stmt_to_plan( - build_count_values_table_provider().await, - &operator_eval_stmt(query), - &state, - ) - .await - .unwrap_or_else(|err| panic!("{query}: {err}")); - assert!( - plan.schema().field_with_unqualified_name("copy").is_err(), - "{query}: {}", - plan.display_indent() - ); - let sample_column = float_sample_column(&plan); - let (_, batches) = execute(plan, &state).await; - assert_eq!( - cv_rows(&batches, &sample_column), - vec![ - ("k0".to_string(), 1_000, 1.0), - ("k1".to_string(), 1_000, 2.0), - ("k2".to_string(), 1_000, 1.0), - ], - "{query}" - ); }