From 1817d0bb7675064b7d17bb2525320699403efa1b Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Wed, 30 Sep 2026 15:56:52 +0800 Subject: [PATCH] fix(promql): close metric-name identity privilege and boundary gaps - count_values never forges the identity: the label it generates is always an ordinary column named as asked, with no metric-name role metadata and no reuse of the marked column's physical name. A plain aggregate argument can no longer inject a sample value into the privileged identity column (which the HTTP layer would otherwise emit as __name__ and TQL would silently strip). With no marker forged, dropping the name of a count_values result cannot collide, so the union-subtree check remains the only duplicate-label-set source. - binary_match_label_mask no longer lets an ordinary on()/ignoring() label hit the marked identity by its physical name; only the semantic __name__ resolves to it, and a real user column sharing the name still matches. - label_join/label_replace move the marked identity to a free internal name before writing an ordinary destination that occupies its physical name, so writing a label can no longer silently replace the identity. - sort_by_label reads a missing or NULL string label as the empty value, so a series that lost its name sorts first ascending like Prometheus, instead of NULLS LAST. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/query/src/promql/planner.rs | 200 ++++--- src/query/src/promql/planner/test.rs | 752 ++++++++++++++++++++++----- 2 files changed, 751 insertions(+), 201 deletions(-) diff --git a/src/query/src/promql/planner.rs b/src/query/src/promql/planner.rs index 27ac010a545..3dec667e0d4 100644 --- a/src/query/src/promql/planner.rs +++ b/src/query/src/promql/planner.rs @@ -821,47 +821,31 @@ impl PromPlanner { // label only. Dropping it from the projected tag columns is required as well: // projecting both would emit two columns with the same name (rejected as an // ambiguous reference). - let metric_name = label == METRIC_NAME; - // The semantic `__name__` replaces the marked identity further below, so only - // an ordinary input label is overwritten here. - if !metric_name { - self.ctx.tag_columns.retain(|tag| tag != label); - group_exprs.retain( - |expr| !matches!(expr, DfExpr::Column(column) if column.name == label), - ); - } - // The generated column reuses the name of the identity it replaces, or a free - // internal name when the input carries no identity. - let label_column = if metric_name { - match Self::metric_name_column(builder.schema())? { - Some(column) => column.name, - None => { - let occupied = builder - .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('_'); - } - name - } + // + // The generated label is always an ordinary label, whatever it is named: the + // marked metric-name identity is metadata the planner attaches where it knows + // a real metric name is written, so a plain aggregate argument must not be + // able to forge it from a sample value. The semantic `__name__` is therefore + // an ordinary column of the result, spelled like the label the user asked + // for. + let label_column = label.to_string(); + self.ctx.tag_columns.retain(|tag| tag != &label_column); + group_exprs.retain( + |expr| !matches!(expr, DfExpr::Column(column) if column.name == label_column), + ); + if label == METRIC_NAME { + // The destination names the semantic `__name__`, which the input carries + // as its marked identity: the generated label replaces it, so the + // identity must not stay grouped or projected next to it. + if let Some(marker) = Self::metric_name_column(builder.schema())? { + self.ctx + .aggregation_field_labels + .retain(|tag| tag != &marker.name); + self.ctx.tag_columns.retain(|tag| tag != &marker.name); + group_exprs.retain(|expr| { + !matches!(expr, DfExpr::Column(column) if column.name == marker.name) + }); } - } else { - label.to_string() - }; - if metric_name { - // The overwritten identity must not stay grouped or projected: the - // generated label is the only marked column. - self.ctx - .aggregation_field_labels - .retain(|tag| tag != &label_column); - self.ctx.tag_columns.retain(|tag| tag != &label_column); - group_exprs.retain( - |expr| !matches!(expr, DfExpr::Column(column) if column.name == label_column), - ); } // The tag columns projected below are unqualified `Column` references, so they // end up qualified with whatever qualifier they carry in the input plan. Give @@ -915,21 +899,14 @@ impl PromPlanner { expr.schema_name().to_string(), )), }; - // The generated name carries the identity role, like the selector - // marker it replaces. - let metadata = metric_name.then(|| { - let mut metadata = std::collections::HashMap::new(); - metadata.insert( - PROMQL_FIELD_ROLE_KEY.to_string(), - PROMQL_METRIC_NAME_ROLE.to_string(), - ); - metadata.into() - }); + // The generated label is an ordinary label of the result: the identity + // role is never inferred from a name, only attached where a real + // metric name is written. DfExpr::Alias(Alias { expr: Box::new(value), relation: label_qualifier.clone(), name: label_column.clone(), - metadata, + metadata: None, }) }) .collect::>(); @@ -2271,16 +2248,13 @@ impl PromPlanner { desc: "metric-name identity already attached" } ); - 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('_'); - } + let name = Self::free_metric_name_column_name( + input + .schema() + .fields() + .iter() + .map(|field| field.name().as_str()), + ); let mut exprs = input .schema() .iter() @@ -2325,6 +2299,19 @@ impl PromPlanner { }) } + /// The first name free among `occupied` for the marked metric-name identity: the internal + /// name, with `_` appended until no listed column carries it. A free name keeps the identity + /// out of the way of every real column, so a physical name spelled like the internal one is + /// never the identity. + fn free_metric_name_column_name<'a>(occupied: impl IntoIterator) -> String { + let occupied = occupied.into_iter().collect::>(); + let mut name = PROMQL_METRIC_NAME_COLUMN.to_string(); + while occupied.contains(name.as_str()) { + name.push('_'); + } + name + } + /// The labels a vector matching compares by default and under `ignoring(...)`. The marked /// metric-name identity is metadata about which series a row came from, not a label the /// operation matches on. @@ -2387,12 +2374,20 @@ impl PromPlanner { /// Physical columns an explicit `on(...)`/`ignoring(...)`/`group_x(...)` label list names. /// `__name__` is semantic: it names the marked identity column and never a physical column /// of that name. Without a marked column the label is not one the operands carry. + /// + /// The physical name of the marked identity is not a label either: no series carries it, so a + /// label spelled like it - the internal name is not a name a user could have meant, and the + /// identity is metadata, not a label - is not a label of the operands and matches as the + /// absent label every operand reads as the empty value. A real user column of that spelling + /// keeps the identity out of its way: the marker is attached under the first free name, so + /// the comparison below reaches only the identity's own name. fn binary_match_label_mask(labels: &[String], metric_name: Option<&str>) -> BTreeSet { labels .iter() .filter_map(|label| match (label.as_str(), metric_name) { (METRIC_NAME, Some(metric_name)) => Some(metric_name.to_string()), (METRIC_NAME, None) => None, + (label, Some(metric_name)) if label == metric_name => None, _ => Some(label.clone()), }) .collect() @@ -2469,13 +2464,7 @@ impl PromPlanner { .find(|name| !occupied.contains(*name)); let target = match reused { Some(name) => name.clone(), - None => { - let mut name = PROMQL_METRIC_NAME_COLUMN.to_string(); - while occupied.contains(&name) { - name.push('_'); - } - name - } + None => Self::free_metric_name_column_name(occupied.iter().map(String::as_str)), }; if let Some(marker) = &left_marker @@ -2939,6 +2928,9 @@ impl PromPlanner { // `create_function_expr`: the function that reads it (`predict_linear`) then depends on an // argument instead of on planner state written by the selector below it. let range_fold_offset = self.ctx.range_fold_offset.take(); + // A label function writes its destination as a label of the result; the label takes the + // physical name it is written with, so the marked identity moves out of that name first. + let input = self.move_metric_name_out_of_label_destination(func, &args.literals, input)?; let input_tag_columns = self.ctx.tag_columns.clone(); let (mut func_exprs, new_tags) = self.create_function_expr( func, @@ -5029,15 +5021,12 @@ impl PromPlanner { let (dst_name, dst_qualifier) = match Self::metric_name_column(input_schema)? { Some(column) => (column.name, column.relation), 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('_'); - } + let name = Self::free_metric_name_column_name( + input_schema + .fields() + .iter() + .map(|field| field.name().as_str()), + ); // Project the identity like the tag columns it is projected next to. let qualifier = ctx .time_index_column @@ -5068,6 +5057,47 @@ impl PromPlanner { )) } + /// Move the marked identity out of the physical name an ordinary label destination writes. + /// + /// A label function writes its destination as a label of its result, and the label is a + /// column the result carries. When the destination is spelled like the physical column of the + /// input's marked identity - the identity is attached under the first free name, so nothing + /// else can be spelled like it and this reaches only the identity - the label would take that + /// column over and the identity would be lost. The identity is metadata, not a label, so it + /// moves to a free name first and the destination keeps the name it was written with. The + /// semantic `__name__` has no such collision: [`Self::write_destination_label`] writes it into + /// the identity itself. + fn move_metric_name_out_of_label_destination( + &mut self, + func: &Function, + literals: &[DfExpr], + input: LogicalPlan, + ) -> Result { + if !matches!(func.name, "label_join" | "label_replace") { + return Ok(input); + } + let Some(DfExpr::Literal(ScalarValue::Utf8(Some(dst_label)), _)) = literals.first() else { + return Ok(input); + }; + if dst_label.as_str() == METRIC_NAME { + return Ok(input); + } + let Some(marker) = Self::metric_name_column(input.schema())? else { + return Ok(input); + }; + if marker.name.as_str() != dst_label.as_str() { + return Ok(input); + } + let target = Self::free_metric_name_column_name( + input + .schema() + .fields() + .iter() + .map(|field| field.name().as_str()), + ); + Self::rename_metric_name_column(input, &mut self.ctx, &marker.name, &target) + } + /// Build expr for `label_join` function /// /// Returns the joined expression and the physical name of the destination column, which is @@ -5292,10 +5322,22 @@ impl PromPlanner { DfExpr::Literal(ScalarValue::Utf8(Some(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. + // identity included - is the empty value. A missing value is read as that + // empty string here instead of sorting as NULL: ascending would otherwise + // order the series with the empty value after the series that carry the label, + // where Prometheus sorts them first, and descending last. let expr = match Self::resolve_label_col(input_schema, label)? { + // Only a string column is a label; a NULL inside one is the empty value + // as well, so the key keeps comparing the text. + Some(column) + if Self::field_column_type(input_schema, &column.name) + .and_then(Self::string_value_data_type) + .is_some() => + { + Self::label_value_expr(&column.name, input_schema)? + } Some(column) => DfExpr::Column(column), - None => lit(ScalarValue::Utf8(None)), + None => lit(""), }; Ok(expr.sort(asc, false)) } diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index 4d61b867969..3b7ac9ac8fb 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -5135,7 +5135,12 @@ async fn native_histogram_value_sort_is_empty_but_label_sort_preserves_samples() for (function, direction) in [("sort_by_label", "ASC"), ("sort_by_label_desc", "DESC")] { let plan = native_histogram_plan(&format!("{function}(some_metric, \"tag_0\")")).await; - assert!(plan.contains(&format!("tag_0 {direction}")), "{plan}"); + assert!( + plan.contains(&format!( + "coalesce(some_metric.tag_0, Utf8(\"\")) {direction}" + )), + "{plan}" + ); assert!(plan.contains(greptime_native_histogram()), "{plan}"); assert!(!plan.contains("Float64(NULL) IS NOT NULL"), "{plan}"); } @@ -9222,6 +9227,93 @@ async fn build_count_values_table_provider_with_tag_value_array( ) } +/// A table provider with one table per `(name, tag value, sample value)` entry, each carrying a +/// single ordinary tag column named `tag_name` and one `Float64` sample at `ts=1000`. +fn build_user_tagged_table_provider( + tag_name: &str, + metrics: &[(&str, &str, f64)], +) -> DfTableSourceProvider { + let catalog_list = MemoryCatalogManager::with_default_setup(); + for (index, (name, tag_value, value)) in metrics.iter().enumerate() { + let table_id = 4_100 + index as u32; + let columns = vec![ + ColumnSchema::new( + tag_name.to_string(), + ConcreteDataType::string_datatype(), + false, + ), + ColumnSchema::new( + "timestamp".to_string(), + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ) + .with_time_index(true), + ColumnSchema::new( + greptime_value().to_string(), + ConcreteDataType::float64_datatype(), + true, + ), + ]; + let schema = Arc::new(Schema::new(columns)); + let table_meta = TableMetaBuilder::empty() + .schema(schema.clone()) + .primary_key_indices(vec![0]) + .value_indices(vec![2]) + .next_column_id(1024) + .build() + .unwrap(); + let table_info = Arc::new( + TableInfoBuilder::default() + .table_id(table_id) + .name((*name).to_string()) + .meta(table_meta) + .build() + .unwrap(), + ); + let batch = RecordBatch::try_new( + schema.arrow_schema().clone(), + vec![ + Arc::new(StringArray::from(vec![(*tag_value).to_string()])), + Arc::new(TimestampMillisecondArray::from(vec![1_000])), + Arc::new(Float64Array::from(vec![*value])), + ], + ) + .unwrap(); + let backing = GreptimeMemTable::new_with_catalog( + *name, + GreptimeRecordBatch::from_df_record_batch(schema, batch), + table_id, + DEFAULT_CATALOG_NAME.to_string(), + DEFAULT_SCHEMA_NAME.to_string(), + ); + let table = Arc::new(Table::new( + table_info, + FilterPushDownType::Unsupported, + backing.data_source(), + )); + + assert!( + catalog_list + .register_table_sync(RegisterTableRequest { + catalog: DEFAULT_CATALOG_NAME.to_string(), + schema: DEFAULT_SCHEMA_NAME.to_string(), + table_name: (*name).to_string(), + table_id, + table, + }) + .is_ok() + ); + } + + DfTableSourceProvider::new( + catalog_list, + false, + QueryContext::arc(), + DummyDecoder::arc(), + false, + ) +} + /// Collects `(label, value)` pairs of a `count_values` result, where `label` is the /// PromQL label generated by `count_values` and `value` is the aggregated sample value. /// @@ -9863,9 +9955,15 @@ async fn test_sort_by_label_resolves_semantic_name_to_the_marker() { .await .unwrap(); + // A series that does not carry the label reads as the empty value, not as NULL: the sort + // key normalizes the missing value before it is compared. let plan_text = plan.display_indent().to_string(); assert!( - plan_text.contains(&format!("{sort_column} {direction}")), + plan_text.contains(&format!( + "Sort: coalesce({sort_column}, Utf8(\"\")) {direction}" + )) || plan_text.contains(&format!( + "Sort: coalesce(cv_metric.{sort_column}, Utf8(\"\")) {direction}" + )), "{query}: {plan_text}" ); let marker = PromPlanner::metric_name_column(plan.schema()) @@ -9887,6 +9985,88 @@ async fn test_sort_by_label_resolves_semantic_name_to_the_marker() { } } +/// A label a series does not carry is the empty string in PromQL, so a sort on such a label +/// orders those series first ascending and last descending - not the other way round, as a NULL +/// would. `label_replace` into the semantic name empties the identity of the series it matches, +/// which leaves one step holding both the empty value and a real one. +#[tokio::test] +async fn test_sort_by_label_missing_value_sorts_as_the_empty_string() { + let state = build_query_engine_state(); + // The inner `label_replace` empties the name of `k1` only. + let query = |function: &str| { + format!(r#"{function}(label_replace(cv_metric, "__name__", "", "k", "k1"), "__name__")"#) + }; + // (function, expected metric names in the order the plan returns the rows) + for (function, expected_names) in [ + ("sort_by_label", vec!["", "cv_metric", "cv_metric"]), + ("sort_by_label_desc", vec!["cv_metric", "cv_metric", ""]), + ] { + let plan = PromPlanner::stmt_to_plan( + build_count_values_table_provider().await, + &operator_eval_stmt(&query(function)), + &state, + ) + .await + .unwrap_or_else(|err| panic!("{function}: {err}")); + + let marker = PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .unwrap_or_else(|| panic!("{function}: the metric name must survive")); + let sample_column = float_sample_column(&plan); + let (_, batches) = execute(plan, &state).await; + + // The row order of the plan is what the sort produced: read the batches as they come. + let mut rows = batches + .iter() + .flat_map(|batch| { + let tag = batch + .column_by_name("k") + .expect("no series tag column") + .as_any() + .downcast_ref::() + .expect("the series tag must be a string column"); + let name = batch + .column_by_name(&marker.name) + .expect("no metric-name column") + .as_any() + .downcast_ref::() + .expect("the metric name must be a string column"); + (0..batch.num_rows()) + .map(|row| { + let name = if name.is_null(row) { + String::new() + } else { + name.value(row).to_string() + }; + (tag.value(row).to_string(), name) + }) + .collect::>() + }) + .collect::>(); + assert_eq!( + rows.iter() + .map(|(_, name)| name.as_str()) + .collect::>(), + expected_names, + "{function}" + ); + // Only the series the replacement matched lost its name, and every sample survives. + rows.sort(); + assert_eq!( + rows, + vec![ + ("k0".to_string(), "cv_metric".to_string()), + ("k1".to_string(), String::new()), + ("k2".to_string(), "cv_metric".to_string()), + ], + "{function}" + ); + let mut samples = values(&batches, &sample_column); + samples.sort_by(f64::total_cmp); + assert_eq!(samples, vec![1.0, 1.0, 2.0], "{function}"); + } +} + #[tokio::test] async fn test_range_functions_control_metric_name_identity() { let state = build_query_engine_state(); @@ -10024,9 +10204,9 @@ async fn test_sort_by_label_semantic_name_without_marker_sorts_as_absent() { .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. + // The name is gone, so the label the sort reads is absent: it sorts as the empty value + // 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 plan = PromPlanner::stmt_to_plan(provider().await, &operator_eval_stmt(&query), &state) .await @@ -10034,7 +10214,7 @@ async fn test_sort_by_label_semantic_name_without_marker_sorts_as_absent() { let plan_text = plan.display_indent().to_string(); assert!( plan_text.contains(&format!( - "Sort: Utf8(NULL) {} NULLS", + "Sort: Utf8(\"\") {} NULLS", if function.ends_with("desc") { "DESC" } else { @@ -10043,6 +10223,13 @@ async fn test_sort_by_label_semantic_name_without_marker_sorts_as_absent() { )), "{query}: {plan_text}" ); + // The physical `__name__` label of the input is not what the semantic name resolves to. + assert!( + plan.schema() + .field_with_unqualified_name("__name__") + .is_ok(), + "{query}: the physical `__name__` label must survive: {plan_text}" + ); let query = format!(r#"{function}(collision_metric, "__name__")"#); let plan = PromPlanner::stmt_to_plan(provider().await, &operator_eval_stmt(&query), &state) @@ -10054,7 +10241,7 @@ async fn test_sort_by_label_semantic_name_without_marker_sorts_as_absent() { assert!( plan.display_indent() .to_string() - .contains(&format!("Sort: {}", marker.name)) + .contains(&format!("Sort: coalesce({}, Utf8(\"\"))", marker.name)) ); } } @@ -10153,13 +10340,16 @@ async fn test_topk_bottomk_preserve_metric_name_and_samples() { } } -/// `count_values("__name__", ...)` overwrites the identity of its input with the formatted -/// sample value: the result carries exactly one marked column, holding the sample value, and -/// the identity of the input must not survive as a second marker. +/// `count_values("__name__", ...)` writes the sample value under the semantic name as an +/// *ordinary* label: the generated column is a plain string column in Prometheus' textual form, +/// no field carries the identity role, and the identity of the input does not survive it. A +/// plain aggregate argument must not be able to forge the internal marker, so nothing later has +/// an identity to drop here - an enclosing expression that drops the name of its input leaves the +/// generated label alone and the result stays a vector. #[tokio::test] -async fn test_count_values_semantic_name_overwrites_metric_identity() { +async fn test_count_values_semantic_name_is_an_ordinary_label() { let state = build_query_engine_state(); - // (query, expected `(generated name, aggregated value)` rows) + // (query, expected `(generated label value, aggregated value)` rows) for (query, expected) in [ ( r#"count_values("__name__", cv_metric)"#, @@ -10170,13 +10360,21 @@ async fn test_count_values_semantic_name_overwrites_metric_identity() { vec![("1", 2.0), ("2", 1.0)], ), ( - r#"sum by (__name__) (count_values("__name__", cv_metric))"#, + r#"count_values("__name__", abs(cv_metric))"#, vec![("1", 2.0), ("2", 1.0)], ), - // Without an input identity the generated column must take neither the name of a - // physical field of the input nor the semantic `__name__` itself. + // An enclosing unary or binary operator drops the name of its input, which is already + // gone: the generated label is a label, not an identity, and the samples are untouched. ( - r#"count_values("__name__", abs(cv_metric))"#, + r#"abs(count_values("__name__", cv_metric))"#, + vec![("1", 2.0), ("2", 1.0)], + ), + ( + r#"-count_values("__name__", cv_metric)"#, + vec![("1", -2.0), ("2", -1.0)], + ), + ( + r#"count_values("__name__", cv_metric) + 0"#, vec![("1", 2.0), ("2", 1.0)], ), ] { @@ -10188,154 +10386,262 @@ async fn test_count_values_semantic_name_overwrites_metric_identity() { .await .unwrap_or_else(|err| panic!("{query}: {err}")); - let markers = plan + assert!( + PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .is_none(), + "{query}: the generated label must not be an identity: {}", + plan.display_indent() + ); + let generated = plan .schema() .fields() .iter() - .filter(|field| { - field - .metadata() - .get(PROMQL_FIELD_ROLE_KEY) - .map(String::as_str) - == Some(PROMQL_METRIC_NAME_ROLE) - }) - .map(|field| field.name().clone()) + .filter(|field| field.name() == METRIC_NAME) .collect::>(); - assert_eq!( - markers, - vec![PROMQL_METRIC_NAME_COLUMN.to_string()], + assert_eq!(generated.len(), 1, "{query}: {}", plan.display_indent()); + assert!( + generated[0].metadata().get(PROMQL_FIELD_ROLE_KEY).is_none(), + "{query}: the generated label must be ordinary: {}", + plan.display_indent() + ); + // The identity of the input does not survive as a second column of that name. + assert!( + plan.schema() + .field_with_unqualified_name(PROMQL_METRIC_NAME_COLUMN) + .is_err(), "{query}: {}", plan.display_indent() ); let (_, batches) = execute(plan, &state).await; - assert_metric_name_in_batches(&batches, PROMQL_METRIC_NAME_COLUMN); assert_eq!( - count_values_rows(&batches, PROMQL_METRIC_NAME_COLUMN), + count_values_rows(&batches, METRIC_NAME), + expected, + "{query}" + ); + assert_unique_label_set_per_timestamp(&batches, METRIC_NAME); + } + + // The generated column is a physical label spelled `__name__`, and the planner reads + // `__name__` only through a marked identity: `by (__name__)` next to such a result groups by + // nothing, exactly like `by (__name__)` on a plan without an identity. The label stays in the + // schema and in the output rows of the inner result, but it is not reachable as the name. + let query = r#"sum by (__name__) (count_values("__name__", cv_metric))"#; + 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!( + PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .is_none(), + "{query}: {}", + plan.display_indent() + ); + assert!( + plan.schema() + .field_with_unqualified_name(METRIC_NAME) + .is_err(), + "the semantic name is not a label to group by here: {}", + plan.display_indent() + ); + let (_, batches) = execute(plan, &state).await; + let mut actual = batches + .iter() + .flat_map(|batch| { + let value_index = batch + .schema() + .fields() + .iter() + .position(|field| { + matches!( + field.data_type(), + ArrowDataType::Float64 | ArrowDataType::Int64 | ArrowDataType::UInt64 + ) + }) + .expect("no numeric value column"); + datafusion::arrow::compute::cast(batch.column(value_index), &ArrowDataType::Float64) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap() + .values() + .to_vec() + }) + .collect::>(); + actual.sort_by(f64::total_cmp); + // The three samples of the two generated groups are summed into one series. + assert_eq!(actual, vec![3.0], "{query}"); +} + +/// Dropping the metric name inserts a label-set check only where the input can actually hold two +/// rows of one label set. A set of series the name did not distinguish stays readable - the check +/// would only add a group-by assertion nothing can violate - and a `count_values` result carries +/// its generated name as an ordinary label, so dropping an input name next to it is as safe. +#[tokio::test] +async fn test_dropping_metric_name_adds_no_check_without_a_union() { + let state = build_query_engine_state(); + // Three series that stay distinct without the name: the samples are read as they are. + for (query, expected) in [ + ( + "abs(cv_metric)", + vec![ + ("k0".to_string(), 1.0), + ("k1".to_string(), 2.0), + ("k2".to_string(), 1.0), + ], + ), + ( + "cv_metric + 0", + vec![ + ("k0".to_string(), 1.0), + ("k1".to_string(), 2.0), + ("k2".to_string(), 1.0), + ], + ), + ] { + 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 + .display_indent() + .to_string() + .contains("prom_assert_unique_match_group"), + "{query}: no label-set check may be inserted: {}", + plan.display_indent() + ); + + let sample_column = float_sample_column(&plan); + let (_, batches) = execute(plan, &state).await; + assert_eq!( + cv_rows(&batches, &sample_column) + .into_iter() + .map(|(tag, _, value)| (tag, value)) + .collect::>(), + expected, + "{query}" + ); + } + + // The generated name of `count_values("__name__", ...)` is a label, not an identity, so an + // enclosing expression that drops the name of its input has nothing to refuse. + for (query, expected) in [ + ( + r#"abs(count_values("__name__", cv_metric))"#, + vec![("1", 2.0), ("2", 1.0)], + ), + ( + r#"count_values("__name__", cv_metric) + 0"#, + vec![("1", 2.0), ("2", 1.0)], + ), + ] { + 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 + .display_indent() + .to_string() + .contains("prom_assert_unique_match_group"), + "{query}: no label-set check may be inserted: {}", + plan.display_indent() + ); + + let (_, batches) = execute(plan, &state).await; + assert_eq!( + count_values_rows(&batches, METRIC_NAME), expected, "{query}" ); } } -/// `count_values("__name__", ...)` writes the semantic metric name, which lives in the marked -/// identity column and never in a physical column of the input: a physical `__name__` label (or a -/// physical column occupying the internal marker name) survives unmarked, while the generated -/// name moves to the sole marked column, suffixed on collision. +/// `count_values("__name__", ...)` behaves like any other destination next to a physical column +/// of the input: the generated label is named `__name__` and is always ordinary - never marked - +/// so a physical label of that name is taken over by the generated one, and a physical column +/// occupying the internal marker's name stays a label of its own. #[tokio::test] -async fn test_count_values_semantic_name_keeps_physical_collision_columns() { +async fn test_count_values_semantic_name_next_to_physical_columns() { let state = build_query_engine_state(); - // (physical tag columns, physical column that must survive, generated name column, query) - for (tags, physical, generated, query) in [ + for (tag, expected_rows, expected_tag_values, query) in [ + // The generated label takes the name over, exactly like an ordinary destination does: + // `__name__` is the only column of that name, the samples of the three input series group + // into one row, and that row carries the generated value instead of the physical labels. ( - vec![METRIC_NAME, "job"], METRIC_NAME, - PROMQL_METRIC_NAME_COLUMN.to_string(), - r#"count_values("__name__", collision_metric) without(job)"#, + vec![("1", 3.0)], + vec!["1"], + r#"count_values("__name__", cv_metric) without(job)"#, ), + // Next to an ordinary label named like the internal marker the generated label is a + // second, distinct label: three series of one sample value stay three groups, each with + // its own sample of the tag. ( - vec![PROMQL_METRIC_NAME_COLUMN, "job"], PROMQL_METRIC_NAME_COLUMN, - format!("{PROMQL_METRIC_NAME_COLUMN}_"), - r#"count_values("__name__", abs(collision_metric)) without(job)"#, + vec![("1", 1.0), ("1", 1.0), ("1", 1.0)], + vec!["k0", "k1", "k2"], + r#"count_values("__name__", abs(cv_metric)) without(job)"#, ), ] { let plan = PromPlanner::stmt_to_plan( - build_test_table_provider_with_distinct_tags(&[("collision_metric", tags.as_slice())]) - .await, + build_count_values_collision_table_provider(tag).await, &operator_eval_stmt(query), &state, ) .await .unwrap_or_else(|err| panic!("{query}: {err}")); - let field = plan - .schema() - .field_with_unqualified_name(physical) - .unwrap_or_else(|_| panic!("{query}: the physical `{physical}` column must survive")); assert!( - field.metadata().get(PROMQL_FIELD_ROLE_KEY).is_none(), - "{query}: the physical `{physical}` column must stay unmarked: {}", + plan.schema() + .fields() + .iter() + .all(|field| field.metadata().get(PROMQL_FIELD_ROLE_KEY).is_none()), + "{query}: no field may carry the identity role: {}", plan.display_indent() ); - let marked = plan - .schema() - .fields() - .iter() - .filter(|field| { - field - .metadata() - .get(PROMQL_FIELD_ROLE_KEY) - .map(String::as_str) - == Some(PROMQL_METRIC_NAME_ROLE) - }) - .map(|field| field.name().clone()) - .collect::>(); assert_eq!( - marked, - vec![generated], + plan.schema() + .fields() + .iter() + .filter(|field| field.name() == METRIC_NAME) + .count(), + 1, "{query}: {}", plan.display_indent() ); + + let (_, batches) = execute(plan, &state).await; + assert_eq!( + count_values_rows(&batches, METRIC_NAME), + expected_rows, + "{query}" + ); + let mut actual_physical = string_values(&batches, tag); + actual_physical.sort(); + assert_eq!( + actual_physical, + expected_tag_values + .iter() + .map(|value| Some((*value).to_string())) + .collect::>(), + "{query}" + ); } } -/// A physical `__name__` label is an ordinary label, not the semantic metric name: with -/// `without(...)` grouping `count_values("__name__", ...)` keeps it and the generated name goes to -/// the sole marked column. Distinct physical labels with equal sample values must still be -/// separate groups. -#[tokio::test] -async fn test_count_values_keeps_distinct_physical_name_labels() { - let state = build_query_engine_state(); - let plan = PromPlanner::stmt_to_plan( - build_count_values_collision_table_provider(METRIC_NAME).await, - &operator_eval_stmt(r#"count_values("__name__", cv_metric) without(job)"#), - &state, - ) - .await - .unwrap(); - - let physical = plan - .schema() - .field_with_unqualified_name(METRIC_NAME) - .expect("the physical `__name__` label must survive"); - assert!(physical.metadata().get(PROMQL_FIELD_ROLE_KEY).is_none()); - let marker = PromPlanner::metric_name_column(plan.schema()) - .unwrap() - .unwrap_or_else(|| { - panic!( - "the generated name must be marked: {}", - plan.display_indent() - ) - }); - assert_eq!(marker.name, PROMQL_METRIC_NAME_COLUMN); - - let (_, batches) = execute(plan, &state).await; - assert_metric_name_in_batches(&batches, &marker.name); - // `k0`, `k1` and `k2` share one sample value and one timestamp: a single group would mean the - // physical `__name__` label was dropped from the grouping key. - assert_eq!( - count_values_rows(&batches, &marker.name), - vec![("1", 1.0), ("1", 1.0), ("1", 1.0)] - ); - let mut physical_labels = batches - .iter() - .flat_map(|batch| { - let labels = batch - .column_by_name(METRIC_NAME) - .expect("the physical `__name__` label must be projected") - .as_any() - .downcast_ref::() - .expect("the physical `__name__` label must be a string column"); - (0..batch.num_rows()) - .map(|row| labels.value(row).to_string()) - .collect::>() - }) - .collect::>(); - physical_labels.sort(); - assert_eq!(physical_labels, vec!["k0", "k1", "k2"]); -} - /// `absent()` drops the metric-name identity of its input - even when the selector matched no /// series - and must still emit one 1-valued sample per evaluation timestamp instead of an /// empty result. @@ -10979,6 +11285,112 @@ async fn test_dropped_metric_name_rejects_duplicate_label_sets() { } } +/// An explicit `on(...)`/`ignoring(...)`/`group_x(...)` label list reads the same semantic labels +/// as every other label read: `__name__` names the marked identity, a label no operand carries +/// matches like `on()`, and the physical name of the internal identity is not a label of any +/// series - the identity is metadata the planner attaches, so the name it happens to occupy means +/// nothing. A real user column of that spelling stays a label like any other. +#[tokio::test] +async fn test_binary_matching_reads_the_same_semantic_labels() { + let state = build_query_engine_state(); + + async fn rows_of( + query: &str, + state: &QueryEngineState, + ) -> Vec<(Option, Option, f64)> { + 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 marker = PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .map(|marker| marker.name); + let sample_column = float_sample_column(&plan); + let (_, batches) = execute(plan, state).await; + binary_result_rows(&batches, marker.as_deref(), "tag", &sample_column) + } + + // Neither operand carries a label of that name, so the label list names nothing and the + // operation matches the way the same query with `on()` does. + for (marker_query, plain_query, expected) in [ + ( + "left_metric + on(__promql_metric_name) right_metric", + "left_metric + on() right_metric", + vec![(None, None, 5.0)], + ), + ( + "left_metric or on(__promql_metric_name) right_metric", + "left_metric or on() right_metric", + vec![(Some("shared"), Some("left_metric"), 2.0)], + ), + ] { + let rows = rows_of(marker_query, &state).await; + assert_eq!( + rows, + rows_of(plain_query, &state).await, + "{marker_query} must match like {plain_query}" + ); + assert_eq!( + rows, + expected_binary_result_rows(&expected), + "{marker_query}" + ); + } + + // A real user column of that spelling keeps its meaning: `on(...)` compares it, and the + // control query matches without it. + async fn collision_rows( + query: &str, + state: &QueryEngineState, + ) -> Vec<(Option, Option, f64)> { + let plan = PromPlanner::stmt_to_plan( + build_user_tagged_table_provider( + PROMQL_METRIC_NAME_COLUMN, + &[ + ("l_metric", "shared", 2.0), + ("r_same", "shared", 3.0), + ("r_other", "other", 3.0), + ], + ), + &operator_eval_stmt(query), + state, + ) + .await + .unwrap_or_else(|err| panic!("{query}: {err}")); + let marker = PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .map(|marker| marker.name); + let sample_column = float_sample_column(&plan); + let (_, batches) = execute(plan, state).await; + binary_result_rows( + &batches, + marker.as_deref(), + PROMQL_METRIC_NAME_COLUMN, + &sample_column, + ) + } + + for (query, expected) in [ + // The user column decides the matching: equal values match, different ones do not. + ( + "l_metric + on(__promql_metric_name) r_same", + vec![(Some("shared"), None, 5.0)], + ), + ("l_metric + on(__promql_metric_name) r_other", vec![]), + // ... and an operation that does not name it matches without it. + ("l_metric + on() r_other", vec![(None, None, 5.0)]), + ] { + assert_eq!( + collision_rows(query, &state).await, + expected_binary_result_rows(&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] @@ -11633,6 +12045,102 @@ async fn test_label_functions_semantic_name_destination_avoids_physical_collisio ); } +/// An ordinary destination of a label function takes the name it was written with, identity or +/// not: a destination spelled like the physical column of the marked identity moves the identity +/// to a free name instead of overwriting it with a label that carries no role, so the metric name +/// survives the rewrite. +#[tokio::test] +async fn test_label_function_destination_takes_the_identity_column_name() { + let state = build_query_engine_state(); + + // `label_replace` never reaches that name: its destination is a user label, and a label name + // starting with `__` is reserved and refused before the plan is built. + let query = r#"label_replace(cv_metric, "__promql_metric_name", "$1", "k", "(.*)")"#; + let err = PromPlanner::stmt_to_plan( + build_count_values_table_provider().await, + &operator_eval_stmt(query), + &state, + ) + .await + .expect_err("a reserved destination label must be refused"); + assert!( + err.to_string() + .contains("Invalid destination label name in label_replace()"), + "{query}: {err}" + ); + + // `label_join` writes its destination under its own name, which moves the identity instead. + let query = r#"label_join(cv_metric, "__promql_metric_name", "", "k")"#; + 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 destination is the only ordinary column of that name, and it carries no role. + let destination = plan + .schema() + .field_with_unqualified_name(PROMQL_METRIC_NAME_COLUMN) + .unwrap_or_else(|_| { + panic!( + "{query}: the destination must exist: {}", + plan.display_indent() + ) + }); + assert!( + destination.metadata().get(PROMQL_FIELD_ROLE_KEY).is_none(), + "{query}: the destination must be ordinary: {}", + plan.display_indent() + ); + let marker = PromPlanner::metric_name_column(plan.schema()) + .unwrap() + .unwrap_or_else(|| panic!("{query}: the identity must survive")); + assert_eq!( + marker.name, + format!("{PROMQL_METRIC_NAME_COLUMN}_"), + "{query}: {}", + plan.display_indent() + ); + assert_eq!( + plan.schema() + .fields() + .iter() + .filter(|field| field + .metadata() + .get(PROMQL_FIELD_ROLE_KEY) + .map(String::as_str) + == Some(PROMQL_METRIC_NAME_ROLE)) + .count(), + 1, + "{query}: {}", + plan.display_indent() + ); + + let (_, batches) = execute(plan, &state).await; + // The identity still names every series after the rewrite ... + assert_eq!( + tagged_string_rows(&batches, &marker.name), + vec![ + ("k0".to_string(), Some("cv_metric".to_string())), + ("k1".to_string(), Some("cv_metric".to_string())), + ("k2".to_string(), Some("cv_metric".to_string())), + ], + "{query}" + ); + // ... and the written label holds the joined value of its own series. + assert_eq!( + tagged_string_rows(&batches, PROMQL_METRIC_NAME_COLUMN), + vec![ + ("k0".to_string(), Some("k0".to_string())), + ("k1".to_string(), Some("k1".to_string())), + ("k2".to_string(), Some("k2".to_string())), + ], + "{query}" + ); +} + /// An ordinary `label_replace`/`label_join` - neither source nor destination names `__name__` - /// 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