mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-13 07:22:32 +00:00
feat(promql): reserve __name__ as the metric name at the PromQL boundary
A table whose own schema carries a column spelled __name__ cannot be interpreted as PromQL input: the semantic metric name is the table's identity, so a second, ordinary business column of that name would be ambiguous. setup_context() now refuses such tables with ReservedMetricNameColumn (InvalidArguments) at the single selector entry, before any context is filled; legitimately rewritten names from label_join/label_replace/count_values are produced above the selector and never pass through it. With the reserved-name contract, an unmarked physical __name__ column can no longer reach the Prometheus HTTP serializer, so the exported_name auto-rename branch in ColumnLayout::infer is removed and tag names are emitted verbatim. Also narrow column_reference_is_constant to an explicit whitelist of value-preserving plan nodes instead of any single-input passthrough, so a future extension cannot silently waive the dropped-name uniqueness check. Tests that constructed a physical __name__ tag as legal input now assert the entry refusal; the internal __promql_metric_name collision tests are unaffected. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
@@ -254,6 +254,16 @@ pub enum Error {
|
||||
#[snafu(implicit)]
|
||||
location: Location,
|
||||
},
|
||||
|
||||
#[snafu(display(
|
||||
"Table {} cannot be queried with PromQL: column '__name__' conflicts with the reserved metric name",
|
||||
table
|
||||
))]
|
||||
ReservedMetricNameColumn {
|
||||
table: String,
|
||||
#[snafu(implicit)]
|
||||
location: Location,
|
||||
},
|
||||
}
|
||||
|
||||
impl ErrorExt for Error {
|
||||
@@ -281,7 +291,8 @@ impl ErrorExt for Error {
|
||||
| SystemTimeOutOfRange { .. }
|
||||
| AtModifierTimestampOutOfRange { .. }
|
||||
| InvalidRegularExpression { .. }
|
||||
| InvalidDestinationLabelName { .. } => StatusCode::InvalidArguments,
|
||||
| InvalidDestinationLabelName { .. }
|
||||
| ReservedMetricNameColumn { .. } => StatusCode::InvalidArguments,
|
||||
|
||||
UnknownTable { .. } => StatusCode::Internal,
|
||||
|
||||
|
||||
@@ -107,10 +107,11 @@ use crate::promql::error::{
|
||||
CatalogSnafu, ColumnNotFoundSnafu, DataFusionPlanningSnafu, ExpectRangeSelectorSnafu,
|
||||
FunctionInvalidArgumentSnafu, InvalidDestinationLabelNameSnafu, InvalidRegularExpressionSnafu,
|
||||
InvalidTimeRangeSnafu, MultiFieldsNotSupportedSnafu, MultipleMetricMatchersSnafu,
|
||||
MultipleVectorSnafu, NoMetricMatcherSnafu, RangeVectorInRangeQuerySnafu, Result,
|
||||
SubqueryTimestampOutOfRangeSnafu, TableNameNotFoundSnafu, TimeIndexNotFoundSnafu,
|
||||
UnexpectedPlanExprSnafu, UnexpectedTokenSnafu, UnknownTableSnafu, UnsupportedExprSnafu,
|
||||
UnsupportedMatcherOpSnafu, ValueNotFoundSnafu, ZeroRangeSelectorSnafu, ZeroSubqueryStepSnafu,
|
||||
MultipleVectorSnafu, NoMetricMatcherSnafu, RangeVectorInRangeQuerySnafu,
|
||||
ReservedMetricNameColumnSnafu, Result, SubqueryTimestampOutOfRangeSnafu,
|
||||
TableNameNotFoundSnafu, TimeIndexNotFoundSnafu, UnexpectedPlanExprSnafu, UnexpectedTokenSnafu,
|
||||
UnknownTableSnafu, UnsupportedExprSnafu, UnsupportedMatcherOpSnafu, ValueNotFoundSnafu,
|
||||
ZeroRangeSelectorSnafu, ZeroSubqueryStepSnafu,
|
||||
};
|
||||
use crate::promql::label_values::signed_millis_since_epoch;
|
||||
use crate::query_engine::QueryEngineState;
|
||||
@@ -1253,8 +1254,9 @@ impl PromPlanner {
|
||||
/// Whether a plan can hold such rows is not a property of one operator - nesting hides the
|
||||
/// operator that produced them from the one that drops the name - so every drop asserts the
|
||||
/// unique label set unless the identity provably cannot distinguish two rows:
|
||||
/// [`Self::column_is_constant`] follows the marked column to the node that computes it and
|
||||
/// checks that every row of the plan carries the same value in it, and a label that is the
|
||||
/// [`Self::column_is_constant`] follows the marked column to the node that computes it,
|
||||
/// through the nodes that hand the rows of their input on and stopping at every other one,
|
||||
/// and checks that every row of the plan carries the same value in it; a label that is the
|
||||
/// same in every row cannot tell two of them apart. A plan without such a column (nothing is
|
||||
/// dropped) needs no check either.
|
||||
fn drop_metric_name(&mut self, mut input: LogicalPlan) -> Result<LogicalPlan> {
|
||||
@@ -1301,10 +1303,11 @@ impl PromPlanner {
|
||||
///
|
||||
/// The value of a column is decided by the node that computes it: a literal is the same in
|
||||
/// every row, and a reference to another column keeps whatever that column holds, so the walk
|
||||
/// follows references down to the node that produces the value. A node that combines the rows
|
||||
/// of several inputs can decide a column per row out of data no single input holds - a union
|
||||
/// of two metrics takes the name from the operand the row came from - so those nodes answer
|
||||
/// `false`, as does anything the walk cannot follow to a literal.
|
||||
/// follows references down to the node that produces the value - and only through nodes
|
||||
/// known to keep the rows of their input and the values of its columns. A node that combines
|
||||
/// the rows of several inputs can decide a column per row out of data no single input holds -
|
||||
/// a union of two metrics takes the name from the operand the row came from - so those nodes
|
||||
/// answer `false`, as does anything the walk cannot follow to a literal.
|
||||
fn column_is_constant(plan: &LogicalPlan, column: &str) -> bool {
|
||||
Self::column_reference_is_constant(plan, &Column::from_name(column))
|
||||
}
|
||||
@@ -1313,7 +1316,9 @@ impl PromPlanner {
|
||||
///
|
||||
/// A qualified reference is resolved where the inputs of a plan meet, since only the node
|
||||
/// that keeps the rows of both of them can tell which one holds the value; everywhere else
|
||||
/// the walk follows the column by name.
|
||||
/// the walk follows the column by name. Following a node is only safe when its output is its
|
||||
/// input's rows and columns, so the walk goes through the nodes [`Self::hands_rows_through`]
|
||||
/// lists and stops at every other node.
|
||||
fn column_reference_is_constant(plan: &LogicalPlan, source: &Column) -> bool {
|
||||
if let LogicalPlan::Projection(projection) = plan {
|
||||
let Some(expr) = projection
|
||||
@@ -1342,6 +1347,10 @@ impl PromPlanner {
|
||||
_ => false,
|
||||
};
|
||||
}
|
||||
let follows_input = Self::hands_rows_through(plan);
|
||||
if !follows_input {
|
||||
return false;
|
||||
}
|
||||
match plan.inputs().as_slice() {
|
||||
// A node that combines several inputs decides a column per row out of all of them.
|
||||
[_, _, ..] => false,
|
||||
@@ -1351,6 +1360,37 @@ impl PromPlanner {
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether `plan` hands the rows of its single input on - dropping, reordering, copying or
|
||||
/// qualifying them, but never recomputing what one of their columns holds.
|
||||
///
|
||||
/// Only such a node keeps the value of a column of its input, so only such a node may be
|
||||
/// followed by [`Self::column_reference_is_constant`]. Every other node may decide the column
|
||||
/// per row out of data the walk cannot see - a node the walk does not know about above all -
|
||||
/// so it is not followed and the caller checks the label set instead of assuming the value is
|
||||
/// the same in every row.
|
||||
fn hands_rows_through(plan: &LogicalPlan) -> bool {
|
||||
match plan {
|
||||
LogicalPlan::Filter(_)
|
||||
| LogicalPlan::Sort(_)
|
||||
| LogicalPlan::Limit(_)
|
||||
| LogicalPlan::Window(_)
|
||||
| LogicalPlan::SubqueryAlias(_)
|
||||
| LogicalPlan::Distinct(_)
|
||||
| LogicalPlan::Repartition(_) => true,
|
||||
// The PromQL planner's own row manipulators: they replay, reorder or divide the rows
|
||||
// of their input and add columns of their own, while the columns they carry keep the
|
||||
// values they came in with.
|
||||
LogicalPlan::Extension(extension) => {
|
||||
let node = extension.node.as_any();
|
||||
node.is::<InstantManipulate>()
|
||||
|| node.is::<RangeManipulate>()
|
||||
|| node.is::<SeriesDivide>()
|
||||
|| node.is::<SeriesNormalize>()
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether a projection's output column `source` is the one `expr` produces.
|
||||
///
|
||||
/// The name a projection gives its output is unqualified - the qualifier is the relation the
|
||||
@@ -4105,6 +4145,24 @@ impl PromPlanner {
|
||||
};
|
||||
let table = self.table_from_source(&source)?;
|
||||
|
||||
// `__name__` is the metric name of a series in PromQL and nothing else, so a table that
|
||||
// brings a column of that spelling has no way to be evaluated: the reserved name would
|
||||
// have to be the name of the table's series and that column at once, and no reader of the
|
||||
// result - the Prometheus HTTP API above all - could tell which one a row carries. The
|
||||
// schema read here is the table's own, before the planner attaches anything, so every
|
||||
// column it holds is a column the user wrote.
|
||||
if table
|
||||
.schema()
|
||||
.column_schemas()
|
||||
.iter()
|
||||
.any(|column| column.name == METRIC_NAME)
|
||||
{
|
||||
return ReservedMetricNameColumnSnafu {
|
||||
table: table_ref.to_quoted_string(),
|
||||
}
|
||||
.fail();
|
||||
}
|
||||
|
||||
// set time index column name
|
||||
let time_index = table
|
||||
.schema()
|
||||
|
||||
@@ -10179,71 +10179,140 @@ async fn test_histogram_quantile_folds_with_identity_then_drops_it() {
|
||||
);
|
||||
}
|
||||
|
||||
/// A table whose *value* column, rather than one of its tags, carries a caller provided spelling.
|
||||
/// The reserved-name rejection reads the schema of the table, so the role a column plays in it
|
||||
/// does not change the verdict.
|
||||
async fn build_value_column_table_provider(table_name: &str, value: &str) -> DfTableSourceProvider {
|
||||
let catalog_list = MemoryCatalogManager::with_default_setup();
|
||||
let columns = vec![
|
||||
ColumnSchema::new(
|
||||
"job".to_string(),
|
||||
ConcreteDataType::string_datatype(),
|
||||
false,
|
||||
),
|
||||
ColumnSchema::new(
|
||||
greptime_timestamp().to_string(),
|
||||
ConcreteDataType::timestamp_millisecond_datatype(),
|
||||
false,
|
||||
)
|
||||
.with_time_index(true),
|
||||
ColumnSchema::new(
|
||||
value.to_string(),
|
||||
ConcreteDataType::float64_datatype(),
|
||||
true,
|
||||
),
|
||||
];
|
||||
let table_meta = TableMetaBuilder::empty()
|
||||
.schema(Arc::new(Schema::new(columns)))
|
||||
.primary_key_indices(vec![0])
|
||||
.value_indices(vec![2])
|
||||
.next_column_id(1024)
|
||||
.build()
|
||||
.unwrap();
|
||||
let table_info = TableInfoBuilder::default()
|
||||
.name(table_name.to_string())
|
||||
.meta(table_meta)
|
||||
.build()
|
||||
.unwrap();
|
||||
|
||||
assert!(
|
||||
catalog_list
|
||||
.register_table_sync(RegisterTableRequest {
|
||||
catalog: DEFAULT_CATALOG_NAME.to_string(),
|
||||
schema: DEFAULT_SCHEMA_NAME.to_string(),
|
||||
table_name: table_name.to_string(),
|
||||
table_id: 1024,
|
||||
table: EmptyTable::from_table_info(&table_info),
|
||||
})
|
||||
.is_ok()
|
||||
);
|
||||
|
||||
DfTableSourceProvider::new(
|
||||
catalog_list,
|
||||
false,
|
||||
QueryContext::arc(),
|
||||
DummyDecoder::arc(),
|
||||
false,
|
||||
)
|
||||
}
|
||||
|
||||
/// A table that brings its own `__name__` column cannot be evaluated: `__name__` is the metric
|
||||
/// name of a series and nothing else, so the column would have to be that name and an ordinary
|
||||
/// column of the row at once. Every selector reads its table through the same entry point, so the
|
||||
/// plain selector, the one inside a call or a label function, the one a range selector wraps and
|
||||
/// the one a subquery wraps are all refused with the table.
|
||||
#[tokio::test]
|
||||
async fn test_sort_by_label_semantic_name_without_marker_sorts_as_absent() {
|
||||
async fn test_reserved_metric_name_column_rejects_every_selector() {
|
||||
let state = build_query_engine_state();
|
||||
let provider = || async {
|
||||
build_test_table_provider_with_distinct_tags(&[("collision_metric", &["__name__", "job"])])
|
||||
.await
|
||||
build_test_table_provider_with_distinct_tags(&[
|
||||
("reserved_name_metric", &["__name__", "job"]),
|
||||
("plain_metric", &["job"]),
|
||||
])
|
||||
.await
|
||||
};
|
||||
let plan = PromPlanner::stmt_to_plan(
|
||||
provider().await,
|
||||
&operator_eval_stmt("abs(collision_metric)"),
|
||||
let expected = concat!(
|
||||
"Table reserved_name_metric cannot be queried with PromQL: ",
|
||||
"column '__name__' conflicts with the reserved metric name"
|
||||
);
|
||||
for query in [
|
||||
// The plain vector selector.
|
||||
"reserved_name_metric",
|
||||
// The selector a matcher that names the metric instead of the table reads.
|
||||
r#"{__name__="reserved_name_metric"}"#,
|
||||
// The selector inside a call and inside a label function.
|
||||
"abs(reserved_name_metric)",
|
||||
r#"sort_by_label(reserved_name_metric, "job")"#,
|
||||
r#"label_join(reserved_name_metric, "name_copy", "/", "job")"#,
|
||||
// The selector a range selector wraps, next to the one a subquery wraps.
|
||||
"rate(reserved_name_metric[5m])",
|
||||
"count_over_time(reserved_name_metric[5m:1m])",
|
||||
] {
|
||||
let err = PromPlanner::stmt_to_plan(provider().await, &operator_eval_stmt(query), &state)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(err.to_string(), expected, "{query}");
|
||||
assert_eq!(err.status_code(), StatusCode::InvalidArguments, "{query}");
|
||||
}
|
||||
|
||||
// The same shapes read a table without a column of the reserved spelling, and a table that
|
||||
// does not exist is still an empty metric rather than a rejection.
|
||||
for query in [
|
||||
"plain_metric",
|
||||
"abs(plain_metric)",
|
||||
"rate(plain_metric[5m])",
|
||||
"count_over_time(plain_metric[5m:1m])",
|
||||
"nonexistent_metric",
|
||||
] {
|
||||
PromPlanner::stmt_to_plan(provider().await, &operator_eval_stmt(query), &state)
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("{query}: {err}"));
|
||||
}
|
||||
|
||||
// The spelling is what is reserved, not the role the column plays: a value column of that
|
||||
// name is refused like a tag column, while a value column under another name is read as the
|
||||
// sample of its series.
|
||||
let err = PromPlanner::stmt_to_plan(
|
||||
build_value_column_table_provider("reserved_field_metric", METRIC_NAME).await,
|
||||
&operator_eval_stmt("reserved_field_metric"),
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert_eq!(
|
||||
err.to_string(),
|
||||
concat!(
|
||||
"Table reserved_field_metric cannot be queried with PromQL: ",
|
||||
"column '__name__' conflicts with the reserved metric name"
|
||||
)
|
||||
);
|
||||
PromPlanner::stmt_to_plan(
|
||||
build_value_column_table_provider("ordinary_field_metric", "value").await,
|
||||
&operator_eval_stmt("ordinary_field_metric"),
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(
|
||||
plan.schema()
|
||||
.field_with_unqualified_name("__name__")
|
||||
.is_ok()
|
||||
);
|
||||
assert!(
|
||||
PromPlanner::metric_name_column(plan.schema())
|
||||
.unwrap()
|
||||
.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 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
|
||||
.unwrap_or_else(|err| panic!("{query}: {err}"));
|
||||
let plan_text = plan.display_indent().to_string();
|
||||
assert!(
|
||||
plan_text.contains(&format!(
|
||||
"Sort: Utf8(\"\") {} NULLS",
|
||||
if function.ends_with("desc") {
|
||||
"DESC"
|
||||
} else {
|
||||
"ASC"
|
||||
}
|
||||
)),
|
||||
"{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)
|
||||
.await
|
||||
.unwrap();
|
||||
let marker = PromPlanner::metric_name_column(plan.schema())
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert!(
|
||||
plan.display_indent()
|
||||
.to_string()
|
||||
.contains(&format!("Sort: coalesce({}, Utf8(\"\"))", marker.name))
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Aggregations follow the metric-name lifecycle: the semantic `__name__` resolves through the
|
||||
@@ -10585,121 +10654,87 @@ async fn test_dropping_an_absent_metric_name_adds_no_check() {
|
||||
}
|
||||
}
|
||||
|
||||
/// A physical column spelled like the semantic `__name__`, or like the internal column the marker
|
||||
/// is attached under, is an ordinary label of its input and never the identity: the planner writes
|
||||
/// the name only through the marked column and reads it only through that mark, so no physical
|
||||
/// column is promoted to the role.
|
||||
/// A physical column spelled like the internal column the marker is attached under is an ordinary
|
||||
/// label of its input and never the identity: the planner writes the name only through the marked
|
||||
/// column and reads it only through that mark, so no physical column is promoted to the role.
|
||||
///
|
||||
/// A destination written as `__name__` next to `without(...)` writes no name at all, and the name
|
||||
/// it was written as is not a grouping label of the result either: the label of that spelling is
|
||||
/// taken over by the destination and dropped with it, while a column of any other spelling stays
|
||||
/// an ordinary label of the result.
|
||||
/// an ordinary label of the result. A column spelled like the *reserved* name never reaches the
|
||||
/// call: its table is refused where the selector enters PromQL.
|
||||
#[tokio::test]
|
||||
async fn test_count_values_semantic_name_next_to_physical_columns() {
|
||||
async fn test_count_values_semantic_name_next_to_the_internal_metric_name_column() {
|
||||
let state = build_query_engine_state();
|
||||
// (physical label column, query)
|
||||
for (tag, query) in [
|
||||
(
|
||||
METRIC_NAME,
|
||||
r#"count_values without(job) ("__name__", abs(cv_metric))"#,
|
||||
),
|
||||
// A physical column occupying the internal name of the marker is not the identity and not
|
||||
// the destination: it stays an ordinary label, is part of the grouping of `without` and
|
||||
// keeps its own values, so the three series stay three groups.
|
||||
(
|
||||
PROMQL_METRIC_NAME_COLUMN,
|
||||
r#"count_values without(job) ("__name__", abs(cv_metric))"#,
|
||||
),
|
||||
] {
|
||||
let plan = PromPlanner::stmt_to_plan(
|
||||
build_count_values_collision_table_provider(tag).await,
|
||||
&operator_eval_stmt(query),
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("{tag}: {query}: {err}"));
|
||||
|
||||
assert!(
|
||||
plan.schema()
|
||||
.fields()
|
||||
.iter()
|
||||
.all(|field| field.metadata().get(PROMQL_FIELD_ROLE_KEY).is_none()),
|
||||
"{tag}: no field may carry the identity role: {}",
|
||||
plan.display_indent()
|
||||
);
|
||||
|
||||
let has_name_column = plan
|
||||
.schema()
|
||||
.fields()
|
||||
.iter()
|
||||
.any(|field| field.name() == METRIC_NAME);
|
||||
let plan_display = plan.display_indent().to_string();
|
||||
let count_column = plan
|
||||
.schema()
|
||||
.fields()
|
||||
.iter()
|
||||
.find(|field| {
|
||||
matches!(
|
||||
field.data_type(),
|
||||
ArrowDataType::Float64 | ArrowDataType::Int64 | ArrowDataType::UInt64
|
||||
)
|
||||
})
|
||||
.expect("no aggregated count column")
|
||||
.name()
|
||||
.clone();
|
||||
let (_, batches) = execute(plan, &state).await;
|
||||
let mut actual_counts = numeric_values(&batches, &count_column);
|
||||
actual_counts.sort_by(f64::total_cmp);
|
||||
if tag == METRIC_NAME {
|
||||
// The destination was written as the only label of the input, and with `without` the
|
||||
// name is not a label of the result: the three samples of one value form one group.
|
||||
assert!(
|
||||
!has_name_column,
|
||||
"{query}: the destination is not a label of the result: {plan_display}"
|
||||
);
|
||||
assert_eq!(actual_counts, vec![3.0], "{tag}: {query}");
|
||||
} else {
|
||||
assert_eq!(actual_counts, vec![1.0, 1.0, 1.0], "{tag}: {query}");
|
||||
let mut tag_values = string_values(&batches, tag)
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.collect::<Vec<_>>();
|
||||
tag_values.sort();
|
||||
assert_eq!(tag_values, vec!["k0", "k1", "k2"], "{tag}: {query}");
|
||||
}
|
||||
}
|
||||
|
||||
// A marked input next to a physical label spelled like the semantic name: the marked column is
|
||||
// the name of the result and the physical label of that spelling is not part of it, so the
|
||||
// three samples - all of value 1 - form one group named `1` instead of three named after their
|
||||
// label.
|
||||
let query = r#"count_values("__name__", cv_metric)"#;
|
||||
let query = r#"count_values without(job) ("__name__", abs(cv_metric))"#;
|
||||
// The physical column is not the identity and not the destination: it stays an ordinary label,
|
||||
// is part of the grouping of `without` and keeps its own values, so the three series stay
|
||||
// three groups.
|
||||
let plan = PromPlanner::stmt_to_plan(
|
||||
build_count_values_collision_table_provider(METRIC_NAME).await,
|
||||
build_count_values_collision_table_provider(PROMQL_METRIC_NAME_COLUMN).await,
|
||||
&operator_eval_stmt(query),
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("{query}: {err}"));
|
||||
let marker = PromPlanner::metric_name_column(plan.schema())
|
||||
.unwrap()
|
||||
.unwrap_or_else(|| panic!("{query}: the generated column must be the name"));
|
||||
|
||||
assert!(
|
||||
!plan
|
||||
.schema()
|
||||
plan.schema()
|
||||
.fields()
|
||||
.iter()
|
||||
.any(|field| field.name() == METRIC_NAME),
|
||||
"{query}: the physical label cannot stay next to the name: {}",
|
||||
.all(|field| field.metadata().get(PROMQL_FIELD_ROLE_KEY).is_none()),
|
||||
"no field may carry the identity role: {}",
|
||||
plan.display_indent()
|
||||
);
|
||||
let count_column = plan
|
||||
.schema()
|
||||
.fields()
|
||||
.iter()
|
||||
.find(|field| {
|
||||
matches!(
|
||||
field.data_type(),
|
||||
ArrowDataType::Float64 | ArrowDataType::Int64 | ArrowDataType::UInt64
|
||||
)
|
||||
})
|
||||
.expect("no aggregated count column")
|
||||
.name()
|
||||
.clone();
|
||||
let (_, batches) = execute(plan, &state).await;
|
||||
assert_metric_name_in_batches(&batches, &marker.name);
|
||||
assert_eq!(
|
||||
count_values_rows(&batches, &marker.name),
|
||||
vec![("1", 3.0)],
|
||||
"{query}"
|
||||
);
|
||||
let mut actual_counts = numeric_values(&batches, &count_column);
|
||||
actual_counts.sort_by(f64::total_cmp);
|
||||
assert_eq!(actual_counts, vec![1.0, 1.0, 1.0], "{query}");
|
||||
let mut tag_values = string_values(&batches, PROMQL_METRIC_NAME_COLUMN)
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.collect::<Vec<_>>();
|
||||
tag_values.sort();
|
||||
assert_eq!(tag_values, vec!["k0", "k1", "k2"], "{query}");
|
||||
}
|
||||
|
||||
/// A table that brings its own `__name__` column has no way into PromQL evaluation, so a
|
||||
/// `count_values` never sees one: the selector inside the call is refused with its table, before a
|
||||
/// destination could take the reserved name over or leave it behind.
|
||||
#[tokio::test]
|
||||
async fn test_count_values_rejects_a_table_with_a_reserved_name_column() {
|
||||
let state = build_query_engine_state();
|
||||
for query in [
|
||||
r#"count_values("__name__", cv_metric)"#,
|
||||
r#"count_values without(job) ("__name__", abs(cv_metric))"#,
|
||||
] {
|
||||
let err = PromPlanner::stmt_to_plan(
|
||||
build_count_values_collision_table_provider(METRIC_NAME).await,
|
||||
&operator_eval_stmt(query),
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
err.to_string()
|
||||
.contains("column '__name__' conflicts with the reserved metric name"),
|
||||
"{query}: {err}"
|
||||
);
|
||||
assert_eq!(err.status_code(), StatusCode::InvalidArguments, "{query}");
|
||||
}
|
||||
}
|
||||
|
||||
/// A `count_values` destination is a label of the result and the label takes the physical name it
|
||||
@@ -12102,114 +12137,47 @@ async fn test_internal_metric_name_column_is_not_an_ordinary_label() {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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.
|
||||
/// A table that brings its own `__name__` column cannot be labelled: a label function reading the
|
||||
/// semantic name would read it out of that column, and one writing it would have the column and
|
||||
/// the name of the series at once. The table is refused where the selector enters PromQL, so the
|
||||
/// label functions never see it - the same three queries read the ordinary `cv_metric` table
|
||||
/// through its marked identity (`test_label_functions_semantic_name_source_into_ordinary_label`,
|
||||
/// `test_label_functions_semantic_name_destination_overwrites_identity`).
|
||||
#[tokio::test]
|
||||
async fn test_label_functions_semantic_name_keeps_physical_name_label() {
|
||||
async fn test_label_functions_reject_a_table_with_a_reserved_name_column() {
|
||||
let state = build_query_engine_state();
|
||||
// (query, ordinary destination to check with its expected values, expected identity values)
|
||||
for (query, ordinary, expected_identity) in [
|
||||
(
|
||||
// The semantic source is the identity, never the physical `__name__` label.
|
||||
r#"label_replace(cv_metric, "name_copy", "$1", "__name__", "(.*)")"#,
|
||||
Some(("name_copy", vec!["cv_metric", "cv_metric", "cv_metric"])),
|
||||
vec!["cv_metric", "cv_metric", "cv_metric"],
|
||||
),
|
||||
(
|
||||
// ... and the semantic destination replaces the identity (matching regex) ...
|
||||
r#"label_replace(cv_metric, "__name__", "renamed", "__name__", "(.*)")"#,
|
||||
None,
|
||||
vec!["renamed", "renamed", "renamed"],
|
||||
),
|
||||
(
|
||||
// ... or keeps it (non-matching regex), including for `label_join`.
|
||||
r#"label_join(cv_metric, "__name__", "/", "__name__", "missing_label")"#,
|
||||
None,
|
||||
vec!["cv_metric/", "cv_metric/", "cv_metric/"],
|
||||
),
|
||||
for query in [
|
||||
// A semantic source ...
|
||||
r#"label_replace(cv_metric, "name_copy", "$1", "__name__", "(.*)")"#,
|
||||
// ... a semantic destination ...
|
||||
r#"label_replace(cv_metric, "__name__", "renamed", "__name__", "(.*)")"#,
|
||||
// ... and a semantic source of `label_join`.
|
||||
r#"label_join(cv_metric, "__name__", "/", "__name__", "missing_label")"#,
|
||||
] {
|
||||
let plan = PromPlanner::stmt_to_plan(
|
||||
let err = PromPlanner::stmt_to_plan(
|
||||
// The metric carries a physical label named `__name__`.
|
||||
build_count_values_collision_table_provider(METRIC_NAME).await,
|
||||
&operator_eval_stmt(query),
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("{query}: {err}"));
|
||||
|
||||
let marker = PromPlanner::metric_name_column(plan.schema())
|
||||
.unwrap()
|
||||
.unwrap_or_else(|| panic!("{query}: the identity must be attached"));
|
||||
assert_eq!(marker.name, PROMQL_METRIC_NAME_COLUMN, "{query}");
|
||||
let physical = plan
|
||||
.schema()
|
||||
.field_with_unqualified_name(METRIC_NAME)
|
||||
.unwrap_or_else(|_| panic!("{query}: the physical `__name__` label must survive"));
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
physical.metadata().get(PROMQL_FIELD_ROLE_KEY).is_none(),
|
||||
"{query}: the physical `__name__` label must stay unmarked",
|
||||
err.to_string()
|
||||
.contains("column '__name__' conflicts with the reserved metric name"),
|
||||
"{query}: {err}"
|
||||
);
|
||||
assert_eq!(err.status_code(), StatusCode::InvalidArguments, "{query}");
|
||||
|
||||
let (_, batches) = execute(plan, &state).await;
|
||||
assert_metric_name_in_batches(&batches, &marker.name);
|
||||
// Every row keeps the physical label and the sample value of its series.
|
||||
let mut rows = batches
|
||||
.iter()
|
||||
.flat_map(|batch| {
|
||||
let physical = 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");
|
||||
let identity = batch
|
||||
.column_by_name(&marker.name)
|
||||
.expect("the identity must be projected")
|
||||
.as_any()
|
||||
.downcast_ref::<StringArray>()
|
||||
.expect("the identity must be a string column");
|
||||
let value = batch
|
||||
.column_by_name(greptime_value())
|
||||
.expect("the sample value must be projected")
|
||||
.as_any()
|
||||
.downcast_ref::<Float64Array>()
|
||||
.expect("the sample value must be a float column");
|
||||
(0..batch.num_rows())
|
||||
.map(|row| {
|
||||
(
|
||||
physical.value(row).to_string(),
|
||||
identity.value(row).to_string(),
|
||||
value.value(row),
|
||||
)
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
rows.sort_by(|left, right| left.0.cmp(&right.0));
|
||||
assert_eq!(
|
||||
rows,
|
||||
["k0", "k1", "k2"]
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(index, label)| (
|
||||
label.to_string(),
|
||||
expected_identity[index].to_string(),
|
||||
1.0
|
||||
))
|
||||
.collect::<Vec<_>>(),
|
||||
"{query}"
|
||||
);
|
||||
if let Some((column, expected)) = ordinary {
|
||||
assert_eq!(
|
||||
string_values(&batches, column),
|
||||
expected
|
||||
.iter()
|
||||
.map(|value| Some(value.to_string()))
|
||||
.collect::<Vec<_>>(),
|
||||
"{query}"
|
||||
);
|
||||
}
|
||||
// The same query reads the ordinary table: what is refused is the table's column, not the
|
||||
// `__name__` the query asks about.
|
||||
PromPlanner::stmt_to_plan(
|
||||
build_count_values_table_provider().await,
|
||||
&operator_eval_stmt(query),
|
||||
&state,
|
||||
)
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("{query}: {err}"));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -431,25 +431,12 @@ impl ColumnLayout {
|
||||
.fail();
|
||||
}
|
||||
|
||||
// Only the marked identity supplies `__name__`; retain an ordinary physical
|
||||
// column under an unoccupied label so distinct series are never merged.
|
||||
// Only the marked identity supplies `__name__`. A table whose own schema holds a column
|
||||
// of that spelling is refused before PromQL evaluation, so an unmarked column of the
|
||||
// reserved name cannot reach this layout.
|
||||
let tag_names = tag_column_indices
|
||||
.iter()
|
||||
.map(|index| {
|
||||
let name = schema.column_name_by_index(*index);
|
||||
if name != METRIC_NAME {
|
||||
return name.to_string();
|
||||
}
|
||||
let mut exported = "exported_name".to_string();
|
||||
while schema
|
||||
.column_schemas()
|
||||
.iter()
|
||||
.any(|column| column.name == exported)
|
||||
{
|
||||
exported.push('_');
|
||||
}
|
||||
exported
|
||||
})
|
||||
.map(|index| schema.column_name_by_index(*index).to_string())
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
Ok(Self {
|
||||
@@ -843,154 +830,24 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn marked_metric_name_overrides_ordinary_name_tag() {
|
||||
// The marked column is the renamed physical identity; the ordinary `__name__`
|
||||
// label column must not override it.
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
fn column_layout_keeps_tag_names_verbatim() {
|
||||
// The layout reads the name of every tag column as it is. A column
|
||||
// spelled like the reserved name is not worked around either: a table
|
||||
// that brings one is refused where it enters PromQL, so no unmarked
|
||||
// column of that spelling can reach this layout.
|
||||
let schema = Schema::new(vec![
|
||||
ColumnSchema::new(
|
||||
"timestamp",
|
||||
ConcreteDataType::timestamp_millisecond_datatype(),
|
||||
false,
|
||||
),
|
||||
ColumnSchema::new("host", ConcreteDataType::string_datatype(), true),
|
||||
ColumnSchema::new("__name__", ConcreteDataType::string_datatype(), true),
|
||||
marked_metric_name_column("renamed_identity"),
|
||||
ColumnSchema::new(METRIC_NAME, ConcreteDataType::string_datatype(), true),
|
||||
ColumnSchema::new("value", ConcreteDataType::float64_datatype(), false),
|
||||
]));
|
||||
let batch = RecordBatch::new(
|
||||
schema.clone(),
|
||||
vec![
|
||||
Arc::new(TimestampMillisecondVector::from_values([1_000])) as _,
|
||||
Arc::new(StringVector::from(vec![Some("host-a")])) as _,
|
||||
Arc::new(StringVector::from(vec![Some("ordinary_name")])) as _,
|
||||
Arc::new(StringVector::from(vec![Some("physical_name")])) as _,
|
||||
Arc::new(Float64Vector::from_values([1.0])) as _,
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let response = PrometheusJsonResponse::record_batches_to_data(
|
||||
RecordBatches::try_new(schema, vec![batch]).unwrap(),
|
||||
ValueType::Vector,
|
||||
)
|
||||
.unwrap();
|
||||
let PrometheusResponse::PromData(PromData {
|
||||
result: PromQueryResult::Vector(series),
|
||||
..
|
||||
}) = response
|
||||
else {
|
||||
panic!("expected vector response");
|
||||
};
|
||||
assert_eq!(series.len(), 1);
|
||||
assert_eq!(
|
||||
series[0].metric.get(METRIC_NAME).map(String::as_str),
|
||||
Some("physical_name")
|
||||
);
|
||||
assert_eq!(
|
||||
series[0].metric.get("host").map(String::as_str),
|
||||
Some("host-a")
|
||||
);
|
||||
assert_eq!(
|
||||
series[0].metric.get("exported_name").map(String::as_str),
|
||||
Some("ordinary_name")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unmarked_ordinary_name_never_restores_metric_identity() {
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
ColumnSchema::new(
|
||||
"timestamp",
|
||||
ConcreteDataType::timestamp_millisecond_datatype(),
|
||||
false,
|
||||
),
|
||||
ColumnSchema::new("__name__", ConcreteDataType::string_datatype(), true),
|
||||
ColumnSchema::new("exported_name", ConcreteDataType::string_datatype(), true),
|
||||
ColumnSchema::new("value", ConcreteDataType::float64_datatype(), false),
|
||||
]));
|
||||
let batch = RecordBatch::new(
|
||||
schema.clone(),
|
||||
vec![
|
||||
Arc::new(TimestampMillisecondVector::from_values([1_000, 1_000])) as _,
|
||||
Arc::new(StringVector::from(vec![
|
||||
Some("physical_a"),
|
||||
Some("physical_b"),
|
||||
])) as _,
|
||||
Arc::new(StringVector::from(vec![Some("other"), Some("other")])) as _,
|
||||
Arc::new(Float64Vector::from_values([1.0, 2.0])) as _,
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
for result_type in [ValueType::Vector, ValueType::Matrix] {
|
||||
let buffered = PrometheusJsonResponse::record_batches_to_data(
|
||||
RecordBatches::try_new(schema.clone(), vec![batch.clone()]).unwrap(),
|
||||
result_type,
|
||||
)
|
||||
.unwrap();
|
||||
let streamed = PrometheusJsonResponse::consume_stream_to_data(
|
||||
RecordBatches::try_new(
|
||||
schema.clone(),
|
||||
vec![batch.slice(0, 1).unwrap(), batch.slice(1, 1).unwrap()],
|
||||
)
|
||||
.unwrap()
|
||||
.as_stream(),
|
||||
result_type,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
serde_json::to_value(&buffered).unwrap(),
|
||||
serde_json::to_value(&streamed).unwrap()
|
||||
);
|
||||
let PrometheusResponse::PromData(PromData { result, .. }) = buffered else {
|
||||
panic!("expected Prometheus data");
|
||||
};
|
||||
let (metrics, values) = match &result {
|
||||
PromQueryResult::Vector(series) => (
|
||||
series
|
||||
.iter()
|
||||
.map(|series| &series.metric)
|
||||
.collect::<Vec<_>>(),
|
||||
series
|
||||
.iter()
|
||||
.map(|series| series.value.as_ref().unwrap().1.clone())
|
||||
.collect::<Vec<_>>(),
|
||||
),
|
||||
PromQueryResult::Matrix(series) => (
|
||||
series
|
||||
.iter()
|
||||
.map(|series| &series.metric)
|
||||
.collect::<Vec<_>>(),
|
||||
series
|
||||
.iter()
|
||||
.map(|series| {
|
||||
serde_json::to_value(&series.values[0].1)
|
||||
.unwrap()
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_string()
|
||||
})
|
||||
.collect::<Vec<_>>(),
|
||||
),
|
||||
_ => panic!("expected vector or matrix"),
|
||||
};
|
||||
assert_eq!(metrics.len(), 2);
|
||||
for metric in &metrics {
|
||||
assert!(!metric.contains_key(METRIC_NAME));
|
||||
assert_eq!(
|
||||
metric.get("exported_name").map(String::as_str),
|
||||
Some("other")
|
||||
);
|
||||
}
|
||||
assert_eq!(
|
||||
metrics[0].get("exported_name_").map(String::as_str),
|
||||
Some("physical_a")
|
||||
);
|
||||
assert_eq!(
|
||||
metrics[1].get("exported_name_").map(String::as_str),
|
||||
Some("physical_b")
|
||||
);
|
||||
assert_eq!(values, vec!["1.0", "2.0"]);
|
||||
}
|
||||
]);
|
||||
let layout = ColumnLayout::infer(&schema).unwrap();
|
||||
assert!(layout.metric_name_column_index.is_none());
|
||||
assert_eq!(layout.tag_names, vec!["host", METRIC_NAME]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
Reference in New Issue
Block a user