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

fix(promql): close metric-name identity privilege and boundary gaps

Browse Source 
- 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>
This commit is contained in:
discord9
2026-10-12 11:13:07 +08:00
parent 6106a6ea08
commit 1817d0bb76
2 changed files with 751 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/query/src/promql/planner.rs
+121 -79
View File
@@ -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::<HashSet<_>>();
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::<Vec<_>>();
@@ -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::<HashSet<_>>();
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<Item = &'a str>) -> String {
let occupied = occupied.into_iter().collect::<HashSet<_>>();
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<String> {
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::<HashSet<_>>();
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<LogicalPlan> {
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))
}
src/query/src/promql/planner/test.rs
+630 -122
View File
@@ -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::<StringArray>()
.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::<StringArray>()
.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::<Vec<_>>()
})
.collect::<Vec<_>>();
assert_eq!(
rows.iter()
.map(|(_, name)| name.as_str())
.collect::<Vec<_>>(),
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::<Vec<_>>();
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::<Float64Array>()
.unwrap()
.values()
.to_vec()
})
.collect::<Vec<_>>();
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::<Vec<_>>(),
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::<Vec<_>>();
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::<Vec<_>>(),
"{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::<StringArray>()
.expect("the physical `__name__` label must be a string column");
(0..batch.num_rows())
.map(|row| labels.value(row).to_string())
.collect::<Vec<_>>()
})
.collect::<Vec<_>>();
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<String>, Option<String>, 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<String>, Option<String>, 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