rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 06:46:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(promql): support metric-name rewrites in native labels (#10176)

Browse Source 
Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-11-11 11:40:40 +08:00
committed by GitHub
parent 8004f3e88d
commit e49bb09480
6 changed files with 1058 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/src/map_label.rs
+14
View File
@@ -255,6 +255,20 @@ impl LabelRewrite {
}
}
/// Whether this change can write a nonempty value.
///
/// An empty value removes the destination instead ([`set_label`]), so a change whose value is
/// necessarily empty only deletes a label and publishes none that its input did not carry.
/// `label_replace`'s template is judged as written: `$1` can expand to nothing for one series and
/// to a value for another, so it may write. A `label_join` with no source labels joins nothing
/// and is always empty.
pub fn writes_value(&self) -> bool {
match &self.0 {
Change::Replace { replacement, .. } => !replacement.is_empty(),
Change::Join { src, .. } => !src.is_empty(),
}
}
/// `label_replace`'s change, failing on an invalid expression or an empty destination.
pub fn replace(dst: &str, replacement: &str, src: &str, regex: &str) -> Result<Self> {
let compiled = Regex::new(&format!("^(?s:{regex})$")).map_err(|_| {
src/promql/src/native/plan.rs
+271 -175
View File
@@ -55,11 +55,24 @@ use crate::vector_match::{
/// its result does not report.
const LE_LABEL: &str = "le";
/// The label a metric name is stored under, which a scan binds beside its tag columns rather than
/// among them.
const METRIC_NAME: &str = "__name__";
/// Metadata labels the runtime removes from histogram result label sets.
const PROMQL_METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"];
/// The label columns `node` reports, from the tags of the scan each input reads and the labels the
/// nodes above it write.
/// The complete label set `node` reports at run time, from the labels each input reports and the
/// ones the nodes above it write.
///
/// This is the identity a [`NativeExpr::Merge`] groups by and a vector match reads, so the metric
/// name is part of it whenever the series carries one. The caller's `source_tags` are the complete
/// label sets of the sources, indexed by the `source_index` of the leaf that reads each: a scan
/// bound to a single metric names every series it hands over even though the name is not one of its
/// tag columns, so the canonical pass includes `__name__` in that source's entry. The query planner
/// derives the *output columns* with its own walk, which reads the same algebra from the tag columns
/// each source declares; the two have to agree on where a kernel drops the name, which is why this
/// walk drops it exactly where the runtime does.
///
/// They follow Prometheus' result labels: a one-to-one match keeps the left side's labels, narrowed
/// by `on` or `ignoring`; `and` and `unless` keep the left side's series and `or` adds the right
@@ -67,41 +80,76 @@ const PROMQL_METADATA_LABELS: [&str; 3] = ["__name__", "__type__", "__unit__"];
/// of each value keeps its operand's; a scalar and `vector(s)` have none; `count_values` keeps its
/// grouping and the label it writes; a fill that answers a missing many-side sample files the one
/// side's matching labels, so they join the set, while a grouped `include` label joins them only when
/// the one side carries it or the matching keeps it in the key a fill reads. `None` for a node whose
/// labels this walk does not derive - a label rewrite of the metric name, an `absent` naming a label
/// no input column carries, and a `count_values` whose destination is the metric name or no label at
/// all, none of which is a column here - so a tree holding one reports no label set here. `None` is
/// not the query's answer: the caller asks the existing planner for the columns the node is named
/// after instead of declining the query.
/// the one side carries it or the matching keeps it in the key a fill reads. A label rewrite writes
/// its destination, and a destination that names the metric name writes that name.
///
/// A node that reports a value of its own drops the name where its kernel does - `timestamp`, a
/// range function whose result is computed (`rate`, `increase`, ...), a math or histogram map,
/// `histogram_quantile`, every non-filtering binary - so a name rewrite below one of them cannot
/// make the name reappear above it.
///
/// `None` for a node whose labels this walk does not derive - a `count_values` whose destination is
/// empty. `None` proves nothing, so the caller treats it as "may alias" rather than as "distinct".
pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<Vec<String>> {
match &node.expr {
NativeExpr::Input(input) => source_tags.get(input.source_index).cloned(),
NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } | NativeExpr::VectorOf { .. } => {
Some(Vec::new())
}
NativeExpr::Instant { input, .. }
| NativeExpr::Window { input, .. }
| NativeExpr::Timestamp { input } => fused_output_tags(input, source_tags),
// These report a value of their own rather than the sample they read, and the runtime drops
// the name for each of them. `source_tags` is the authority on what an input carries, so a
// discovery input arrives carrying `__name__` and these arms are what decides it does not
// survive.
NativeExpr::Instant { input, op } => {
let tags = fused_output_tags(input, source_tags)?;
Some(match op.value {
InstantValue::Timestamp => without_metric_name(tags),
InstantValue::Sample | InstantValue::InfoTimestamp => tags,
})
}
NativeExpr::Window { input, op } => {
let tags = fused_output_tags(input, source_tags)?;
Some(if op.function.drops_metric_name() {
without_metric_name(tags)
} else {
tags
})
}
NativeExpr::Timestamp { input } => {
Some(without_metric_name(fused_output_tags(input, source_tags)?))
}
// A label rewrite changes the label set itself: `label_replace` adds its destination label
// and an empty replacement removes it, so the input's labels are not the output's. A map of
// math alone keeps them.
// and an empty replacement removes it, so the input's labels are not the output's. A math or
// histogram kernel drops the metric name where it runs, and the ops apply in written order,
// so what follows a rewrite sees the set the rewrite left.
NativeExpr::Map { input, op } => {
let mut tags = fused_output_tags(input, source_tags)?;
for operation in op.ops.iter() {
if let MapOp::Label(rewrite) = operation {
let destination = rewrite.destination();
if destination == "__name__" {
return None;
match operation {
MapOp::Label(rewrite) => {
let destination = rewrite.destination();
if tags.iter().any(|tag| tag == destination) {
// The input carries the label, so the rewrite changes its value - and an
// empty replacement may remove it, which the column still represents.
tags.retain(|tag| tag != destination);
tags.push(destination.to_string());
} else if rewrite.writes_value() {
// A fresh destination is a label the output only carries when the
// rewrite can write a value; a delete-only rewrite of a label no input
// carried publishes nothing.
tags.push(destination.to_string());
}
}
tags.retain(|tag| tag != destination);
tags.push(destination.to_string());
MapOp::Math(_) | MapOp::Histogram(_) => tags = without_metric_name(tags),
}
}
Some(tags)
}
// Its parameters are scalars, which carry no labels, so what the node reports are its
// input's labels.
NativeExpr::ParameterizedMap { input, .. } => fused_output_tags(input, source_tags),
// Its parameters are scalars, which carry no labels; the map computes a value of its own
// with the input, and its kernel drops the metric name.
NativeExpr::ParameterizedMap { input, .. } => {
Some(without_metric_name(fused_output_tags(input, source_tags)?))
}
// `scalar()` reports one value and no labels, like the scalar it produces.
NativeExpr::ScalarOf { .. } => Some(Vec::new()),
// Its parameter is a scalar, which carries no labels, and ranking keeps each selected
@@ -117,13 +165,9 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Opti
}
// An aggregate reports the grouping labels its input carries: `by(...)` keeps exactly the
// ones it names, and `without(...)` keeps the rest and drops the metric name, as Prometheus
// does.
//
// What this derives is the set of *output columns*, which is what the node's schema needs.
// It is not the complete runtime label set: a chunk's labels also carry the metric name,
// which is bound separately from these tag columns rather than being one of them. A
// grouping by `__name__` keeps that name in the labels while contributing no column here,
// and `sum by (__name__)` over one metric is still one column set.
// does. A grouping by `__name__` keeps the name where the input reports one; a named scan
// only reports it among the runtime labels, so the aggregate keeps it there and publishes no
// column for it, which is what `sum by (__name__)` over one metric reports.
NativeExpr::Aggregate { input, op } => {
let input_tags = fused_output_tags(input, source_tags)?;
Some(match &op.group {
@@ -134,7 +178,7 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Opti
Grouping::Without(dropped) => input_tags
.into_iter()
.filter(|tag| {
tag != "__name__"
tag != METRIC_NAME
&& !dropped.iter().any(|name| name.as_ref() == tag.as_str())
})
.collect(),
@@ -148,63 +192,51 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Opti
| BoundVectorOp::LimitK { .. }
| BoundVectorOp::LimitRatio { .. }
| BoundVectorOp::Sort(_) => fused_output_tags(input, source_tags),
BoundVectorOp::Absent { labels } => {
let input_tags = fused_output_tags(input, source_tags)?;
let named = labels
// `absent` reports the labels its own matchers name, not the input's set and not the
// table's columns: it answers one series carrying exactly those, so a label the input
// does not carry is still a label of the result.
BoundVectorOp::Absent { labels } => Some(
labels
.tags()
.iter()
.map(|label| label.name.as_ref())
.collect::<Vec<_>>();
// `absent` reports matcher-named labels, not all input labels or the table's columns;
// only named labels present as input columns can be represented as frame columns.
if named
.iter()
.any(|name| !input_tags.iter().any(|tag| tag == name))
{
return None;
}
Some(
input_tags
.into_iter()
.filter(|tag| named.contains(&tag.as_str()))
.collect(),
)
}
.map(|label| label.name.to_string())
.collect(),
),
// `count_values` writes the value it counts into its own label and then keeps its
// grouping: `by(...)` keeps the labels named plus that destination, `without(...)`
// keeps the rest minus the labels named and the metric name. What the node reports is
// therefore the grouping applied to the labels its input carries, with the destination
// among them - not the input's set, and not the scan's, since the grouping narrows both.
//
// The destination is a column the node creates, so a rewrite of it above this node is a
// rewrite of an existing label. A destination naming the metric name, or no label at
// all, writes something that is not a column here, which leaves the set underivable for
// the same reason a map's `__name__` destination does. This walk derives the labels; it
// does not decide which groupings the kernel accepts, and those refusals stay where they
// are.
BoundVectorOp::CountValues { label, group }
if !label.is_empty() && label.as_ref() != "__name__" =>
{
// The destination is a label the node creates, so a rewrite of it above this node is a
// rewrite of an existing label. A destination naming the metric name writes the name
// the node groups by, which is the one label its result carries. This walk derives the
// labels; it does not decide which groupings the kernel accepts, and those refusals stay
// where they are.
BoundVectorOp::CountValues { label, group } if !label.is_empty() => {
let mut tags = fused_output_tags(input, source_tags)?;
let keeps_destination = match group {
Grouping::By(kept) => {
tags.retain(|tag| kept.iter().any(|name| name.as_ref() == tag.as_str()));
// `by` keeps the label the node writes whether the input carried it or not.
true
}
Grouping::Without(dropped) => {
tags.retain(|tag| {
tag != "__name__"
&& !dropped.iter().any(|name| name.as_ref() == tag.as_str())
});
// `without` removes it only when the grouping names it.
!dropped.iter().any(|name| name.as_ref() == label.as_ref())
}
};
if keeps_destination && !tags.iter().any(|tag| tag.as_str() == label.as_ref()) {
if !tags.contains(&label.to_string()) {
tags.push(label.to_string());
}
Some(tags)
Some(match group {
Grouping::By(names) => {
let mut output = tags
.into_iter()
.filter(|tag| names.iter().any(|name| name.as_ref() == tag))
.collect::<Vec<_>>();
// `by` keeps the label the node writes whether the input carried it or not.
if !output.contains(&label.to_string()) {
output.push(label.to_string());
}
output
}
Grouping::Without(names) => tags
.into_iter()
.filter(|tag| {
tag != METRIC_NAME && !names.iter().any(|name| name.as_ref() == tag)
})
.collect(),
})
}
BoundVectorOp::CountValues { .. } => None,
// A classic histogram function folds the bucket series that share the other labels into
@@ -223,105 +255,133 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Opti
left,
right,
matching,
} => match (&left.output, &right.output) {
(NativeType::Scalar(_), _) => fused_output_tags(right, source_tags),
(_, NativeType::Scalar(_)) => fused_output_tags(left, source_tags),
_ => {
let kept = |tags: Vec<String>| -> Vec<String> {
match &matching.labels {
MatchingLabels::On(names) => tags
.into_iter()
.filter(|tag| names.iter().any(|name| name.as_ref() == tag.as_str()))
.collect(),
MatchingLabels::Ignoring(names) => tags
.into_iter()
.filter(|tag| !names.iter().any(|name| name.as_ref() == tag.as_str()))
.collect(),
}
};
let left_tags = fused_output_tags(left, source_tags)?;
let right_tags = fused_output_tags(right, source_tags)?;
// One-to-one matching narrows the left side's labels with `on`/`ignoring`.
// A group retains the many side's labels and copies the named labels from the one
// side: `group_left(container)` keeps the left labels and adds `container` from
// the right.
let mut tags = match &matching.cardinality {
Cardinality::OneToOne => kept(left_tags.clone()),
Cardinality::ManyToOne(_) => left_tags.clone(),
Cardinality::OneToMany(_) => right_tags.clone(),
};
let (includes, include_tags): (&[Arc<str>], &Vec<String>) =
match &matching.cardinality {
Cardinality::OneToOne => (&[], &left_tags),
Cardinality::ManyToOne(labels) => (labels, &right_tags),
Cardinality::OneToMany(labels) => (labels, &left_tags),
};
for tag in include_tags {
if includes.iter().any(|label| label.as_ref() == tag.as_str())
&& !tags.contains(tag)
{
tags.push(tag.clone());
}
}
// A grouped result carries an include label only when a row can give it one.
// `result_metric` removes the label from the many side's labels unless the one side
// has it, so a paired row and a row filed from a one-side series take it from the
// one series, which is what the loop above declares. A row filed from a many-side
// series takes it from that series' match key instead, so it carries the label only
// when the matching keeps the label in the key. Everything else the many side
// carries is the base above, so a label neither of those can provide has to leave
// the set: nothing would ever fill its column.
if !includes.is_empty() {
let many_rows_filled = match &matching.cardinality {
// `group_right` puts the one side on the left, so the fills trade places:
// `fill_left` is the one that answers a missing one-side sample there.
Cardinality::OneToMany(_) => matching.fill.lhs,
Cardinality::ManyToOne(_) | Cardinality::OneToOne => matching.fill.rhs,
}
.is_some();
let in_match_key = |label: &str| match &matching.labels {
MatchingLabels::On(names) => {
names.iter().any(|name| name.as_ref() == label)
}
// A match key is built from a series' tags, which never hold the metric name.
MatchingLabels::Ignoring(names) => {
label != "__name__" && !names.iter().any(|name| name.as_ref() == label)
} => {
let keeps_name = keeps_metric_name(matching);
let scalar_side = matches!(&left.output, NativeType::Scalar(_))
|| matches!(&right.output, NativeType::Scalar(_));
let tags = match (&left.output, &right.output) {
(NativeType::Scalar(_), _) => fused_output_tags(right, source_tags)?,
(_, NativeType::Scalar(_)) => fused_output_tags(left, source_tags)?,
_ => {
let kept = |tags: Vec<String>| -> Vec<String> {
match &matching.labels {
MatchingLabels::On(names) => tags
.into_iter()
.filter(|tag| {
names.iter().any(|name| name.as_ref() == tag.as_str())
})
.collect(),
MatchingLabels::Ignoring(names) => tags
.into_iter()
.filter(|tag| {
!names.iter().any(|name| name.as_ref() == tag.as_str())
})
.collect(),
}
};
tags.retain(|tag| {
// The name is not an ordinary label here: this walk reports it by
// `__name__` in the set, which the nodes above decide.
tag == "__name__"
|| !includes.iter().any(|label| label.as_ref() == tag.as_str())
|| include_tags.contains(tag)
|| (many_rows_filled && in_match_key(tag))
});
}
// A fill answers a missing sample from the series of the other side, and the kernel
// files that row at the label set the series it has gives it
// (`vector_match::binary_field`): a many-side series keeps all of its labels, which
// `tags` already holds, while a one-side series keeps only its matching labels plus
// the group labels the modifier names, which the loop above adds. So what a fill can
// add here is the one side's matching labels, and only a fill that can file a
// one-side row has any to add: one where the many side is the missing side. Which
// fill that is follows the kernel, which trades the two for `group_right`, whose one
// side is the left - there it is `fill_right` that leaves the many side missing.
let (one_tags, fills_many_side) = match &matching.cardinality {
Cardinality::OneToMany(_) => (&left_tags, matching.fill.rhs),
Cardinality::ManyToOne(_) | Cardinality::OneToOne => {
(&right_tags, matching.fill.lhs)
let left_tags = fused_output_tags(left, source_tags)?;
let right_tags = fused_output_tags(right, source_tags)?;
// One-to-one matching narrows the left side's labels with `on`/`ignoring`.
// A group retains the many side's labels and copies the named labels from the one
// side: `group_left(container)` keeps the left labels and adds `container` from
// the right.
let mut tags = match &matching.cardinality {
Cardinality::OneToOne => kept(left_tags.clone()),
Cardinality::ManyToOne(_) => left_tags.clone(),
Cardinality::OneToMany(_) => right_tags.clone(),
};
// `result_metric` drops the metadata labels before it matches or copies an
// included label, so an arithmetic result starts without the name. An include can
// still put it back from the one side, which is why the final rule below does not
// run on this branch.
if !keeps_name {
tags.retain(|tag| tag != METRIC_NAME);
}
};
if fills_many_side.is_some() {
for tag in kept(one_tags.clone()) {
if !tags.contains(&tag) {
tags.push(tag);
let (includes, include_tags): (&[Arc<str>], &Vec<String>) =
match &matching.cardinality {
Cardinality::OneToOne => (&[], &left_tags),
Cardinality::ManyToOne(labels) => (labels, &right_tags),
Cardinality::OneToMany(labels) => (labels, &left_tags),
};
for tag in include_tags {
if includes.iter().any(|label| label.as_ref() == tag.as_str())
&& !tags.contains(tag)
{
tags.push(tag.clone());
}
}
// A grouped result carries an include label only when a row can give it one.
// `result_metric` removes the label from the many side's labels unless the one side
// has it, so a paired row and a row filed from a one-side series take it from the
// one series, which is what the loop above declares. A row filed from a many-side
// series takes it from that series' match key instead, so it carries the label only
// when the matching keeps the label in the key. Everything else the many side
// carries is the base above, so a label neither of those can provide has to leave
// the set: nothing would ever fill its column.
if !includes.is_empty() {
let many_rows_filled = match &matching.cardinality {
// `group_right` puts the one side on the left, so the fills trade places:
// `fill_left` is the one that answers a missing one-side sample there.
Cardinality::OneToMany(_) => matching.fill.lhs,
Cardinality::ManyToOne(_) | Cardinality::OneToOne => matching.fill.rhs,
}
.is_some();
let in_match_key = |label: &str| match &matching.labels {
MatchingLabels::On(names) => {
names.iter().any(|name| name.as_ref() == label)
}
// A match key is built from a series' tags, which never hold the metric name.
MatchingLabels::Ignoring(names) => {
label != "__name__"
&& !names.iter().any(|name| name.as_ref() == label)
}
};
tags.retain(|tag| {
!includes.iter().any(|label| label.as_ref() == tag.as_str())
|| include_tags.contains(tag)
|| (many_rows_filled && in_match_key(tag))
});
}
// A fill answers a missing sample from the series of the other side, and the kernel
// files that row at the label set the series it has gives it
// (`vector_match::binary_field`): a many-side series keeps all of its labels, which
// `tags` already holds, while a one-side series keeps only its matching labels plus
// the group labels the modifier names, which the loop above adds. So what a fill can
// add here is the one side's matching labels, and only a fill that can file a
// one-side row has any to add: one where the many side is the missing side. Which
// fill that is follows the kernel, which trades the two for `group_right`, whose one
// side is the left - there it is `fill_right` that leaves the many side missing.
let (one_tags, fills_many_side) = match &matching.cardinality {
Cardinality::OneToMany(_) => (&left_tags, matching.fill.rhs),
Cardinality::ManyToOne(_) | Cardinality::OneToOne => {
(&right_tags, matching.fill.lhs)
}
};
// A filled row keeps only the one side's matching labels, so the name can reach
// it only through an `on` key that names the metric name, and only while the
// operation reports its samples' names.
let fill_carries_name = keeps_name
&& matches!(&matching.labels, MatchingLabels::On(names)
if names.iter().any(|name| name.as_ref() == METRIC_NAME));
if fills_many_side.is_some() {
for tag in kept(one_tags.clone()) {
if tag == METRIC_NAME && !fill_carries_name {
continue;
}
if !tags.contains(&tag) {
tags.push(tag);
}
}
}
tags
}
Some(tags)
}
},
};
Some(if scalar_side && !keeps_name {
without_metric_name(tags)
} else {
tags
})
}
NativeExpr::Set { left, right, op } => {
let mut tags = fused_output_tags(left, source_tags)?;
if op.op == SetOp::Or {
@@ -355,6 +415,23 @@ pub fn fused_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Opti
}
}
}
/// Whether a binary operation reports the samples of the side it keeps, name included.
///
/// A filtering comparison reports those samples, so the name survives it. Every other binary
/// operation computes new values, and `bool` reports the comparison's own outcome rather than
/// either operand's samples, so both drop the name.
fn keeps_metric_name(matching: &BinaryMatching) -> bool {
matches!(
matching.op,
BinaryOp::Eq | BinaryOp::Ne | BinaryOp::Gt | BinaryOp::Lt | BinaryOp::Ge | BinaryOp::Le
) && !matching.return_bool
}
/// The labels a node reports after the metric name is dropped, which is what a kernel that computes
/// a value of its own does to the samples it reports.
fn without_metric_name(tags: Vec<String>) -> Vec<String> {
tags.into_iter().filter(|tag| tag != METRIC_NAME).collect()
}
/// What one node produces.
///
@@ -1012,10 +1089,21 @@ pub struct NativeNode {
///
/// The rewrite is judged by the labels it writes and the ones its input carries, which is why the
/// caller supplies what it knows about each source's tags, indexed by the `source_index` the leaves
/// were bound to: a destination that names no label of the input writes a fresh one, and a fresh
/// label cannot make two series equal. That is the difference between `label_replace(x, "dst", "$1",
/// "host", "(.*)")` - which no merge has to collect - and `label_replace(x, "host", "same", "host",
/// ".*")`, which folds every series of `x` into one label set.
/// were bound to: a destination that names no ordinary label of the input writes a fresh one, and a
/// fresh ordinary label cannot make two series equal. That is the difference between
/// `label_replace(x, "dst", "$1", "host", "(.*)")` - which no merge has to collect - and
/// `label_replace(x, "host", "same", "host", ".*")`, which folds every series of `x` into one label
/// set.
///
/// A source's entry is its complete runtime label set, so `__name__` is one of its labels whenever
/// its series carry a name - a scan bound to one metric names every series from its binding. That is
/// what makes `label_replace(x, "__name__", "same", "__name__", ".*")` a rewrite of an existing
/// label: two metrics whose ordinary labels are equal can be told apart by name alone, and rewriting
/// that name folds them into one identity. A name write is collected even when the derived set no
/// longer carries `__name__` - after `abs`, a `timestamp` or a range function dropped it - because
/// the scan below those kernels can still hold several metrics whose ordinary labels are equal, and
/// writing one name puts them under one identity; the walk has no proof that the input reads a
/// single metric, so the conservative answer is the one that collects.
///
/// Everything else either keeps the set it read or answers a set of its own:
///
@@ -1104,6 +1192,12 @@ pub fn duplicates_canonical_labels(node: &NativeNode, source_tags: &[Vec<String>
///
/// The walk is over the ops in written order, starting from the labels the input reports. A
/// destination the input already carries can fold two series into one; a fresh destination cannot.
/// A `__name__` destination is always treated as carried, even when the derived set has lost it: a
/// kernel such as `abs`, `timestamp` or a range function drops the name, but the scan below it can
/// still hold several metrics whose ordinary labels are equal, and writing one name puts those
/// identities under one label set. Nothing here proves the input reads a single metric, so the
/// conservative answer is the one that collects.
///
/// `None` when the input's label set is not derivable, which is treated as "it may carry the
/// destination" by the caller: an unknown set proves nothing, so the conservative answer is the one
/// that collects.
@@ -1117,10 +1211,12 @@ fn map_writes_existing_label(
for operation in op.ops.iter() {
if let MapOp::Label(rewrite) = operation {
let destination = rewrite.destination();
if destination == "__name__" || tags.iter().any(|tag| tag == destination) {
if destination == METRIC_NAME || tags.iter().any(|tag| tag == destination) {
existing = true;
}
if !tags.iter().any(|tag| tag == destination) {
// A label the input did not carry is one the rewrite publishes only when it can write a
// value; a delete-only rewrite adds nothing a later op could read as carried.
if rewrite.writes_value() && !tags.iter().any(|tag| tag == destination) {
tags.push(destination.to_string());
}
}
src/promql/src/native/plan/tests.rs
+234 -3
View File
@@ -26,6 +26,7 @@ use super::{
NativeType, duplicates_canonical_labels, fused_output_tags, reads_across_series,
};
use crate::extension_plan::{FusedSeries, FusedSeriesBinding};
use crate::histogram_eval::NativeHistogramFunction;
use crate::info::InfoSelector;
use crate::map::MathOp;
use crate::range_eval::{RangeFunction, RangeWindow};
@@ -1924,13 +1925,18 @@ fn info_reads_the_sample_times_of_one_field() {
/// `label_replace(x, dst, "$1", src, "(.*)")` over `inner`, a rewrite that writes `dst`.
fn rewritten(inner: NativeNode, dst: &str, src: &str) -> NativeNode {
replaced(inner, dst, "$1", src)
}
/// `label_replace(x, dst, replacement, src, "(.*)")` over `inner`.
fn replaced(inner: NativeNode, dst: &str, replacement: &str, src: &str) -> NativeNode {
let output = inner.output.clone();
NativeNode {
expr: NativeExpr::Map {
input: Box::new(inner),
op: BoundMap {
ops: Arc::new([MapOp::Label(Arc::new(
crate::map_label::LabelRewrite::replace(dst, "$1", src, "(.*)")
crate::map_label::LabelRewrite::replace(dst, replacement, src, "(.*)")
.expect("the rewrite compiles"),
))]),
},
@@ -1979,6 +1985,225 @@ fn a_rewrite_of_an_existing_label_aliases() {
));
}
/// A scan bound to one metric names every series it hands over, so `__name__` is one of its runtime
/// labels even though it is not a tag column. A rewrite of that name writes an existing label and
/// can fold two metrics whose ordinary labels are equal into one identity.
#[test]
fn a_rewrite_of_the_metric_name_aliases_a_named_source() {
let out = grid(0, 60_000, 4);
let inner = input_on(Arc::clone(&out));
let named = vec![vec!["host".to_string(), "__name__".to_string()]];
assert!(duplicates_canonical_labels(
&rewritten(inner, "__name__", "__name__"),
&named
));
// The delete-only form can fold the same two identities, so it is collected too.
let inner = input_on(Arc::clone(&out));
assert!(duplicates_canonical_labels(
&replaced(inner, "__name__", "", "__name__"),
&named
));
}
/// A name write is collected even above a kernel that dropped the name: the scan below `abs`, a
/// `timestamp` or a range function can still hold several metrics whose ordinary labels are equal,
/// so writing one name puts those identities under one label set. A fresh ordinary destination is
/// still uncollected, because no scan can carry an ordinary label it does not declare.
#[test]
fn a_name_rewrite_above_a_dropped_name_is_collected() {
let out = grid(0, 60_000, 4);
let named = vec![vec!["host".to_string(), "__name__".to_string()]];
let selected = || input_on(Arc::clone(&out));
let timestamp = |inner: NativeNode| NativeNode {
expr: NativeExpr::Timestamp {
input: Box::new(inner),
},
output: evaluated(fields(1), Arc::clone(&out)),
};
for (name, dropped) in [
("abs", map_over(selected())),
("timestamp", timestamp(selected())),
// `rate` computes a value of its own, so it drops the name too.
(
"rate",
window_fields(input_with_fields(1), Arc::clone(&out), 1),
),
] {
assert!(
duplicates_canonical_labels(
&replaced(dropped.clone(), "__name__", "same", "missing"),
&named
),
"{name}: a name write may alias metrics the kernel left indistinguishable"
);
assert!(
!duplicates_canonical_labels(&rewritten(dropped, "dst", "host"), &named),
"{name}: a fresh ordinary destination cannot alias"
);
}
}
/// The label walk keeps a name a rewrite writes and drops it wherever the kernel does, so the
/// query's output columns never claim a name the runtime does not report.
#[test]
fn a_written_name_survives_only_where_the_tree_keeps_it() {
let out = grid(0, 60_000, 4);
let selected = || input_on(Arc::clone(&out));
// A scan bound to one metric: its runtime labels carry the name, its columns do not.
let runtime = vec![vec!["host".to_string(), "__name__".to_string()]];
let columns = vec![vec!["host".to_string()]];
let written = || rewritten(selected(), "__name__", "__name__");
let name_kept = |tags: Option<Vec<String>>| {
tags.is_some_and(|tags| tags.iter().any(|tag| tag == "__name__"))
};
assert_eq!(
Some(vec!["host".to_string(), "__name__".to_string()]),
fused_output_tags(&written(), &runtime)
);
// A column set gains the name because the rewrite writes a value into it.
assert_eq!(
Some(vec!["host".to_string(), "__name__".to_string()]),
fused_output_tags(&written(), &columns)
);
// A delete-only rewrite over a scan that publishes no name column publishes none.
assert_eq!(
Some(vec!["host".to_string()]),
fused_output_tags(&replaced(selected(), "__name__", "", "__name__"), &columns)
);
// The delete-only rewrite over a set that carries the name keeps it: whether a series keeps its
// value depends on the regex, so the label is still part of the set.
assert_eq!(
Some(vec!["host".to_string(), "__name__".to_string()]),
fused_output_tags(&replaced(selected(), "__name__", "", "__name__"), &runtime)
);
let histogram = |inner: NativeNode| NativeNode {
expr: NativeExpr::Map {
input: Box::new(inner),
op: BoundMap {
ops: Arc::new([MapOp::Histogram(NativeHistogramFunction::Count)]),
},
},
output: evaluated(fields(1), Arc::clone(&out)),
};
let timestamp = |inner: NativeNode| NativeNode {
expr: NativeExpr::Timestamp {
input: Box::new(inner),
},
output: evaluated(fields(1), Arc::clone(&out)),
};
for (name, dropped) in [
("abs", map_over(written())),
("histogram_count", histogram(written())),
("timestamp", timestamp(written())),
("rate", window_fields(written(), Arc::clone(&out), 1)),
] {
assert!(!name_kept(fused_output_tags(&dropped, &runtime)), "{name}");
assert!(!name_kept(fused_output_tags(&dropped, &columns)), "{name}");
}
}
/// `group_left(__name__)` copies the one side's name onto an arithmetic result: the kernel drops the
/// metadata before it copies an included label, so the name the one side carries is what the result
/// reports, while a one side without the name removes it from the result.
#[test]
fn an_included_metric_name_survives_an_arithmetic_match() {
let out = grid(0, 60_000, 4);
let input = |source_index: usize| {
let output = NativeType::Samples(Arc::new(SeriesSchema {
fields: fields(1),
eval_grid: None,
}));
NativeNode {
expr: NativeExpr::Input(BoundInput {
output: output.clone(),
source_index,
ordering: SeriesOrdering::KeyThenTime,
}),
output,
}
};
let binary = |left: NativeNode, right: NativeNode, matching: BinaryMatching| NativeNode {
expr: NativeExpr::Binary {
left: Box::new(left),
right: Box::new(right),
matching,
},
output: evaluated(fields(1), Arc::clone(&out)),
};
let group_left = || BinaryMatching {
op: BinaryOp::Mul,
return_bool: false,
labels: MatchingLabels::On(vec![Arc::from("host")]),
cardinality: Cardinality::ManyToOne(vec![Arc::from("__name__")]),
fill: FillValues::default(),
};
let source_tags = vec![
vec!["host".to_string(), "__name__".to_string()],
vec!["host".to_string(), "__name__".to_string()],
];
let tagged = |source_index: usize| input(source_index);
// The one side (right) carries a rewritten name, so the include puts it on the result.
let renamed = replaced(tagged(1), "__name__", "renamed", "__name__");
assert_eq!(
Some(vec!["host".to_string(), "__name__".to_string()]),
fused_output_tags(&binary(tagged(0), renamed, group_left()), &source_tags)
);
// A one side without the name cannot give the include one, so the result drops it.
let unnamed_tags = vec![
vec!["host".to_string(), "__name__".to_string()],
vec!["host".to_string()],
];
assert_eq!(
Some(vec!["host".to_string()]),
fused_output_tags(&binary(tagged(0), tagged(1), group_left()), &unnamed_tags)
);
// `group_right` trades the sides: the include is read from the left (one) side there.
let group_right = || BinaryMatching {
op: BinaryOp::Mul,
return_bool: false,
labels: MatchingLabels::On(vec![Arc::from("host")]),
cardinality: Cardinality::OneToMany(vec![Arc::from("__name__")]),
fill: FillValues::default(),
};
let renamed = replaced(tagged(0), "__name__", "renamed", "__name__");
assert_eq!(
Some(vec!["host".to_string(), "__name__".to_string()]),
fused_output_tags(&binary(renamed, tagged(1), group_right()), &source_tags)
);
// A `bool` comparison drops the name like arithmetic does, and the include still copies the one
// side's name afterwards: the drop happens before the include.
let bool_comparison = || BinaryMatching {
op: BinaryOp::Eq,
return_bool: true,
labels: MatchingLabels::On(vec![Arc::from("host")]),
cardinality: Cardinality::ManyToOne(vec![Arc::from("__name__")]),
fill: FillValues::default(),
};
let renamed = replaced(tagged(1), "__name__", "renamed", "__name__");
assert_eq!(
Some(vec!["host".to_string(), "__name__".to_string()]),
fused_output_tags(&binary(tagged(0), renamed, bool_comparison()), &source_tags)
);
// A filtering comparison keeps the name of the side it reports, but an included name the one
// side does not carry still leaves the set: `result_metric` removes an included label it cannot
// copy, and the base name is not exempt.
let filtering = || BinaryMatching {
op: BinaryOp::Eq,
return_bool: false,
labels: MatchingLabels::On(vec![Arc::from("host")]),
cardinality: Cardinality::ManyToOne(vec![Arc::from("__name__")]),
fill: FillValues::default(),
};
assert_eq!(
Some(vec!["host".to_string()]),
fused_output_tags(&binary(tagged(0), tagged(1), filtering()), &unnamed_tags)
);
}
/// A source the caller says nothing about proves nothing: the pass has to assume the destination
/// was one of its labels, because two series sharing one is exactly the wrong answer to ship.
#[test]
@@ -2121,14 +2346,20 @@ fn count_values_reports_its_grouping_and_the_label_it_writes() {
&source_tags
)
);
// A destination naming the metric name writes no column this walk can name.
// A destination naming the metric name writes the name, which is the one label the node's
// result carries when its grouping names nothing else.
assert_eq!(
None,
tags(&["__name__"]),
fused_output_tags(
&counted("__name__", Grouping::By(Arc::new([]))),
&source_tags
)
);
// An empty destination cannot be a label, so the walk declines rather than name a column.
assert_eq!(
None,
fused_output_tags(&counted("", Grouping::By(Arc::new([]))), &source_tags)
);
}
/// A consumer that files its rows by the label set a sample ends under answers one series per set,
src/query/src/promql/planner/native.rs
+178 -20
View File
@@ -63,8 +63,8 @@ use promql::extension_plan::{
use promql::native::execute::NativeExecutionState;
use promql::native::plan::{
AggregateKind, BoundAggregate, BoundInput, BoundVectorOp, Grouping, InstantValue, MapOp,
NativeExpr, NativeNode, NativeTree, NativeType, binary_output_field_kind, fused_output_tags,
reads_across_series, uniform_field_kind,
NativeExpr, NativeNode, NativeTree, NativeType, binary_output_field_kind, reads_across_series,
uniform_field_kind,
};
use promql::vector_match::{Cardinality, MatchingLabels, SetOp};
use promql_parser::label::{METRIC_NAME, MatchOp, Matchers};
@@ -1321,6 +1321,23 @@ impl PromPlanner {
.iter()
.map(|(_, _, _, _, tags, _, _)| tags.clone())
.collect::<Vec<_>>();
// The complete runtime label set of each input: a scan bound to one metric names every
// series it hands over, so `__name__` is one of its labels even though it is not a tag
// column. The canonical pass judges a rewrite by the labels it writes and the ones its input
// carries, so rewriting the name of a named scan is a rewrite of an existing label - and a
// discovered source declares the name among its tags itself. The output-column walks keep
// `source_tags`: the name becomes a column only when a rewrite writes one, or when the
// source declares it.
let runtime_source_tags = prepared
.iter()
.map(|(_, _, metric, _, tags, _, _)| {
let mut tags = tags.clone();
if metric.is_some() && !tags.iter().any(|tag| tag == METRIC_NAME) {
tags.push(METRIC_NAME.to_string());
}
tags
})
.collect::<Vec<_>>();
// Whether any input's table is split into regions along a partition column. The canonical
// boundary is the only part of this route that claims a whole-query label-set answer, and a
// region-local producer cannot see the pieces another region holds, so this fact decides
@@ -1334,7 +1351,7 @@ impl PromPlanner {
// matters and nowhere else, so a tree with no collapsing rewrite keeps every partition.
let tree = match promql::native::canonicalize::merge_label_rewrites(
&compiled.tree,
&source_tags,
&runtime_source_tags,
) {
Ok(tree) => tree,
Err(_) => return Ok(None),
@@ -2375,7 +2392,7 @@ fn canonical_boundary_stages(
.iter()
.map(|field| field.name.to_string())
.collect::<Vec<_>>();
let output_tags = fused_output_tags(tree.root(), source_tags)?;
let output_tags = native_output_tags(tree.root(), source_tags)?;
let output_schema = fused_output_schema(&fields, &output_tags, time_index, root_kind).ok()?;
// The producer: the statement's own tree, run per partition, exporting a frame.
@@ -2764,7 +2781,7 @@ fn segment_tree(
)?;
match plan {
Some(plan) => {
let Some(cut_tags) = fused_output_tags(cut, &current_tags) else {
let Some(cut_tags) = native_output_tags(cut, &current_tags) else {
return Ok(None);
};
Some((plan, cut_tags))
@@ -3386,11 +3403,18 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<
for operation in op.ops.iter() {
if let MapOp::Label(rewrite) = operation {
let destination = rewrite.destination();
if destination == "__name__" {
return None;
if tags.iter().any(|tag| tag == destination) {
// The input carries the label - a discovered source declares `__name__`
// among its tags - so the rewrite changes its value, and an empty
// replacement may remove it, which the nullable column still represents.
tags.retain(|tag| tag != destination);
tags.push(destination.to_string());
} else if rewrite.writes_value() {
// A fresh destination is a column only when the rewrite can write a
// value: that is how a `__name__` destination publishes a name column,
// and how a delete-only rewrite publishes none.
tags.push(destination.to_string());
}
tags.retain(|tag| tag != destination);
tags.push(destination.to_string());
}
}
return Some(tags);
@@ -3486,6 +3510,8 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<
matching,
} => {
let keeps_name = keeps_metric_name(matching);
let scalar_side = matches!(&left.output, NativeType::Scalar(_))
|| matches!(&right.output, NativeType::Scalar(_));
let tags = match (&left.output, &right.output) {
(NativeType::Scalar(_), _) => native_output_tags(right, source_tags)?,
(_, NativeType::Scalar(_)) => native_output_tags(left, source_tags)?,
@@ -3518,6 +3544,13 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<
Cardinality::ManyToOne(_) => left_tags.clone(),
Cardinality::OneToMany(_) => right_tags.clone(),
};
// `result_metric` drops the metadata labels before it matches or copies an
// included label, so an arithmetic result starts without the name. An include can
// still put it back from the one side, which is why the final rule below does not
// run on this branch.
if !keeps_name {
tags.retain(|tag| tag != METRIC_NAME);
}
let (includes, include_tags) = match &matching.cardinality {
Cardinality::OneToOne => (&[][..], &[][..]),
Cardinality::ManyToOne(labels) => {
@@ -3561,10 +3594,7 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<
}
};
tags.retain(|tag| {
// The name is not an ordinary label here: `keeps_name` above decides
// whether the walk reports it.
tag == METRIC_NAME
|| !includes.iter().any(|label| label.as_ref() == tag.as_str())
!includes.iter().any(|label| label.as_ref() == tag.as_str())
|| include_tags.contains(tag)
|| (many_rows_filled && in_match_key(tag))
});
@@ -3585,8 +3615,17 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<
(right_tags.as_slice(), matching.fill.lhs)
}
};
// A filled row keeps only the one side's matching labels, so the name can reach
// it only through an `on` key that names the metric name, and only while the
// operation reports its samples' names.
let fill_carries_name = keeps_name
&& matches!(&matching.labels, MatchingLabels::On(names)
if names.iter().any(|name| name.as_ref() == METRIC_NAME));
if fills_many_side.is_some() {
for tag in kept(one_tags.to_vec()) {
if tag == METRIC_NAME && !fill_carries_name {
continue;
}
if !tags.contains(&tag) {
tags.push(tag);
}
@@ -3595,10 +3634,10 @@ fn native_output_tags(node: &NativeNode, source_tags: &[Vec<String>]) -> Option<
tags
}
};
Some(if keeps_name {
tags
} else {
Some(if scalar_side && !keeps_name {
without_metric_name(tags)
} else {
tags
})
}
NativeExpr::Set { left, right, op } => {
@@ -4108,15 +4147,15 @@ mod tests {
use datafusion::physical_plan::ExecutionPlan;
use datatypes::series::SeriesOrdering;
use datatypes::series::samples::{FieldSpec, SeriesSchema};
use promql::native::plan::{NativeExpr, NativeNode, NativeTree};
use promql::native::plan::{NativeExpr, NativeNode, NativeTree, fused_output_tags};
use session::context::QueryContextBuilder;
use super::{
ArrowDataType, ArrowTimeUnit, FieldKind, METRIC_NAME, NATIVE_PROMQL_EXTENSION_KEY,
NativeType, aggregate_reads_histogram, collect_selectors_with_set_operations, cut_segments,
fused_output_schema, fused_output_tags, independent_scan_pair, native_output_tags,
native_promql_requested, native_raw_plan_bounds_region_reads, node_children,
plans_over_remote_regions, scan_time_filter,
fused_output_schema, independent_scan_pair, native_output_tags, native_promql_requested,
native_raw_plan_bounds_region_reads, node_children, plans_over_remote_regions,
scan_time_filter,
};
#[test]
@@ -4696,6 +4735,99 @@ mod tests {
);
}
/// A `__name__` rewrite publishes a name column, canonicalizes the complete runtime label set,
/// and leaves the name dropped wherever a kernel computes a value of its own.
#[test]
fn name_rewrite_publishes_a_column_and_is_canonicalized() {
let columns = [vec!["host".to_string(), "kind".to_string()]];
let runtime = [vec![
"host".to_string(),
"kind".to_string(),
METRIC_NAME.to_string(),
]];
let tree =
compiled_test_tree(r#"label_replace(x, "__name__", "same", "__name__", "(.*)")"#);
assert_eq!(
Some(vec!["host".into(), "kind".into(), METRIC_NAME.into()]),
native_output_tags(tree.root(), &columns),
"a written name is a column the output publishes"
);
let collected = promql::native::canonicalize::merge_label_rewrites(&tree, &runtime)
.expect("the pass produces a legal tree");
assert!(
matches!(collected.root().expr, NativeExpr::Merge { .. }),
"a name rewrite writes an existing runtime label: {collected}"
);
assert_eq!(
native_output_tags(tree.root(), &columns),
native_output_tags(collected.root(), &columns),
"collecting the pieces does not change the columns"
);
// A delete-only rewrite over a scan whose columns carry no name publishes none.
let deleted = compiled_test_tree(r#"label_replace(x, "__name__", "", "__name__", "(.*)")"#);
assert_eq!(
Some(vec!["host".into(), "kind".into()]),
native_output_tags(deleted.root(), &columns)
);
// A kernel that computes a value of its own drops the name a rewrite below it wrote.
for query in [
r#"abs(label_replace(x, "__name__", "same", "__name__", "(.*)"))"#,
r#"histogram_count(label_replace(x, "__name__", "same", "__name__", "(.*)"))"#,
r#"timestamp(label_replace(x, "__name__", "same", "__name__", "(.*)"))"#,
r#"sum(label_replace(x, "__name__", "same", "__name__", "(.*)"))"#,
] {
let tree = compiled_test_tree(query);
let tags = native_output_tags(tree.root(), &columns)
.unwrap_or_else(|| panic!("{query} derives its columns"));
assert!(
!tags.iter().any(|tag| tag == METRIC_NAME),
"{query}: {tags:?}"
);
}
}
/// A name write is collected even above a kernel that dropped the name: the scan below it can
/// still hold several metrics whose ordinary labels are equal, so writing one name can fold them
/// into one identity. The output still publishes the written name.
#[test]
fn a_name_rewrite_above_a_dropped_name_is_collected() {
let runtime = [vec![
"host".to_string(),
"kind".to_string(),
METRIC_NAME.to_string(),
]];
let tree =
compiled_test_tree(r#"label_replace(abs(x), "__name__", "same", "missing", "(.*)")"#);
let collected = promql::native::canonicalize::merge_label_rewrites(&tree, &runtime)
.expect("the pass produces a legal tree");
assert!(
matches!(collected.root().expr, NativeExpr::Merge { .. }),
"the merge sits above the name rewrite: {collected}"
);
assert_eq!(
1,
collected.to_string().matches("Merge").count(),
"exactly one merge: {collected}"
);
// The rewrite writes a name over a dropped one, so the output publishes a name column.
assert_eq!(
Some(vec!["host".into(), "kind".into(), METRIC_NAME.into()]),
native_output_tags(collected.root(), &[vec!["host".into(), "kind".into()]])
);
// A fresh ordinary destination above the same kernel is still uncollected: no scan can carry
// an ordinary label it does not declare.
let fresh = compiled_test_tree(r#"label_replace(abs(x), "dst", "$1", "host", "(.*)")"#);
let collected = promql::native::canonicalize::merge_label_rewrites(&fresh, &runtime)
.expect("the pass produces a legal tree");
assert!(
!collected.to_string().contains("Merge"),
"a fresh ordinary destination does not alias: {collected}"
);
}
#[test]
fn group_right_copies_included_labels_into_output_tags() {
let sources = [
@@ -4709,6 +4841,32 @@ mod tests {
);
}
/// `group_left(__name__)` copies the one side's name onto an arithmetic result: the operation
/// drops the name before it copies an included label, so the copied name is a column. A one side
/// without a name cannot provide one, so the column leaves the set.
#[test]
fn an_included_metric_name_is_an_output_column() {
let sources = [
vec!["host".to_string()],
vec!["host".to_string(), METRIC_NAME.to_string()],
];
let included = compiled_test_tree(
r#"x * on(host) group_left(__name__) label_replace(y, "__name__", "renamed", "__name__", "(.*)")"#,
);
assert_eq!(
Some(vec!["host".to_string(), METRIC_NAME.to_string()]),
native_output_tags(included.root(), &sources)
);
let unnamed = compiled_test_tree(r#"x * on(host) group_left(__name__) y"#);
assert_eq!(
Some(vec!["host".to_string()]),
native_output_tags(
unnamed.root(),
&[vec!["host".to_string()], vec!["host".to_string()]]
)
);
}
#[test]
fn numeric_aggregate_projection_keeps_grouping_tags_and_value_schema() {
let sources = [vec![
src/query/src/promql/planner/test.rs
+96
View File
@@ -5559,6 +5559,10 @@ async fn native_fused_aggregate_splits_into_a_final_over_a_partial_producer() {
"stdvar(some_metric)",
"sum by (tag_0) (rate(some_metric[5m]))",
"sum without (tag_0) (some_metric)",
// The output columns are the whole tree's, not the Final half's state input's: `dst` is
// created by the prefix rewrite, so a grouping that keeps or names it publishes it.
r#"sum without (tag_0) (label_replace(some_metric, "dst", "$1", "tag_0", "(.*)"))"#,
r#"sum by (dst) (label_replace(some_metric, "dst", "$1", "tag_0", "(.*)"))"#,
"sum(rate(some_metric[5m]))",
"ceil(sum(some_metric))",
// A literal scalar on either side of the fold. Below the aggregate it is part of the prefix,
@@ -5671,6 +5675,12 @@ async fn native_fused_aggregate_splits_into_a_final_over_a_partial_producer() {
"stdvar(some_metric)" => &["field_0", "timestamp"],
"sum by (tag_0) (rate(some_metric[5m]))" => &["field_0", "tag_0", "timestamp"],
"sum without (tag_0) (some_metric)" => &["field_0", "timestamp"],
r#"sum without (tag_0) (label_replace(some_metric, "dst", "$1", "tag_0", "(.*)"))"# => {
&["dst", "field_0", "timestamp"]
}
r#"sum by (dst) (label_replace(some_metric, "dst", "$1", "tag_0", "(.*)"))"# => {
&["dst", "field_0", "timestamp"]
}
"sum(rate(some_metric[5m]))" => &["field_0", "timestamp"],
"ceil(sum(some_metric))" => &["field_0", "timestamp"],
"sum(avg(some_metric) by (tag_0))" => &["field_0", "timestamp"],
@@ -6754,6 +6764,92 @@ async fn native_fused_label_rewrites_keep_legacy_output_labels() {
}
}
/// A `__name__` destination is native-only: the existing planner refuses the destination outright,
/// so the only plan that can answer it is this one. What it must get right is which rewrites publish
/// the name as a column and which leave the output without one.
#[tokio::test]
async fn native_fused_metric_name_rewrites_publish_or_drop_the_name() {
for (query, keeps_name) in [
// The rewrite at the statement root, and a join that writes a value into the name.
(
r#"label_replace(some_metric, "__name__", "same", "__name__", "(.*)")"#,
true,
),
(r#"label_join(some_metric, "__name__", "-", "tag_0")"#, true),
// A delete-only rewrite of a scan that publishes no name column publishes none.
(
r#"label_replace(some_metric, "__name__", "", "__name__", "(.*)")"#,
false,
),
(r#"label_join(some_metric, "__name__", "-")"#, false),
// A kernel above the rewrite computes a value of its own and drops the name with it.
(
r#"abs(label_replace(some_metric, "__name__", "same", "__name__", "(.*)"))"#,
false,
),
(
r#"histogram_count(label_replace(some_metric, "__name__", "same", "__name__", "(.*)"))"#,
false,
),
(
r#"timestamp(label_replace(some_metric, "__name__", "same", "__name__", "(.*)"))"#,
false,
),
// A fresh name above a dropped one is a new label, and still becomes a column.
(
r#"label_replace(abs(some_metric), "__name__", "same", "__name__", "(.*)")"#,
true,
),
] {
let plan = build_native_fused_plan(query, true).await;
let display = plan.display_indent().to_string();
assert!(
find_fused_series(&plan).is_some(),
"`{query}` must run natively: {display}"
);
assert_eq!(
plan.schema()
.fields()
.iter()
.any(|field| field.name() == METRIC_NAME),
keeps_name,
"`{query}`: {display}"
);
}
}
/// A rewrite of the metric name over a discovered selector writes a label the source declares, so
/// two metrics whose ordinary labels are equal are collected by the canonical boundary.
#[tokio::test]
async fn native_metric_discovery_name_rewrite_canonicalizes_by_name() {
let plan = build_native_fused_plan_with_provider(
r#"label_replace({tag_0="x"}, "__name__", "same", "__name__", "(.*)")"#,
true,
metric_discovery_provider(&["alpha_total", "beta_total"]).await,
)
.await;
let display = plan.display_indent().to_string();
assert!(
plan.schema()
.fields()
.iter()
.any(|field| field.name() == METRIC_NAME),
"the rewritten name is a column of the answer: {display}"
);
let fused = find_fused_series(&plan).expect("fused plan");
assert!(
fused.children().iter().any(|child| matches!(
child,
LogicalPlan::Extension(Extension { node })
if node
.as_any()
.downcast_ref::<FusedSeries>()
.is_some_and(FusedSeries::is_canonical_producer)
)),
"the name rewrite runs through a canonical boundary: {display}"
);
}
#[tokio::test]
async fn native_fused_preparation_decline_preserves_legacy_plan() {
let query = "absent(vector(1)) + on(tag_0) some_metric{tag_0=\"x\",__field__=\"field_0\"}";
tests-integration/src/tests/promql_test.rs
+265 -3
View File
@@ -5780,9 +5780,8 @@ async fn native_promql_label_rewrites_reach_the_native_route_and_keep_prometheus
);
}
// A `__name__` destination is not covered here on purpose: the existing planner refuses that
// destination outright ("Invalid destination label name in label_replace()"), so no route can
// answer it and admitting the shape natively would only turn a refusal into a different one.
// A `__name__` destination is native-only - the existing planner refuses it - and is covered by
// `native_promql_rewrites_the_metric_name` below.
// Two stored series rewritten to one label set, holding values at different evaluation points,
// are one series carrying both points rather than two rows.
@@ -5800,6 +5799,269 @@ async fn native_promql_label_rewrites_reach_the_native_route_and_keep_prometheus
);
}
/// A `__name__` destination is native-only: the existing planner refuses that destination, so this
/// test reads the native route's own answer and the plan that produced it, and never compares
/// against a legacy run.
///
/// Three logical metrics share one physical table. `name_left` and `name_right` hold the same two
/// ordinary label sets ten minutes apart, so rewriting both names to one value puts their samples
/// under one canonical identity; `name_clash` holds a sample where `name_left` already has one under
/// `host="h"`, so the same rewrite has a genuine same-point conflict to report.
#[tokio::test(flavor = "multi_thread")]
async fn native_promql_rewrites_the_metric_name() {
common_telemetry::init_default_ut_logging();
let mito_config = mito2::config::MitoConfig {
experimental_series_scan_v2: true,
..Default::default()
};
let (standalone, observer) = native_test_instance(
GreptimeDbStandaloneBuilder::new("native_promql_metric_name_rewrite")
.with_mito_config(mito_config),
)
.await;
let instance = standalone.fe_instance().clone();
execute_all(
&instance,
"CREATE TABLE name_phy (ts TIMESTAMP TIME INDEX, val DOUBLE) ENGINE=metric WITH ('physical_metric_table'='true'); \
CREATE TABLE name_left (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(host)) ENGINE=metric WITH ('on_physical_table'='name_phy'); \
CREATE TABLE name_right (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(host)) ENGINE=metric WITH ('on_physical_table'='name_phy'); \
CREATE TABLE name_clash (host STRING, ts TIMESTAMP TIME INDEX, val DOUBLE, PRIMARY KEY(host)) ENGINE=metric WITH ('on_physical_table'='name_phy'); \
INSERT INTO name_left (host, ts, val) VALUES ('h', 0, 1), ('i', 0, 2); \
INSERT INTO name_right (host, ts, val) VALUES ('h', 600000, 10), ('i', 600000, 20); \
INSERT INTO name_clash (host, ts, val) VALUES ('h', 0, 100)",
QueryContext::arc(),
)
.await;
let run = |expr: &str, range: &str| {
let sql = format!("TQL EVAL {range} {expr}");
let instance = instance.clone();
async move {
instance
.do_query(&sql, native_query_context(true, 4))
.await
.remove(0)
}
};
// A nameless selector is answered by enumerating the schema's metric tables, which the
// frontend's PromQL entry point bounds by the caller's grants before it plans. A SQL
// `TQL EVAL` statement prepares no such scope, so the test states the unrestricted one a
// caller with every grant would get.
let discovery_ctx = Arc::new({
let mut ctx = QueryContext::with_db_name(None);
ctx.set_extension(QUERY_PARALLELISM_HINT, 4.to_string());
ctx.set_extension(NATIVE_PROMQL_EXTENSION_KEY, "true");
ctx.set_metric_discovery_scope(MetricDiscoveryScope::Unrestricted);
ctx
});
let run_discovery = |expr: &str, range: &str| {
let sql = format!("TQL EVAL {range} {expr}");
let instance = instance.clone();
let ctx = Arc::clone(&discovery_ctx);
async move { instance.do_query(&sql, ctx).await.remove(0) }
};
let labels = |pairs: &[(&str, &str)]| {
pairs
.iter()
.map(|(name, value)| ((*name).to_string(), (*value).to_string()))
.collect::<std::collections::BTreeMap<_, _>>()
};
// The boundary is a partition-local producer that merges what its own scan partition held and
// the canonical router that collects every partition's pieces of one label set. The router is
// only built over a canonical producer, so finding it by name is what proves the plan answered
// through that boundary rather than some other way.
let expect_canonical_boundary = |context: &str| {
assert!(
observer.fused().is_some(),
"{context}: a native fused node answered"
);
assert!(
canonical_router(&observer).is_some(),
"{context}: the canonical router collects each label set: {:?}",
observer
.all_plans()
.iter()
.map(display_plan_tree)
.collect::<Vec<_>>()
);
};
// Renaming the name keeps series whose other labels differ distinct, and publishes the new name.
observer.clear();
let renamed = run(
r#"label_replace(name_left, "__name__", "renamed", "__name__", "(.*)")"#,
"(0, 0, '1s')",
)
.await
.unwrap();
let (columns, rows) = native_labelled_rows(renamed).await;
assert!(
columns.iter().any(|column| column == "__name__"),
"the rewritten name is a column: {columns:?}"
);
assert_eq!(
rows,
vec![
(0, labels(&[("__name__", "renamed"), ("host", "h")]), 1.0),
(0, labels(&[("__name__", "renamed"), ("host", "i")]), 2.0),
],
"a name rewrite reports the new name and keeps distinct series distinct"
);
expect_canonical_boundary("a name rewrite of a named scan");
// Deleting the name reports no name, and a delete-only rewrite of a scan that publishes no name
// column publishes none: whether a series keeps its value depends on the regex, and a deleted
// value is the only one this rewrite can write.
observer.clear();
let deleted = run(
r#"label_replace(name_left, "__name__", "", "__name__", "(.*)")"#,
"(0, 0, '1s')",
)
.await
.unwrap();
let (columns, rows) = native_labelled_rows(deleted).await;
assert!(
!columns.iter().any(|column| column == "__name__"),
"a deleted name publishes no column: {columns:?}"
);
assert_eq!(
rows,
vec![
(0, labels(&[("host", "h")]), 1.0),
(0, labels(&[("host", "i")]), 2.0),
],
"a deleted name is absent from the answer"
);
expect_canonical_boundary("a name deletion");
// `label_join` writes the name as well, and an empty source list deletes it.
for (expr, expected) in [
(
r#"label_join(name_left, "__name__", "-", "host")"#,
vec![
(0, labels(&[("__name__", "h"), ("host", "h")]), 1.0),
(0, labels(&[("__name__", "i"), ("host", "i")]), 2.0),
],
),
(
r#"label_join(name_left, "__name__", "-")"#,
vec![
(0, labels(&[("host", "h")]), 1.0),
(0, labels(&[("host", "i")]), 2.0),
],
),
] {
observer.clear();
let joined = run(expr, "(0, 0, '1s')").await.unwrap();
let (_, rows) = native_labelled_rows(joined).await;
assert_eq!(rows, expected, "{expr}");
expect_canonical_boundary(expr);
}
// Two metrics told apart only by name: rewriting both names to one value merges the points each
// of them holds, because the canonical identity is the complete label set with the new name.
observer.clear();
let complementary = run_discovery(
r#"label_replace({__name__=~"name_left|name_right", host=~"h|i"}, "__name__", "same", "__name__", "(.*)")"#,
"(0, 600, '10m')",
)
.await
.unwrap();
let (_, rows) = native_labelled_rows(complementary).await;
assert_eq!(
rows,
vec![
(0, labels(&[("__name__", "same"), ("host", "h")]), 1.0),
(
600_000,
labels(&[("__name__", "same"), ("host", "h")]),
10.0
),
(0, labels(&[("__name__", "same"), ("host", "i")]), 2.0),
(
600_000,
labels(&[("__name__", "same"), ("host", "i")]),
20.0
),
],
"one identity carries both metrics' points"
);
expect_canonical_boundary("two metrics renamed to one name");
// A name written above `abs` is collected even though `abs` dropped the name: the two metrics are
// still told apart by name alone below it, so writing one name folds their samples into one
// identity. Without the collection each metric would answer as its own series under one label set.
observer.clear();
let rewritten_above_abs = run_discovery(
r#"label_replace(abs({__name__=~"name_left|name_right", host=~"h|i"}), "__name__", "same", "", ".*")"#,
"(0, 600, '10m')",
)
.await
.unwrap();
let (_, rows) = native_labelled_rows(rewritten_above_abs).await;
assert_eq!(
rows,
vec![
(0, labels(&[("__name__", "same"), ("host", "h")]), 1.0),
(
600_000,
labels(&[("__name__", "same"), ("host", "h")]),
10.0
),
(0, labels(&[("__name__", "same"), ("host", "i")]), 2.0),
(
600_000,
labels(&[("__name__", "same"), ("host", "i")]),
20.0
),
],
"one identity carries both metrics' points after `abs` dropped their names"
);
expect_canonical_boundary("a name rewrite above a dropped name");
// The same shape at one point is that identity's duplicate-label-set error, and it is reported
// even though the two series came from different metrics and a math kernel erased their names.
observer.clear();
let renamed_above_abs_clash = run_discovery(
r#"label_replace(abs({__name__=~"name_left|name_clash", host="h"}), "__name__", "same", "", ".*")"#,
"(0, 0, '1s')",
)
.await;
let error = match renamed_above_abs_clash {
Ok(output) => collect_error(output).await,
Err(error) => error.to_string(),
};
assert!(
error.contains("cannot contain metrics with the same labelset"),
"a name written above `abs` reports the duplicate label set: {error}"
);
assert!(
observer.fused().is_some(),
"the collision after a dropped name is reported by a native plan"
);
// A point where the other metric already holds a sample is the duplicate-label-set error, and it
// is reported even though the two series came from different metrics.
observer.clear();
let clash = run_discovery(
r#"label_replace({__name__=~"name_left|name_clash", host="h"}, "__name__", "same", "__name__", "(.*)")"#,
"(0, 0, '1s')",
)
.await;
let error = match clash {
Ok(output) => collect_error(output).await,
Err(error) => error.to_string(),
};
assert!(
error.contains("cannot contain metrics with the same labelset"),
"two metrics renamed to one identity report the duplicate label set: {error}"
);
assert!(
observer.fused().is_some(),
"the collision is reported by a native plan"
);
}
#[tokio::test(flavor = "multi_thread")]
async fn native_promql_executes_fused_node_and_matches_legacy() {
common_telemetry::init_default_ut_logging();