rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-13 15:32:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(promql): unify semantic label resolution and label rewrites

Browse Source 
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>
This commit is contained in:
discord9
2026-10-12 11:13:07 +08:00
parent e86a22733c
commit 6106a6ea08
2 changed files with 602 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/query/src/promql/planner.rs
+282 -320
View File
@@ -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<LogicalPlan> {
/// 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<LogicalPlan> {
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::<Vec<_>>();
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::<UnionDistinctOn>()
}
_ => false,
};
Ok(if unioned {
TreeNodeRecursion::Stop
} else {
TreeNodeRecursion::Continue
})
});
unioned
}
fn negate_field_columns(&mut self, input: LogicalPlan) -> Result<LogicalPlan> {
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<Option<Column>> {
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<DfExpr>,
query_engine_state: &QueryEngineState,
input_schema: &DFSchemaRef,
) -> Result<Option<(DfExpr, String, bool)>> {
) -> Result<Option<(DfExpr, String)>> {
// 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::<HashSet<_>>();
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<DfExpr>,
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::<Result<Vec<_>>>()?;
// 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<DfExpr>,
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::<Result<Vec<_>>>()?;
// 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<DfExpr> {
@@ -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::<datafusion::common::TableReference>,
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),
src/query/src/promql/planner/test.rs
+320 -101
View File
@@ -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<_>>(),
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::<Vec<_>>(),
"{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}"
);
}