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

feat(promql): lower independently bound vector binary operands

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-10-25 13:14:35 +08:00
parent cdf7c01531
commit 39fcdbdbcf
3 changed files with 1202 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/query/src/promql/native_compile.rs
+89 -27
View File
@@ -77,11 +77,11 @@
//! `<expression>` → `Window` whose input is evaluated (the subquery, whose input grid is its own),
//! `<expression>` → `Map` (a pointwise math, native histogram or label function over what its
//! argument evaluates to) and, over
//! any of those, `Binary` (an arithmetic or comparison operation with a constant on one side),
//! `Scalar` (that constant), `Aggregate` (one of the nine aggregators that fold a group's values
//! into one value per grid point, optionally grouped) and `Vector` (an operation over the whole
//! vector at each point: a sort, `absent`, a classic histogram function, a selecting aggregator or
//! `count_values`). Everything else is a
//! any of those, `Binary` (an arithmetic or comparison operation between vectors or with a
//! constant on one side), `Scalar` (that constant), `Aggregate` (one of the nine aggregators that
//! fold a group's values into one value per grid point, optionally grouped) and `Vector` (an
//! operation over the whole vector at each point: a sort, `absent`, a classic histogram function, a
//! selecting aggregator or `count_values`). Everything else is a
//! [`DataFusionError::NotImplemented`] that names what was found and quotes the shapes above,
//! because rejecting a query the compiler does not understand is the only honest alternative to
//! compiling it into something else.
@@ -107,11 +107,10 @@
//! its own: `-x` binds as the multiplication by `-1` that it is, which the multiplication kernel
//! already computes.
//!
//! What a tree cannot yet express is deliberate too. The one input binding carries the ordering it
//! guarantees and what it produces, but not the selector it reads, so a tree with two selectors
//! cannot tell its inputs apart and an operation between two vectors is refused. A statement whose
//! value is a scalar has nothing to read, so it is refused as well: a constant is bound next to a
//! vector that gives it something to broadcast over.
//! Vector matching is bound with each selector's own input, including the labels and cardinality
//! written on a vector–vector operation. A scalar has no input of its own: it is bound next to the
//! vector that gives it something to broadcast over, while a statement whose value is only a scalar
//! has nothing to read and is refused.
//!
//! An aggregate is bound as [`AggregateStage::Complete`], which is a claim about the whole execution
//! rather than about the node: it says that every contribution to every group is in this execution's
@@ -146,7 +145,9 @@ use promql::native::plan::{
};
use promql::range_eval::{RangeFunction, RangeWindow, subquery_grid};
use promql::select::{LabelMatcher, MatchOp, SortOrder, absent_labels};
use promql::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels};
use promql::vector_match::{
BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels, binary_output_schema,
};
use promql_parser::label::{MatchOp as ParserMatchOp, Matchers};
use promql_parser::parser::token::TokenType;
use promql_parser::parser::value::ValueType;
@@ -433,13 +434,10 @@ fn negate(context: &Context, operand: &PromExpr) -> Result<NativeNode> {
))
}
/// Compiles a binary operation whose operands are not both vectors.
/// Compiles a binary operation, preserving written operand order and vector matching modifiers.
///
/// The tree holds one input binding, and that binding carries the ordering it guarantees and what it
/// produces but not the selector it reads, so two vectors cannot be told apart: `a + b` and `a + a`
/// would compile to the same tree. That is why a vector–vector operation is refused until an input
/// names its selector, and why an operation over one vector and one scalar is the shape the tree can
/// bind: a scalar carries no labels, so broadcasting it is unambiguous.
/// A scalar carries no labels, so broadcasting it over a vector is unambiguous. Vector operands
/// retain their own input bindings and matching, including a fill on either side.
fn binary_operation(context: &Context, binary: &BinaryExpr) -> Result<NativeNode> {
// A set operator combines two vectors' series rather than their values, and the tree has no node
// for it at all. Its name is reported before the operands are looked at, so a set operation
@@ -459,14 +457,27 @@ fn binary_operation(context: &Context, binary: &BinaryExpr) -> Result<NativeNode
binary.op
)));
};
// Two vectors are refused before the modifier is looked at, so a vector–vector query with
// matching says what is actually missing. The refusal names the missing identity rather than the
// query, because that is what whoever adds it has to add.
// Two vectors are matched by their labels, so the node carries the modifiers the query wrote and
// each operand is compiled from its own selector.
if binary.lhs.value_type() == ValueType::Vector && binary.rhs.value_type() == ValueType::Vector
{
return Err(not_implemented(&format!(
"the binary operation `{binary}` between two vectors, whose matching and lowering are not supported yet"
)));
let left = compile_expr(context, &binary.lhs)?;
let right = compile_expr(context, &binary.rhs)?;
let (Some(lhs), Some(rhs)) = (left.output.samples(), right.output.samples()) else {
return Err(DataFusionError::Internal(format!(
"the binary operation `{binary}` compiled a vector operand to something other than \
samples: a vector-typed expression compiles to samples, so the two disagree"
)));
};
let output = NativeType::Samples(binary_output_schema(lhs, rhs)?);
return Ok(NativeNode {
expr: NativeExpr::Binary {
left: Box::new(left),
right: Box::new(right),
matching: vector_matching(binary, op)?,
},
output,
});
}
// What is left has a constant on one side, and a modifier is refused rather than dropped. This
// is deliberately narrower than Prometheus, which drops an `on` or `ignoring` written on a
@@ -551,12 +562,63 @@ fn scalar_matching(op: BinaryOp, return_bool: bool) -> BinaryMatching {
}
}
/// The matching a vector–vector operation records.
///
/// Each part is read from the modifier the query wrote, and a modifier it did not write is the
/// default PromQL applies: every label but `__name__` matches (`ignoring` nothing), and the
/// cardinality is one-to-one. Fill fields retain the parser's left/right values, which name the
/// sides as written: the kernel swaps them together with the operands under `group_right`.
///
/// `ManyToMany` is the cardinality of a set operator, which [`binary_operation`] refuses before
/// asking for a matching, so reaching it means the two disagree; it is reported as a shape with no
/// node rather than panicking on a path a query cannot take today.
fn vector_matching(binary: &BinaryExpr, op: BinaryOp) -> Result<BinaryMatching> {
let modifier = binary.modifier.as_ref();
let labels = match modifier.and_then(|modifier| modifier.matching.as_ref()) {
None => MatchingLabels::Ignoring(Vec::new()),
Some(LabelModifier::Include(labels)) => MatchingLabels::On(label_names(&labels.labels)),
Some(LabelModifier::Exclude(labels)) => {
MatchingLabels::Ignoring(label_names(&labels.labels))
}
};
let cardinality = match modifier.map(|modifier| &modifier.card) {
None | Some(VectorMatchCardinality::OneToOne) => Cardinality::OneToOne,
Some(VectorMatchCardinality::ManyToOne(labels)) => {
Cardinality::ManyToOne(label_names(&labels.labels))
}
Some(VectorMatchCardinality::OneToMany(labels)) => {
Cardinality::OneToMany(label_names(&labels.labels))
}
Some(VectorMatchCardinality::ManyToMany) => {
return Err(not_implemented(&format!(
"the many-to-many matching in `{binary}`, which is the cardinality of a set \
operator: it combines two vectors' series, and this node is an arithmetic or \
comparison operation over values"
)));
}
};
let fill = modifier.map_or_else(FillValues::default, |modifier| FillValues {
lhs: modifier.fill_values.lhs,
rhs: modifier.fill_values.rhs,
});
Ok(BinaryMatching {
op,
return_bool: binary.return_bool(),
labels,
cardinality,
fill,
})
}
fn label_names(labels: &[String]) -> Vec<Arc<str>> {
labels.iter().map(|name| Arc::from(name.as_str())).collect()
}
/// The binary operator a PromQL token denotes, or `None` when there is no node for it.
///
/// A comparison is bound like an arithmetic operator, because the node carries `return_bool`
/// separately: `x > 0` and `x > bool 0` are one operator with one field changed. [`arithmetic_op`]
/// is not reused because it decides what a *fold* may do at compile time and a comparison folds
/// nothing, while this map decides what a *node* computes once the query runs.
/// A comparison carries `return_bool` separately, so `x > 0` and `x > bool 0` use the same
/// operator. Unlike [`arithmetic_op`], which decides what a constant fold may compute, this map
/// selects the operation the runtime executes.
fn binary_op(op: &TokenType) -> Option<BinaryOp> {
match op.id() {
token::T_ADD => Some(BinaryOp::Add),
src/query/src/promql/native_compile/tests.rs
+178 -21
View File
@@ -31,7 +31,7 @@ use promql::native::plan::{
Grouping, MapOp, NativeExpr, NativeNode, NativeTree, NativeType,
};
use promql::range_eval::RangeFunction;
use promql::vector_match::{BinaryMatching, BinaryOp};
use promql::vector_match::{BinaryMatching, BinaryOp, Cardinality, MatchingLabels};
use promql_parser::parser::{
AtModifier, BinModifier, EvalStmt, Expr as PromExpr, NumberLiteral, Offset,
VectorMatchFillValues,
@@ -41,6 +41,27 @@ use super::{
CompiledInput, SelectorRequirement, SelectorTiming, compile, compile_selectors, validate_inputs,
};
/// Whether a matching names its labels with `on` (`true`) or `ignoring` (`false`), and which.
fn matching_labels(labels: &MatchingLabels) -> (bool, Vec<&str>) {
match labels {
MatchingLabels::On(names) => (true, names.iter().map(|name| &**name).collect()),
MatchingLabels::Ignoring(names) => (false, names.iter().map(|name| &**name).collect()),
}
}
/// Which cardinality a matching declares, and the labels it copies from the one side.
fn matching_cardinality(cardinality: &Cardinality) -> (&'static str, Vec<&str>) {
match cardinality {
Cardinality::OneToOne => ("one-to-one", Vec::new()),
Cardinality::ManyToOne(names) => {
("many-to-one", names.iter().map(|name| &**name).collect())
}
Cardinality::OneToMany(names) => {
("one-to-many", names.iter().map(|name| &**name).collect())
}
}
}
/// The raw series schema a caller binds from the table: one field and no evaluation grid.
fn raw_schema() -> Arc<SeriesSchema> {
Arc::new(SeriesSchema {
@@ -1623,33 +1644,169 @@ fn a_constant_operand_binds_the_value_it_folds_to() {
}
}
/// Binary operations between vectors remain unsupported in the native lowering stage.
/// Two vectors bind independently, while the compatibility entry refuses to hide the two inputs.
#[test]
fn an_operation_between_two_vectors_is_refused_for_the_missing_selector_identity() {
for query in [
"a + b",
"a > b",
"rate(a[5m]) * rate(b[5m])",
"sum(a) + sum(b)",
fn a_two_vector_operation_binds_one_input_per_selector() {
let compiled = compile_selectors(
&eval_stmt("a + b"),
&|_| Ok(raw_schema()),
SeriesOrdering::KeyThenTime,
)
.unwrap_or_else(|error| panic!("`a + b` was refused: {error}"));
assert_eq!(compiled.inputs.len(), 2);
assert_eq!(compiled.inputs[0].requirement.metric.as_deref(), Some("a"));
assert_eq!(compiled.inputs[1].requirement.metric.as_deref(), Some("b"));
assert!(matches!(
compiled.tree.root().expr,
NativeExpr::Binary { .. }
));
let error = compile(
&eval_stmt("a + b"),
raw_schema(),
SeriesOrdering::KeyThenTime,
)
.expect_err("`a + b` requires separately bound selector inputs");
assert!(matches!(error, DataFusionError::NotImplemented(_)));
assert!(
error.to_string().contains("use `compile_selectors`"),
"{error}"
);
}
#[test]
fn vector_binary_matching_binds_labels_cardinality_fills_and_bool() {
for (query, on, labels, card, copies, lhs, rhs, bool_result) in [
(
"a + on(job) b",
true,
vec!["job"],
"one-to-one",
vec![],
None,
None,
false,
),
(
"a / ignoring(zone) b",
false,
vec!["zone"],
"one-to-one",
vec![],
None,
None,
false,
),
(
"a + on(job) group_left(zone) b",
true,
vec!["job"],
"many-to-one",
vec!["zone"],
None,
None,
false,
),
(
"a - on(job) group_right(zone) b",
true,
vec!["job"],
"one-to-many",
vec!["zone"],
None,
None,
false,
),
(
"a + fill_left(0) b",
false,
vec![],
"one-to-one",
vec![],
Some(0.0),
None,
false,
),
(
"a + fill_right(1.5) b",
false,
vec![],
"one-to-one",
vec![],
None,
Some(1.5),
false,
),
(
"a + fill(2) b",
false,
vec![],
"one-to-one",
vec![],
Some(2.0),
Some(2.0),
false,
),
(
"a > bool ignoring(zone) b",
false,
vec!["zone"],
"one-to-one",
vec![],
None,
None,
true,
),
] {
let error = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime)
.expect_err(&format!("`{query}` compiled, and it should not have"));
assert!(
matches!(error, DataFusionError::NotImplemented(_)),
"`{query}` did not report NotImplemented: {error}"
let compiled = compile_selectors(
&eval_stmt(query),
&|_| Ok(raw_schema()),
SeriesOrdering::KeyThenTime,
)
.unwrap_or_else(|error| panic!("`{query}` was refused: {error}"));
let (_, _, matching) = binary_parts(compiled.tree.root());
assert_eq!(matching_labels(&matching.labels), (on, labels), "{query}");
assert_eq!(
matching_cardinality(&matching.cardinality),
(card, copies),
"{query}"
);
let message = error.to_string();
assert!(
message.contains("between two vectors"),
"`{query}` did not say what it found: {message}"
);
assert!(
message.contains("selector"),
"`{query}` did not name what an input is missing: {message}"
assert_eq!(
(matching.fill.lhs, matching.fill.rhs),
(lhs, rhs),
"{query}"
);
assert_eq!(matching.return_bool, bool_result, "{query}");
}
}
#[test]
fn vector_binary_matching_keeps_ignoring_and_group_left_separate() {
let compiled = compile_selectors(
&eval_stmt("a / ignoring(instance) group_left(host) b"),
&|_| Ok(raw_schema()),
SeriesOrdering::KeyThenTime,
)
.unwrap();
let (_, _, matching) = binary_parts(compiled.tree.root());
assert_eq!(matching_labels(&matching.labels), (false, vec!["instance"]));
assert_eq!(
matching_cardinality(&matching.cardinality),
("many-to-one", vec!["host"])
);
}
#[test]
fn vector_binary_set_operators_remain_rejected_explicitly() {
let error = compile_selectors(
&eval_stmt("a and b"),
&|_| Ok(raw_schema()),
SeriesOrdering::KeyThenTime,
)
.expect_err("set operators have no binary value node");
assert!(error.to_string().contains("set operator `and`"), "{error}");
}
/// A set operator has no node at all, so it is refused as the set operator it is rather than as an
/// arithmetic operation the compiler could not bind: `and`, `or` and `unless` combine two vectors'
/// series, which is not what a binary value node does.
src/query/tests/native_multi_selector.rs
+935
View File
@@ -0,0 +1,935 @@
// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! A two-selector query compiled by the real compiler and answered by the real executor.
//!
//! This is an external integration test: the tree comes from
//! [`compile_selectors`], never from a hand-built node, and it is run through
//! [`promql::native::execute`] the way a consumer above these crates would run it. What it checks
//! is what per-selector binding exists for: `a + b` reads two inputs, each selector's samples are
//! fed at the position the compiler says that selector's input sits at, and the operation matches
//! the two vectors by the labels the query wrote.
//!
//! Every case asserts the series count, the label set of each output series and the value at every
//! grid point, because a case that only asserted "something came out" would pass under a wrong label
//! mapping. The grid is five points one second apart starting at the Unix epoch, and the statement's
//! lookback is five seconds, so an instant selection carries the last sample of the previous points
//! forward.
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::{Duration, UNIX_EPOCH};
use datafusion::execution::memory_pool::{GreedyMemoryPool, MemoryPool};
use datatypes::arrow::array::Float64Array;
use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
use datatypes::series::samples::{FieldSpec, SampleRef, Samples, SeriesFields, SeriesSchema};
use datatypes::series::{Label, SeriesLabels, SeriesOrdering};
use promql::evaluated::EvaluatedChunk;
use promql::native::execute::{NativeExecutionState, NativeExecutor};
use promql::native::plan::NativeExpr;
use query::promql::native_compile::SelectorRequirement;
use query::promql::native_compile::{CompiledTree, compile_selectors};
/// The layout every selector of these queries is fed in: one float field of raw samples.
fn raw_schema() -> Arc<SeriesSchema> {
Arc::new(SeriesSchema {
fields: Arc::new([FieldSpec {
name: "value".into(),
}]),
eval_grid: None,
})
}
/// `query` over `0..=4000ms` on a `1000ms` grid with a five-second lookback, which is five points.
fn eval_stmt(query: &str) -> promql_parser::parser::EvalStmt {
promql_parser::parser::EvalStmt {
expr: promql_parser::parser::parse(query).unwrap(),
start: UNIX_EPOCH,
end: UNIX_EPOCH + Duration::from_millis(4_000),
interval: Duration::from_millis(1_000),
lookback_delta: Duration::from_millis(5_000),
}
}
/// `query` compiled by the real compiler, with its tree and the inputs it reads.
fn compile(query: &str) -> CompiledTree {
let schema = raw_schema();
let bindings = |requirement: &SelectorRequirement| {
let field = requirement
.matchers
.matchers
.iter()
.find(|matcher| matcher.name == "__field__")
.map_or("value", |matcher| matcher.value.as_str());
if field == "value" {
Ok(Arc::clone(&schema))
} else {
Ok(Arc::new(SeriesSchema {
fields: Arc::new([FieldSpec { name: field.into() }]),
eval_grid: None,
}))
}
};
compile_selectors(&eval_stmt(query), &bindings, SeriesOrdering::KeyThenTime)
.unwrap_or_else(|error| panic!("`{query}` was refused: {error}"))
}
/// The metric `name` with `labels`, as a source series of the samples a scan would read.
fn metric(name: &str, labels: &[(&str, &str)]) -> SeriesLabels {
let tags = labels
.iter()
.map(|(name, value)| Label {
name: (*name).into(),
value: (*value).into(),
})
.collect();
SeriesLabels::try_new(Some(name.into()), tags).expect("distinct label names")
}
/// One raw chunk of `series`, holding `rows` at their source timestamps, closing the series.
///
/// The chunk is built against `schema`, the layout the compiler bound that selector to, so a chunk
/// the input would refuse cannot be pushed here by accident.
fn raw_chunk(
schema: &SeriesSchema,
key: &str,
series: SeriesLabels,
rows: &[(i64, f64)],
) -> RawSeriesChunk {
let timestamps: Vec<i64> = rows.iter().map(|&(time, _)| time).collect();
let values: Vec<f64> = rows.iter().map(|&(_, value)| value).collect();
let fields = schema
.fields
.iter()
.map(|_| Samples::from_float(Float64Array::from(values.clone())))
.collect();
RawSeriesChunk::try_new(
schema,
Buffer::from_slice_ref(key.as_bytes()),
Some(series),
RawTimestamps::try_new(ScalarBuffer::from(timestamps)).expect("increasing timestamps"),
SeriesFields::try_from_vec(fields).unwrap(),
true,
)
.expect("the chunk is bound to the schema it is built against")
}
/// Builds only source rows in the requirement's metric/matcher/time/field selection.
fn scan_rows(
requirement: &SelectorRequirement,
metric_name: &str,
labels: &[(&str, &str)],
rows: &[(i64, f64)],
) -> Vec<(i64, f64)> {
if requirement
.metric
.as_deref()
.is_some_and(|name| name != metric_name)
{
return Vec::new();
}
if !requirement.matchers.matchers.iter().all(|matcher| {
let value = if matcher.name == "__name__" {
metric_name
} else {
labels
.iter()
.find(|(name, _)| *name == matcher.name)
.map_or("", |(_, value)| *value)
};
matcher.is_match(value)
}) {
return Vec::new();
}
let (start, end, inclusive_start) = requirement.raw_scan_bounds().unwrap();
rows.iter()
.copied()
.filter(|(time, _)| (*time > start || (inclusive_start && *time == start)) && *time <= end)
.collect()
}
/// Scans the small fixture source independently for each compiled input and pushes only selected rows.
fn scan_push(
compiled: &CompiledTree,
input: usize,
key: &str,
metric_name: &str,
labels: &[(&str, &str)],
rows: &[(i64, f64)],
) -> Option<(usize, RawSeriesChunk)> {
let input_spec = &compiled.inputs[input];
let rows = scan_rows(&input_spec.requirement, metric_name, labels, rows);
(!rows.is_empty()).then(|| {
(
input,
raw_chunk(
input_spec.schema.as_ref(),
key,
metric(
metric_name,
&labels
.iter()
.copied()
.filter(|(name, _)| *name != "__field__")
.collect::<Vec<_>>(),
),
&rows,
),
)
})
}
/// Runs `compiled` over `pushes`, each naming the input its chunk belongs to by that input's index.
///
/// A binary node retains both operands and decides nothing until its input ends, so every pull
/// before [`NativeExecutor::finish`] has to answer with nothing; a pull that answered earlier would
/// mean a node computed over half a vector.
fn run(compiled: &CompiledTree, pushes: Vec<(usize, RawSeriesChunk)>) -> Vec<EvaluatedChunk> {
let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 20));
let state = NativeExecutionState::try_new(&compiled.tree, Arc::clone(&pool))
.expect("the runtime executes this tree");
let mut input_ids = Vec::with_capacity(compiled.inputs.len());
for input in compiled.inputs.iter() {
let ids: Vec<_> = state.input_ids(input.source_index).collect();
assert_eq!(
ids.len(),
1,
"each compiled source has exactly one runtime input"
);
input_ids.push(ids[0]);
}
let mut executor = NativeExecutor::new(state);
for (index, chunk) in pushes {
executor
.push_raw(input_ids[index], chunk)
.expect("the chunk is pushed to its compiled input");
let pulled = executor.next_chunk().expect("the pull succeeds");
assert!(
pulled.is_none(),
"binary output arrived before both inputs ended: {pulled:?}"
);
}
let chunks = executor.finish().expect("every series closed");
assert_eq!(executor.state().reserved_bytes(), 0, "nothing held");
assert_eq!(pool.reserved(), 0, "a finished execution reserves nothing");
chunks
}
/// The label set a chunk's series is filed under, rendered with the metric name first.
fn label_set(chunk: &EvaluatedChunk) -> String {
let metric = &chunk.series.metric;
let mut parts = Vec::new();
if let Some(name) = metric.name() {
parts.push(format!("__name__=\"{name}\""));
}
for label in metric.tags() {
parts.push(format!("{}=\"{}\"", label.name, label.value));
}
format!("{{{}}}", parts.join(", "))
}
/// What every output series holds, as its rendered label set and its value at each grid position.
///
/// A series may arrive as one chunk or as several: a chunk says which series it carries and closes
/// it, and how a node batches what it produces is not this test's to pin. Rows are therefore
/// appended to the entry for their label set, and a series split over two chunks reads as one.
fn results(chunks: &[EvaluatedChunk]) -> BTreeMap<String, Vec<(usize, f64)>> {
let mut output: BTreeMap<String, Vec<(usize, f64)>> = BTreeMap::new();
for chunk in chunks {
let field = chunk
.series
.fields
.as_slice()
.first()
.expect("the result carries the one field it was bound to");
let entry = output.entry(label_set(chunk)).or_default();
for (position, row) in chunk.series.time.positions() {
match field.sample(row) {
SampleRef::Float(value) => entry.push((position, value)),
_ => panic!("the row at {row} holds no float"),
}
}
}
output
}
/// The value at every one of the five grid points.
fn every_point(value: f64) -> Vec<(usize, f64)> {
(0..5).map(|position| (position, value)).collect()
}
/// `a + b` over two genuinely different inputs: different metric names, different matcher sets and
/// different samples. Each input is fed at the position its own selector was bound to, and the sum
/// is asserted point by point.
#[test]
fn two_selectors_are_summed_from_two_inputs() {
let compiled = compile("a{job=\"x\"} + b{job=\"x\", instance=\"i1\"}");
assert_eq!(
compiled.inputs.len(),
2,
"`a + b` did not bind one input per selector"
);
assert_eq!(
compiled.inputs[0].requirement.metric.as_deref(),
Some("a"),
"the first input is not the left selector's"
);
assert_eq!(
compiled.inputs[1].requirement.metric.as_deref(),
Some("b"),
"the second input is not the right selector's"
);
// The two selectors asked for different series, so each input records its own matcher set.
assert_eq!(
compiled.inputs[0].requirement.matchers.matchers.len(),
1,
"the left input did not record the one matcher its selector wrote"
);
assert_eq!(
compiled.inputs[1].requirement.matchers.matchers.len(),
2,
"the right input did not record the two matchers its selector wrote"
);
assert_ne!(
compiled.inputs[0].source_index, compiled.inputs[1].source_index,
"two selectors were bound to one position"
);
// `a` holds 1 at t=0 and 2 at t=2000, so an instant selection carries 1 forward to p0 and p1 and
// 2 to p2..p4. `b` holds 10 at t=0 and 40 at t=4000: `[10, 10, 10, 10, 40]`.
let chunks = run(
&compiled,
vec![
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-job=x-instance=i1",
metric("a", &[("instance", "i1"), ("job", "x")]),
&[(0, 1.0), (2_000, 2.0)],
),
),
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"b-job=x-instance=i1",
metric("b", &[("instance", "i1"), ("job", "x")]),
&[(0, 10.0), (4_000, 40.0)],
),
),
],
);
let output = results(&chunks);
assert_eq!(output.len(), 1, "the sum is not one series: {output:?}");
// Arithmetic drops the metric name and keeps the labels the match was made on.
assert_eq!(
output.get("{instance=\"i1\", job=\"x\"}"),
Some(&vec![(0, 11.0), (1, 11.0), (2, 12.0), (3, 12.0), (4, 42.0)]),
"the sum is not the values each side selected: {output:?}"
);
}
/// `a / on(job) b` matches on `job` alone, so two series that differ in `instance` pair up. The same
/// inputs through `a / b` — the default matching, which matches on every label but `__name__` — pair
/// nothing, and that difference is what proves the `on` list reached the kernel.
#[test]
fn noncommutative_vector_operations_keep_left_right_values() {
for (query, expected) in [("a / on() b", 0.5), ("a - on() b", -1.0)] {
let compiled = compile(query);
let chunks = run(
&compiled,
vec![
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"left",
metric("a", &[]),
&[(0, 2.0)],
),
),
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"right",
metric("b", &[]),
&[(0, 4.0)],
),
),
],
);
let output = results(&chunks);
assert_eq!(output.len(), 1, "{query}: {output:?}");
assert_eq!(
output.values().next().unwrap(),
&every_point(expected),
"{query}"
);
}
}
#[test]
fn an_on_matching_pairs_series_the_default_matching_would_not() {
// Both inputs are the same in both runs: only the compiled matching differs.
let pushes = |compiled: &CompiledTree| {
vec![
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-job=x-instance=i1",
metric("a", &[("instance", "i1"), ("job", "x")]),
&[(0, 1.0), (2_000, 2.0)],
),
),
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"b-job=x-instance=i2",
metric("b", &[("instance", "i2"), ("job", "x")]),
&[(0, 10.0), (4_000, 40.0)],
),
),
]
};
let matched = compile("a{job=\"x\"} / on(job) b{job=\"x\"}");
let output = results(&run(&matched, pushes(&matched)));
assert_eq!(output.len(), 1, "`on(job)` did not match: {output:?}");
// `on` keeps only the labels it matched on, so `instance` is gone.
assert_eq!(
output.get("{job=\"x\"}"),
Some(&vec![
(0, 0.1),
(1, 0.1),
(2, 0.2),
(3, 0.2),
(4, 2.0 / 40.0),
]),
"`on(job)` did not divide each side's value: {output:?}"
);
// The default matching matches on `instance` too, and the two sides disagree on it.
let unmatched = compile("a{job=\"x\"} / b{job=\"x\"}");
let output = results(&run(&unmatched, pushes(&unmatched)));
assert!(
output.is_empty(),
"the default matching paired series that differ in `instance`: {output:?}"
);
}
/// `group_left(env)` reuses one right series for two left series and copies `env` from the right
/// side into each result, which is the cardinality a one-to-one matching cannot express.
#[test]
fn a_group_left_reuses_the_one_side_and_copies_its_label() {
let compiled = compile("a{job=\"x\"} / on(job) group_left(env) b{job=\"x\"}");
let chunks = run(
&compiled,
vec![
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-job=x-instance=i1",
metric("a", &[("instance", "i1"), ("job", "x")]),
&[(0, 1.0)],
),
),
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-job=x-instance=i3",
metric("a", &[("instance", "i3"), ("job", "x")]),
&[(0, 2.0)],
),
),
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"b-env=prod-job=x",
metric("b", &[("env", "prod"), ("job", "x")]),
&[(0, 10.0)],
),
),
],
);
let output = results(&chunks);
// One right series matched both left series, so both are answered, each keeping its own
// `instance` and gaining the copied `env`.
assert_eq!(
output.len(),
2,
"the left series was not reused: {output:?}"
);
assert_eq!(
output.get("{env=\"prod\", instance=\"i1\", job=\"x\"}"),
Some(&every_point(0.1)),
"the first left series did not keep its own labels beside the copied one: {output:?}"
);
assert_eq!(
output.get("{env=\"prod\", instance=\"i3\", job=\"x\"}"),
Some(&every_point(0.2)),
"the second left series did not keep its own labels beside the copied one: {output:?}"
);
}
/// `fill_right(100)` stands in for a right sample a left series found no match for: `a * on(job) b`
/// drops the left series `job="y"`, and the filled operation answers it from the fill value.
#[test]
fn a_fill_answers_a_left_series_that_found_no_match() {
let pushes = |compiled: &CompiledTree| {
vec![
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-job=x",
metric("a", &[("job", "x")]),
&[(0, 2.0)],
),
),
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-job=y",
metric("a", &[("job", "y")]),
&[(0, 3.0)],
),
),
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"b-job=x",
metric("b", &[("job", "x")]),
&[(0, 10.0)],
),
),
]
};
let unfilled = compile("a * on(job) b");
let output = results(&run(&unfilled, pushes(&unfilled)));
assert_eq!(
output.len(),
1,
"a series with no match was answered without a fill: {output:?}"
);
assert_eq!(
output.get("{job=\"x\"}"),
Some(&every_point(20.0)),
"the matched series is not the product of the two sides: {output:?}"
);
let filled = compile("a * on(job) fill_right(100) b");
let output = results(&run(&filled, pushes(&filled)));
assert_eq!(
output.len(),
2,
"`fill_right(100)` did not answer the unmatched series: {output:?}"
);
assert_eq!(
output.get("{job=\"x\"}"),
Some(&every_point(20.0)),
"the fill changed the series that did match: {output:?}"
);
assert_eq!(
output.get("{job=\"y\"}"),
Some(&every_point(300.0)),
"the filled series is not `3 * 100`: {output:?}"
);
}
/// `group_right(env)` makes the right operand the many side and copies `env` from the left one, so
/// one left series answers two right series.
#[test]
fn group_right_answers_each_right_series_and_copies_its_label() {
let pushes = |compiled: &CompiledTree| {
vec![
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-env=prod-job=x",
metric("a", &[("env", "prod"), ("job", "x")]),
&[(0, 2.0)],
),
),
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"b-job=x-instance=i1",
metric("b", &[("instance", "i1"), ("job", "x")]),
&[(0, 10.0)],
),
),
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"b-job=x-instance=i2",
metric("b", &[("instance", "i2"), ("job", "x")]),
&[(0, 20.0)],
),
),
]
};
let compiled = compile("a{job=\"x\"} / on(job) group_right(env) b");
let output = results(&run(&compiled, pushes(&compiled)));
assert_eq!(
output.len(),
2,
"the many side was not answered once per right series: {output:?}"
);
assert_eq!(
output.get("{env=\"prod\", instance=\"i1\", job=\"x\"}"),
Some(&every_point(0.2)),
"the first right series is not `2 / 10`: {output:?}"
);
assert_eq!(
output.get("{env=\"prod\", instance=\"i2\", job=\"x\"}"),
Some(&every_point(0.1)),
"the second right series is not `2 / 20`: {output:?}"
);
}
/// `fill_left` and `fill_right` fill the side written in the query, also under `group_right`, where
/// the many side is the right one: prometheus/prometheus#19939. The two values differ so a swap is
/// visible. `fill_left(5)` answers a right series with no left counterpart as `5 - 4`, and
/// `fill_right(7)` answers a left series with no right counterpart as `2 - 7`; the pre-fix kernel
/// had the two trade places, answering `7 - 4` and `2 - 5`.
#[test]
fn group_right_fills_the_side_the_query_wrote() {
let compiled = compile("a{job=\"x\"} - on(job) group_right() fill_left(5) fill_right(7) b");
let pushes = vec![
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-job=x",
metric("a", &[("job", "x")]),
&[(0, 2.0)],
),
),
// This left series has no right counterpart at all, so `fill_right` answers every point.
(
0,
raw_chunk(
compiled.inputs[0].schema.as_ref(),
"a-job=z",
metric("a", &[("job", "z")]),
&[(0, 2.0)],
),
),
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"b-job=x-instance=i1",
metric("b", &[("instance", "i1"), ("job", "x")]),
&[(0, 10.0)],
),
),
// This right series has no left counterpart, so `fill_left` answers every point.
(
1,
raw_chunk(
compiled.inputs[1].schema.as_ref(),
"b-job=y-instance=i2",
metric("b", &[("instance", "i2"), ("job", "y")]),
&[(0, 4.0)],
),
),
];
let output = results(&run(&compiled, pushes));
assert_eq!(output.len(), 3, "every series is answered: {output:?}");
assert_eq!(
output.get("{instance=\"i1\", job=\"x\"}"),
Some(&every_point(-8.0)),
"the matched series is not `2 - 10`: {output:?}"
);
assert_eq!(
output.get("{job=\"z\"}"),
Some(&every_point(-5.0)),
"`fill_right(7)` did not fill the right side the query wrote: {output:?}"
);
assert_eq!(
output.get("{instance=\"i2\", job=\"y\"}"),
Some(&every_point(1.0)),
"`fill_left(5)` did not fill the left side the query wrote: {output:?}"
);
}
#[test]
fn requirements_drive_scans_for_repeated_metric_nested_windows_and_fields() {
let stmt =
eval_stmt("abs(rate(a{__field__=\"value\"}[2s])) + on() rate(a{__field__=\"other\"}[3s])");
let compiled = compile_selectors(
&stmt,
&|requirement| {
let field = requirement
.matchers
.matchers
.iter()
.find(|matcher| matcher.name == "__field__")
.map_or("value", |matcher| matcher.value.as_str());
Ok(Arc::new(SeriesSchema {
fields: Arc::new([FieldSpec { name: field.into() }]),
eval_grid: None,
}))
},
SeriesOrdering::KeyThenTime,
)
.unwrap();
assert_eq!(compiled.inputs.len(), 2);
assert_eq!(
compiled.inputs[0].requirement.metric,
compiled.inputs[1].requirement.metric
);
assert_ne!(
compiled.inputs[0].requirement.raw_scan_bounds().unwrap(),
compiled.inputs[1].requirement.raw_scan_bounds().unwrap()
);
// Each selector binds only its requested field before its selected scan rows reach the executor.
// The two fields rise at different rates, `2` and `10` per second, so a sample that reached the
// other selector's window would not read as the same answer.
let mut pushes = Vec::new();
for (field, rows) in [
(
"value",
&[(0, 1.0), (1_000, 3.0), (2_000, 5.0), (3_000, 7.0)][..],
),
(
"other",
&[(0, 10.0), (1_000, 20.0), (2_000, 30.0), (3_000, 40.0)][..],
),
] {
for index in 0..compiled.inputs.len() {
if let Some(push) = scan_push(&compiled, index, "a", "a", &[("__field__", field)], rows)
{
pushes.push(push);
}
}
}
assert_eq!(
pushes.len(),
2,
"each field requirement selects only its own source field"
);
assert_eq!(pushes[0].0, 0);
assert_eq!(pushes[1].0, 1);
assert!(pushes.iter().all(|(_, chunk)| chunk.fields.len() == 1));
let output_chunks = run(&compiled, pushes);
let output = results(&output_chunks);
assert_eq!(
output.len(),
1,
"the same-metric windows did not match: {output:?}"
);
// The scan behind the two-second window sits under `abs`, which maps over it, and the one
// behind the three-second window under the window itself: each is read at its own path.
assert_eq!(compiled.inputs[0].source_index, 0);
assert_eq!(compiled.inputs[1].source_index, 1);
assert_eq!(compiled.inputs[0].schema.fields[0].name.as_ref(), "value");
assert_eq!(compiled.inputs[1].schema.fields[0].name.as_ref(), "other");
assert_eq!(output_chunks[0].series.fields.len(), 1);
let root_output = compiled
.tree
.root()
.output
.samples()
.expect("the addition reports evaluated samples");
// The operation pairs its operands' fields by position and names the output after the left
// side, which is the field the two-second window read.
assert_eq!(root_output.fields[0].name.as_ref(), "value");
// Each window answers its own source rows: the two-second window over a `2`-per-second field
// answers `1.5`, `2`, `2` and the three-second window over a `10`-per-second field answers `5`,
// `10`, `10`, so their sum is `6.5`, `12`, `12`.
assert_eq!(
output.values().next().unwrap(),
&vec![(1, 6.5), (2, 12.0), (3, 12.0)],
"the two separately scanned rates were not added: {output:?}"
);
let nested = compile("(a + b) / c");
let rows_a = [
(0, 2.0),
(1_000, 2.0),
(2_000, 2.0),
(3_000, 2.0),
(4_000, 2.0),
];
let rows_b = [
(0, 6.0),
(1_000, 6.0),
(2_000, 6.0),
(3_000, 6.0),
(4_000, 6.0),
];
let rows_c = [
(0, 4.0),
(1_000, 4.0),
(2_000, 4.0),
(3_000, 4.0),
(4_000, 4.0),
];
let pushes = [
scan_push(&nested, 0, "a", "a", &[], &rows_a).unwrap(),
scan_push(&nested, 1, "b", "b", &[], &rows_b).unwrap(),
scan_push(&nested, 2, "c", "c", &[], &rows_c).unwrap(),
];
let output = results(&run(&nested, pushes.into()));
assert_eq!(
output.len(),
1,
"nested/repeated selectors did not match: {output:?}"
);
assert_eq!(output.values().next().unwrap(), &every_point(2.0));
}
#[test]
fn compiled_comparisons_preserve_filter_and_bool_semantics() {
for (query, expected_bool) in [("a > b", false), ("a > bool b", true)] {
let compiled = compile(query);
let NativeExpr::Binary { matching, .. } = &compiled.tree.root().expr else {
panic!("`{query}` did not compile to a binary node");
};
assert_eq!(
matching.return_bool, expected_bool,
"comparison mode for `{query}`"
);
let requirement = &compiled.inputs[0].requirement;
assert_eq!(requirement.metric.as_deref(), Some("a"));
let (start, end, inclusive) = requirement.raw_scan_bounds().unwrap();
assert!(start <= 0 && end >= 4_000);
assert!(!inclusive);
let rows_a = [
(0, 1.0),
(1_000, 2.0),
(2_000, 3.0),
(3_000, 4.0),
(4_000, 5.0),
];
let rows_b = [
(0, 0.0),
(1_000, 1.0),
(2_000, 4.0),
(3_000, 2.0),
(4_000, 6.0),
];
let left = scan_push(&compiled, 0, "a", "a", &[], &rows_a).unwrap();
let right = scan_push(&compiled, 1, "b", "b", &[], &rows_b).unwrap();
let output = results(&run(&compiled, vec![left, right]));
let expected = if expected_bool {
BTreeMap::from([(
"{}".to_string(),
vec![(0, 1.0), (1, 1.0), (2, 0.0), (3, 1.0), (4, 0.0)],
)])
} else {
BTreeMap::from([(
"{__name__=\"a\"}".to_string(),
vec![(0, 1.0), (1, 2.0), (3, 4.0)],
)])
};
assert_eq!(
output, expected,
"wrong vector comparison result for `{query}`"
);
}
}
#[test]
fn ignoring_matching_uses_compiled_selectors_and_scanned_inputs() {
let compiled = compile("a{job=\"x\"} / ignoring(instance) b{job=\"x\"}");
let NativeExpr::Binary { matching, .. } = &compiled.tree.root().expr else {
panic!("the expression did not compile to a binary node");
};
assert_eq!(
matching.labels,
promql::vector_match::MatchingLabels::Ignoring(vec![Arc::from("instance")])
);
let rows = [(0, 1.0), (2_000, 2.0), (4_000, 9.0), (5_000, 12.0)];
let left = scan_push(
&compiled,
0,
"left",
"a",
&[("job", "x"), ("instance", "i1")],
&rows,
)
.unwrap();
let right = scan_push(
&compiled,
1,
"right",
"b",
&[("job", "x"), ("instance", "i2")],
&rows,
)
.unwrap();
assert_eq!(left.0, 0);
assert_eq!(right.0, 1);
assert_eq!(left.1.time.len(), 3);
assert_eq!(right.1.time.len(), 3);
let output = results(&run(&compiled, vec![left, right]));
assert_eq!(
output.len(),
1,
"ignoring matching did not pair differing instance labels: {output:?}"
);
assert_eq!(
output.values().next().unwrap(),
&vec![(0, 1.0), (1, 1.0), (2, 1.0), (3, 1.0), (4, 1.0)],
"ignoring match output did not divide the vectors: {output:?}"
);
assert!(scan_push(&compiled, 0, "wrong-metric", "b", &[("job", "x")], &rows).is_none());
assert!(scan_push(&compiled, 0, "wrong-matcher", "a", &[("job", "y")], &rows).is_none());
}
#[test]
fn historical_range_sources_are_not_clipped_to_the_statement_start() {
let stmt = promql_parser::parser::EvalStmt {
expr: promql_parser::parser::parse("sum_over_time(a[2s]) + sum_over_time(a[3s])").unwrap(),
start: UNIX_EPOCH,
end: UNIX_EPOCH,
interval: Duration::from_secs(1),
lookback_delta: Duration::from_secs(5),
};
let compiled =
compile_selectors(&stmt, &|_| Ok(raw_schema()), SeriesOrdering::KeyThenTime).unwrap();
assert_eq!(compiled.inputs.len(), 2);
let rows = [(-2_500, 10.0), (-1_500, 20.0), (-500, 30.0)];
let short = scan_push(&compiled, 0, "a-short", "a", &[], &rows).unwrap();
let long = scan_push(&compiled, 1, "a-long", "a", &[], &rows).unwrap();
assert_eq!(short.1.time.len(), 2);
assert_eq!(long.1.time.len(), 3);
let output = results(&run(&compiled, vec![short, long]));
assert_eq!(
output,
BTreeMap::from([("{}".to_string(), vec![(0, 110.0)])]),
"historical range samples were clipped or a selector reused the other window"
);
}