From d0bbfc6893ffc4cf32ba58dee4eb1a20cb28eb5c Mon Sep 17 00:00:00 2001 From: discord9 Date: Mon, 2 Nov 2026 15:13:50 +0800 Subject: [PATCH] feat(promql): derive canonical ownership and insert the merge consumers need Signed-off-by: discord9 --- src/promql/src/native/canonicalize.rs | 441 +++++ src/promql/src/native/canonicalize/tests.rs | 1704 +++++++++++++++++++ src/promql/src/native/mod.rs | 1 + src/promql/src/native/plan.rs | 996 +++++++++-- src/query/src/promql/planner/native.rs | 65 +- src/query/src/promql/planner/test.rs | 3 +- 6 files changed, 3094 insertions(+), 116 deletions(-) create mode 100644 src/promql/src/native/canonicalize.rs create mode 100644 src/promql/src/native/canonicalize/tests.rs diff --git a/src/promql/src/native/canonicalize.rs b/src/promql/src/native/canonicalize.rs new file mode 100644 index 00000000000..72d07485afb --- /dev/null +++ b/src/promql/src/native/canonicalize.rs @@ -0,0 +1,441 @@ +// 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. + +//! Inserts the canonical merge a consumer of whole series needs - and only where one is needed. +//! +//! A fused input hands over the pieces of a metric one scan partition holds, so a consumer cannot +//! assume its input is grouped by canonical label set. [`NativeExpr::Merge`] is the node that +//! collects the pieces of one canonical label set into one series, and this module is where a tree +//! gains one. It is a narrow pass: a merge holds a whole label set's rows before anything above it +//! runs, and most consumers read the pieces correctly as they arrive, so a merge is inserted at a +//! handful of places and refused everywhere else. +//! +//! # The three conditions +//! +//! A merge is inserted at a consumer only when all three hold, asked in this order: +//! +//! 1. **The consumer needs one unified canonical series.** Some kernels fold a group one point at a +//! time, some read a series' rows as they come, some pick series against each other: only a +//! consumer that needs *one series* per key, rather than one contribution at a time, can use a +//! merge at all. +//! 2. **The input does not already provide it.** [`NativeProperties::grouping`] is that fact: a +//! [`CanonicalGrouping::CanonicalComplete`] input already answers one series per canonical key, +//! so a merge over it would collect nothing. +//! 3. **Co-location is proven for this key.** A merge reads every piece of one canonical label set, +//! which only an execution holding all of them can answer. That is +//! [`merges_are_placeable_in`](crate::native::plan::merges_are_placeable_in)'s question, and it +//! is the caller's: this pass has no placement to read, so it answers conditions 1 and 2 and +//! leaves condition 3 to the route that knows which execution the tree will run in. A merge is +//! never inserted to make an unplaceable tree placeable - a shape the placement cannot answer is +//! declined, not rewritten. +//! +//! None of these is a reason by itself: that the input is +//! [`CanonicalGrouping::SourcePiecesOnly`], that the operator is stateful, that the operator +//! [`reads_across_series`](crate::native::plan::reads_across_series), or that the node is the root. +//! Each of them is true of consumers that answer their input correctly one contribution at a time. +//! +//! # What each consumer needs +//! +//! | consumer | merge? | why | +//! | --- | --- | --- | +//! | [`NativeExpr::Map`], scalar-broadcast [`NativeExpr::Binary`] | no | point-wise per contribution. Label rewrites and filtering comparisons keep `Emitted` occupancy checks; scalar broadcast does not require a unified input series | +//! | [`NativeExpr::Aggregate`], stage `Complete` | no | the fold is per point: `AggregateRuntime::accumulate` folds each chunk into states indexed by `(group, grid position)`, so several contributions to one group at one position are the same accumulation as one contiguous series would give. Completeness of a *group* is the placement's contract (`needs_single_execution_in`), not a merge's | +//! | [`NativeExpr::Aggregate`], stages `Partial`/`Final` | no | `Partial` answers aggregate state, which is not an evaluated vector and has no series for a merge to collect; `Final` folds the state of every partition and is the stage that *is* the merge of contributions | +//! | [`NativeExpr::Window`] | **yes, over a child that proves nothing** | a window spans its input's contributions of one label set and needs one unified pass over them. The runtime does that pass through `MergeInput`, but it builds one only for a child that reports its fragments, and a child that hands series back unchanged says nothing - it reports `shared = false` like a node that has already unified - so nothing collects the fragments and the window answers one series per fragment. [`unifies_its_own_contributions`] is the one rule that separates "already unified" from "cannot say", and this arm collects exactly the children that answer "cannot say": the boundary itself, and a `Vector` with `topk`/`limit_ratio`/`sort` or a filtering `Broadcast` above it | +//! | [`NativeExpr::Binary`] over two vectors, [`NativeExpr::Set`] | no | the kernel files its rows by *result label set* and hands over one series per set, refusing two values of a field at one point (`Results::sort_rows`, `same_labelset`). Matching-key completeness is `matching_co_located`/placement's question, and per-point cardinality is what that refusal enforces | +//! | [`NativeExpr::Vector`], `topk`/`limit_ratio`/`sort` | no, but a window above it collects | each ranks, keeps or orders *series*, and hands back the series it read with their labels: several contributions of one label set are ranked by the value each holds at a point, which is what one whole series would hold there. What it hands over therefore aliases exactly as its input does, which is why it passes [`unifies_its_own_contributions`]'s answer on - and why a window over one of these is the shape the next section's third row is about | +//! | [`NativeExpr::Vector`], `limitk` | **yes** | `k` counts *series* in input order (`select::limitk` walks the group's members and stops at `k`), so pieces of one label set would occupy several slots and be counted against `k`. The runtime merges its input only when the child shares runs, and a scan leaf does not, so the tree has to carry the merge | +//! | [`NativeExpr::Vector`], `histogram_quantile`/`histogram_fraction` | no | the kernel merges label-set pieces itself before it reads buckets (`vector_series` calls `merge_same_labelset`) | +//! | [`NativeExpr::Vector`], `count_values` | no | it counts per group and value at each point, so contributions are folded the way an aggregate folds them | +//! | [`NativeExpr::Vector`], `absent` | no | it reads presence per point over the whole vector, so the union of the contributions is the presence it needs | +//! | [`NativeExpr::Info`] | no | the kernel indexes the info side by `(name, identifying values)` and keeps the newest sample per point, then merges the results that share a label set and refuses a field held twice at one point | +//! | [`NativeExpr::Instant`], [`NativeExpr::Timestamp`] | no | a selection answers one series from one series' rows | +//! | the root | no | the Prometheus response assembler groups batches by label set (`prometheus_resp::merge_batch`); SQL/TQL consumes rows, so neither needs one native chunk per label set. The root remains streaming and avoids an unnecessary collection barrier | +//! +//! Duplicate-position enforcement is specific: `merge_same_labelset` rejects overlapping fields +//! at a position. Window-internal collection and the inserted `limitk` merge enforce it; vector-vector +//! `Binary`/`Set` and `Info` enforce their own duplicate checks. Map occupancy checks label rewrites. +//! Aggregate, `topk`, `limit_ratio`, `sort`, `absent`, `count_values`, and root output do not add a +//! general duplicate-position check; response assembly combines label keys and appends samples. +//! +//! `MergeInput` is what performs the window's joint processing - it orders the contributions, keeps +//! one state per label set, and decides end-of-series from the `is_last` of the collected fragments +//! (`MergeInput::check`, then `merge_same_labelset`, then the evaluator). Whether a window gets one +//! is decided by [`unifies_its_own_contributions`], which `RuntimeNode::build` and this pass both +//! call, so the two cannot disagree about a child. It answers in three values, and this pass acts on +//! exactly one of them: +//! +//! | answer | the nodes that answer it | the runtime | this pass | +//! | --- | --- | --- | --- | +//! | one series per canonical label set | `Aggregate`, `Binary` over two vectors, `Set`, `Info`, `Merge`, `Vector` with `limitk`/`absent`/`histogram_*`/`count_values`, a `Window` whose child reports its fragments | `shared = false`: nothing above collects | nothing to insert - the child already answers what the window needs | +//! | several, and the node says so | `Map`, `Timestamp`, `Instant::Timestamp`, a label-dropping `Broadcast`, a `Window` that neither drops the metric name nor has a child that reports fragments, `Vector` with `topk`/`limit_ratio`/`sort` over one of these | `shared = true`: the window's `MergeInput` collects them | nothing to insert - a merge here would turn that collection off (`Merge` answers "already unified") and move the stream checks to a node without the evaluators | +//! | several, and the node says nothing | a boundary ([`NativeExpr::Input`]), and every node that hands its input's series back unchanged: a filtering `Broadcast`, a `Vector` with `topk`/`limit_ratio`/`sort` | `shared = false`: the window gets no `MergeInput` | **the one the pass collects** | +//! +//! The third row is why the pass exists. A boundary binds the pieces some upstream execution +//! produced and claims nothing about them: a scan hands over one whole storage series at a time - +//! which the streaming evaluator follows chunk by chunk, `RangeEvaluator::push` refusing a chunk +//! that changes the series' primary key or goes back inside it - while a tree split at a repartition +//! (`A -> Repartition -> B`, `B`'s tree `Window(Input)`) starts at an input whose fragments are the +//! pieces of one canonical label set, spread over the executions the repartition fanned them out to. +//! The flag alone cannot tell those apart either: a leaf is not shared because a merge binds samples +//! evaluated on a grid and a scan's are raw, not because its pieces are one series per label set. So +//! the answer is "cannot say" rather than a proof, and every node that hands its input's series back +//! unchanged passes that answer on. [`canonical`] then settles the rest: a raw boundary +//! cannot take a merge and needs none, and a boundary whose [`NativeProperties`] already claim +//! [`CanonicalGrouping::CanonicalComplete`] is not collected again. +//! +//! An `Instant` selection is not one of them: it reads raw samples (`NativeNode::validate` refuses +//! an evaluated input), so it cannot sit between an evaluated boundary and a window at all, and a +//! window over one is the fixed shape rather than a defeating one. +//! +//! Two nodes hand series back unchanged between an evaluated boundary and a window, and both were +//! measured answering two series where one is owed: a `Vector` that keeps the series it ranked, +//! `topk`/`limit_ratio`/`sort` (`avg_over_time(topk(1, x)[5m:1m])`), and a scalar `Broadcast` whose +//! filtering comparison keeps the label set it read (`avg_over_time((x > 0)[5m:1m])`). Both answer +//! "cannot say" and both are collected by [`rewrite`]'s `Window` arm, which states the rule next to +//! the code that applies it. That arm collects them *above* the child, which is the earliest place a +//! merge can sit: `topk` ranks the fragments it was handed, so the merge reunites the pieces the +//! window reads but cannot make `topk` rank them as one series. A collection below `topk` is a +//! different question - it would change what `topk` keeps - and is not what this pass inserts. +//! +//! A *different* label-set collision - two storage series that resolve to one canonical label set - +//! is an error rather than a unification, and every node that can produce one refuses it: the +//! window's own `Emitted` table reports "no merge resolves two values of one field at one position" +//! for two series writing the same position, and the evaluator refuses a changed primary key inside +//! one series. That is Prometheus' rule too (duplicate series is an error), and it is why no +//! collection is warranted there: a merge would have added the two values up instead. +//! +//! # What removing a merge takes with it: the duplicate check +//! +//! [`crate::grid_series::merge_same_labelset`] does two things at once: it collects the pieces of +//! one label set, and it refuses two values of one field at one position under one label set +//! ("vector cannot contain metrics with the same labelset"). The second is not a side effect - it +//! is the error a query owes when two source series resolve to one label set - so every merge this +//! pass used to insert was also the place the tree reported it. +//! +//! Every path this change stopped collecting therefore has to say where that check went, and the +//! answer is per consumer: +//! +//! - `Binary` over two vectors, `Set`: the kernel still makes it. `Results::sort_rows` reports two +//! rows of one result label set at one point, and the label sets it files under are canonical +//! (`key_of`/`label_set`), so the error survives. It is narrower than a merge's - it fires where +//! both operands hold a value, not where either does - which is what Prometheus checks too. +//! - `Map`, `Window`, `Timestamp`, a label-dropping broadcast: `Emitted` (in `execute`) still makes +//! it. Those nodes keep an occupancy table per output label set and refuse a second series +//! writing a position a first one wrote (`occupy`), which is the same check for the same reason. +//! - `Vector` with `histogram_quantile`/`histogram_fraction`: the kernel calls +//! `merge_same_labelset` itself, so the check is exactly where it was. +//! - `Info`: the kernel keeps it. Its results are filed by `label_set` and each label set's rows +//! are sorted by point, where a field held twice records "vector cannot contain metrics with the +//! same labelset" itself. +//! - `Vector` with `limitk`: the merge this pass keeps *is* the check. `MergeInput::check` refuses +//! a stream that breaks its contract and `merge_same_labelset` refuses two values of one field at +//! one position, so nothing is lost here. +//! - **`Aggregate`, `Vector` with `topk`/`limit_ratio`/`sort`/`absent`/`count_values`, and the +//! root: nothing checks.** Those kernels fold, rank, count or emit whatever they are handed, and +//! two fragments of one label set now pass through unremarked. That is a real loss of an error +//! check, and it is the reason this change is not the whole of the work: Part C has to separate +//! the duplicate check from the collection - one pass over the evaluated input that refuses +//! `(label set, position)` held twice - and it is required before the removal is safe. The spots +//! are marked in [`rewrite`]'s arms. +//! +//! How exposed each of those is depends on where a same-position duplicate can even come from. One +//! metric's own pieces cannot make one: storage holds one value per (series, timestamp), so two +//! fragments of one canonical label set hold disjoint positions. A duplicate needs either a label +//! rewrite upstream - and every rewriting node keeps its `Emitted` table above - or two scan +//! partitions that overlap and hand over the same sample twice. The second is the case no node +//! below a rewrite covers, and it is the one Part C has to answer. +//! +//! The pass reads [`NativeProperties`] from the tree rather than storing them on it, so a rewritten +//! tree keeps no fact that the rewrite itself invalidated: each node's facts are derived again from +//! its rewritten children. + +use datafusion::common::Result as DfResult; + +use crate::native::plan::{ + BoundVectorOp, CanonicalGrouping, NativeExpr, NativeNode, NativeProperties, NativeTree, + NativeType, unifies_its_own_contributions, +}; + +#[cfg(test)] +pub(crate) mod tests; + +/// Rewrites `tree` so that every consumer that needs one canonical series is handed one. +/// +/// `inputs` holds what the caller knows about the fused inputs the tree was compiled over, indexed +/// by the `source_index` of the [`BoundInput`](crate::native::plan::BoundInput) each leaf was bound +/// to; a boundary the caller says nothing about proves nothing. The index is what a leaf reads, so a +/// tree with several sources is judged source by source: one scan's proof is never another's. What +/// this adds is only a [`NativeExpr::Merge`] over a node whose fragments may hold one label set: +/// every operation binding and every `source_index` stays where it was, so the rewrite changes which +/// series a consumer reads and not what the query computes. +/// +/// Where a merge goes is decided by what each node needs of its input, and the module documentation +/// is the table: two consumers need one series per canonical key that their input cannot already +/// provide. `limitk` counts series in input order, and a window spanning its input's contributions +/// needs one unified pass and gets none from the runtime when it sits on a fused input. Every other +/// node reads its input as it arrives, and what a merge *above* such a node would collect is a +/// question about that node's output, asked at the consumer of the output and not here. +/// +/// Running the pass over its own output changes nothing: each insertion puts a merge over an input +/// that then reports [`CanonicalGrouping::CanonicalComplete`], so a second run finds nothing to add, +/// and the window's merge is not re-inserted either - the window then sits on a +/// [`NativeExpr::Merge`] and not on the boundary this pass collects. +/// +/// The result is checked by [`NativeTree::try_new`], because a merge has to sit over an evaluated +/// vector and the pass may only put it there. +/// +/// See the module documentation for what removing a merge takes with it: the duplicate-label-set +/// check a collection used to perform. Part C has to restore it for the paths this pass stopped +/// collecting, and the arms of [`rewrite`] mark which those are. +pub fn insert_native_merge_inside_current_fused( + tree: &NativeTree, + inputs: &[NativeProperties], +) -> DfResult { + let root = rewrite(tree.root(), inputs)?; + NativeTree::try_new(root) +} + +/// Wraps `node` in a [`NativeExpr::Merge`] unless it already answers one series per canonical label +/// set. +/// +/// This is conditions 1 and 2 of the module's rule asked of one node: the caller has already decided +/// that this consumer needs one unified series (condition 1), and what is left is whether the input +/// provides it (condition 2) - [`CanonicalGrouping`] is that fact. Condition 3, co-location, is the +/// placement's question and never this function's. +/// +/// A node whose output is not an evaluated vector keeps its place: a merge binds samples evaluated on +/// a grid, so a scalar, raw samples and aggregate state have nothing a merge could take, and no +/// consumer asks for one over those. +/// +/// The facts are read through [`NativeNode::properties`], which derives them from the rewritten node +/// and the boundary every time. A merge proves one series per label set from the pieces it was +/// handed, and it cannot claim that a fragment which never arrived was collected: that is why the +/// ownership it reports stays its input's, unchanged. +fn canonical(node: NativeNode, boundary: &[NativeProperties]) -> NativeNode { + let properties = node.properties(boundary); + if properties.grouping == CanonicalGrouping::CanonicalComplete { + return node; + } + if !matches!(&node.output, NativeType::Samples(schema) if schema.eval_grid.is_some()) { + return node; + } + // A merge reports its input's layout: the same fields on the same grid. + let output = node.output.clone(); + NativeNode { + expr: NativeExpr::Merge { + input: Box::new(node), + }, + output, + } +} + +/// Rewrites `node` bottom-up: its children first, then the merge the consumer needs of them. +/// +/// `boundary` is what the caller knows about the fused inputs this tree was compiled over, indexed by +/// `source_index`. It is the one thing every node's facts are derived from, leaves included, so the +/// rewrite never has to say which of its own lists a node should read: [`NativeNode::properties`] +/// asks a leaf for its own entry and every other node for its children's. +/// +/// Each arm states why that consumer needs or does not need a merge; the module documentation is the +/// same table in one place. +fn rewrite(node: &NativeNode, boundary: &[NativeProperties]) -> DfResult { + let expr = match &node.expr { + // A source is the boundary it was bound to and there is nothing below it to rewrite. A + // constant reads nothing. + NativeExpr::Input(_) | NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } => { + node.expr.clone() + } + NativeExpr::VectorOf { input } => NativeExpr::VectorOf { + input: Box::new(rewrite(input, boundary)?), + }, + // A selection answers one series from one series' rows, so it reads the pieces as they are. + NativeExpr::Instant { input, op } => NativeExpr::Instant { + input: Box::new(rewrite(input, boundary)?), + op: op.clone(), + }, + // A merge is the canonicalizer itself, so it asks nothing of its input: fragments are what it + // was written to receive. + NativeExpr::Merge { input } => NativeExpr::Merge { + input: Box::new(rewrite(input, boundary)?), + }, + // A window may receive several independent contributions of one label set, and a window that + // spans them needs one correct unified pass. The runtime does that pass through `MergeInput` + // when its child reports `shared`, and `unifies_its_own_contributions` is what decides both + // that flag and this arm, so the two cannot disagree about a child. Its three answers split + // the children into three groups, and only the third needs a merge: + // + // - `Some(true)`: the child has established one series per label set (`Aggregate`, `Binary` + // over two vectors, `Set`, `Info`, `Merge`, a `Vector` that reads the whole vector), or it + // is a window that collects its own fragments. Nothing collects what needs no collecting. + // - `Some(false)`: the child may hand over several and *says so* - a `Map`, a `Timestamp`, an + // `Instant` selection reporting a timestamp, a label-dropping `Broadcast`, a `Window`, a + // `Vector` over such a child. `RuntimeNode::build` gives the window its `MergeInput` here, + // and a merge in the tree would turn that off (`NativeExpr::Merge` reports + // `unifies_its_own_contributions = Some(true)`, so the window would stream an already-merged + // series) and move the stream checks to a node without the evaluators. + // - `None`: the child cannot say. A boundary ([`NativeExpr::Input`]) is the node that answers + // this, and a node that hands its input's series back unchanged passes the answer on: a + // `Vector` with `topk`/`limit_ratio`/`sort` or a filtering `Broadcast`. Nothing below such + // a child unified anything and nothing above it will, so this is the one child the pass has + // to collect - and `canonical` still asks the rest: whether the caller's facts already claim + // [`CanonicalGrouping::CanonicalComplete`] for that boundary (a boundary that proves one + // series per label set is not collected again) and whether what the child hands over is a + // shape a merge can take at all (an evaluated vector; a scan's raw samples cannot take one, + // and they need none, because the runtime streams one whole storage series at a time and + // the evaluator follows it chunk by chunk). + // + // The two measured shapes this closes are `avg_over_time(topk(1, x)[5m:1m])` + // (`Window(Vector(Input))`) and `avg_over_time((x > 0)[5m:1m])` + // (`Window(Binary(Input, Scalar))`), each of which answered one series per fragment before: + // both children are `None` here, because both hand the boundary's fragments on unchanged. + NativeExpr::Window { input, op } => { + let input = rewrite(input, boundary)?; + // The child this pass has to collect: one that neither proves it hands over one series + // per label set nor reports that it hands over several. Read as a negation of + // "proves" - `Some(false)` as well as `None` - this would collect a `Map`, a + // `Timestamp`, a name-dropping `Window` and a label-dropping `Broadcast` too, and turn + // off the `MergeInput` the runtime builds for each of them: the blanket rule this pass + // exists to be narrower than. The two answers `false` covers are different facts, and + // only `None` is the one nothing else collects; the merge-count table in this module's + // tests pins both sides of that line. + // + // `canonical` asks the rest of the question - whether the caller's facts already say + // this input answers one series per label set, and whether what it hands over is a + // shape a merge can take. Co-location, the third condition, stays the caller's. + let input = if unifies_its_own_contributions(&input).is_none() { + canonical(input, boundary) + } else { + input + }; + NativeExpr::Window { + input: Box::new(input), + op: op.clone(), + } + } + // A map is point-wise, so it reads each piece as it is; what it does to a label set is a fact + // about its output, not a reason to collect its input. + NativeExpr::Map { input, op } => NativeExpr::Map { + input: Box::new(rewrite(input, boundary)?), + op: op.clone(), + }, + NativeExpr::ParameterizedMap { + input, + op, + parameters, + } => NativeExpr::ParameterizedMap { + input: Box::new(rewrite(input, boundary)?), + op: *op, + parameters: parameters + .iter() + .map(|n| rewrite(n, boundary)) + .collect::>()?, + }, + // A timestamp reads one series' positions, so it reads each piece as it is. + NativeExpr::Timestamp { input } => NativeExpr::Timestamp { + input: Box::new(rewrite(input, boundary)?), + }, + // `info` indexes the info side by `(name, identifying values)` and keeps the newest sample of + // each point, then merges the results that share a label set: both sides are read whole by a + // kernel that does the grouping itself, so neither side is collected here. + // + // The duplicate check is the kernel's on both sides: its results are filed by `label_set` + // and a field held twice at one point records "vector cannot contain metrics with the same + // labelset", while the info side's own collision records "found duplicate series for info + // metric". Removing the merges here therefore took no check with it. + NativeExpr::Info { + base, + info, + selector, + } => NativeExpr::Info { + base: Box::new(rewrite(base, boundary)?), + info: Box::new(rewrite(info, boundary)?), + selector: selector.clone(), + }, + // An aggregate folds a group one point at a time: `AggregateRuntime::accumulate` folds every + // chunk into states indexed by `(group, grid position)`, so several contributions to one + // group at one position are that group's accumulation and not a series that has to be + // assembled first. That a `Complete` aggregate needs *one execution* is the placement's + // contract (`needs_single_execution_in`), which is a different answer from a merge. + // + // PART C: the merge that used to sit here was the only duplicate-label-set check on this + // path. `FloatAggregator` folds whatever it is handed, so two fragments of one label set now + // pass through unremarked and their values are summed. A check separated from the collection + // has to be added here before this removal is safe. + NativeExpr::Aggregate { input, op } => NativeExpr::Aggregate { + input: Box::new(rewrite(input, boundary)?), + op: op.clone(), + }, + // A vector operation is decided one operator at a time, because only one of them needs its + // input collected: `limitk` counts series in input order, so pieces of one label set would + // be counted against `k` separately. `topk`, `limit_ratio` and `sort` rank, keep or order + // series by what each holds at a point, `histogram_*` merges label-set pieces itself, and + // `count_values` and `absent` fold every contribution at a point. The module documentation + // is the table with the kernel each line was read from. + // + // `limitk`'s merge is the exception, and it keeps the check: `MergeInput::check` refuses a + // stream that breaks its contract and `merge_same_labelset` refuses two values of one field + // at one position. + // + // PART C: for the other six the merge that used to sit here was the only + // duplicate-label-set check on this path. `topk`, `limit_ratio` and `sort` hand the pieces + // back as separate series, `count_values` and `absent` fold them, and only `histogram_*` + // calls `merge_same_labelset` itself. + NativeExpr::Vector { input, op } => { + let input = rewrite(input, boundary)?; + NativeExpr::Vector { + input: Box::new(match op { + BoundVectorOp::LimitK { .. } + if matches!(unifies_its_own_contributions(&input), Some(false)) => + { + input + } + BoundVectorOp::LimitK { .. } => canonical(input, boundary), + BoundVectorOp::TopK { .. } + | BoundVectorOp::LimitRatio { .. } + | BoundVectorOp::Sort(_) + | BoundVectorOp::Absent { .. } + | BoundVectorOp::Histogram(_) + | BoundVectorOp::CountValues { .. } => input, + }), + op: op.clone(), + } + } + // A binary operation is point-wise: a scalar side is broadcast over the vector side's series + // one at a time, and two vectors are matched by the kernel, which files its rows by *result* + // label set and refuses two values of a field at one point. Neither needs its input collected + // into one series per label set, and matching-key completeness is the placement's question + // (`matching_co_located`), not a merge's. + NativeExpr::Binary { + left, + right, + matching, + } => NativeExpr::Binary { + left: Box::new(rewrite(left, boundary)?), + right: Box::new(rewrite(right, boundary)?), + matching: matching.clone(), + }, + // A set operation keeps each side's samples as they are and files them by result label set, + // refusing two values of a field at one point, so it reads contributions directly too. + NativeExpr::Set { left, right, op } => NativeExpr::Set { + left: Box::new(rewrite(left, boundary)?), + right: Box::new(rewrite(right, boundary)?), + op: op.clone(), + }, + }; + Ok(NativeNode { + expr, + output: node.output.clone(), + }) +} diff --git a/src/promql/src/native/canonicalize/tests.rs b/src/promql/src/native/canonicalize/tests.rs new file mode 100644 index 00000000000..60efbf00d75 --- /dev/null +++ b/src/promql/src/native/canonicalize/tests.rs @@ -0,0 +1,1704 @@ +// 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. + +//! What the insertion pass has to get right: it inserts a merge only where a consumer needs one, +//! the tree it produces is one its own validator accepts, and the facts it derives are the ones the +//! nodes below it actually preserve. +//! +//! The trees here are built node by node, over the one-field schemas of the `rate(x[5m])` fixture: +//! a raw scan, the window that folds it onto a grid, and - for `info` - the instant selection that +//! reports each info sample's own time. The compiler lives in the `query` crate, which depends on +//! this one, so a test in this crate cannot call it. +//! +//! Each test is named for the consumer it pins, because the whole point of the pass is that the +//! consumers are decided one at a time: a future change that re-generalizes them has to change a +//! test that says which kernel behaviour it would be overriding. + +use std::sync::Arc; + +use datafusion::execution::memory_pool::{MemoryPool, UnboundedMemoryPool}; +use datatypes::arrow::array::Float64Array; +use datatypes::arrow::buffer::{Buffer, ScalarBuffer}; +use datatypes::series::raw::{RawSeriesChunk, RawTimestamps}; +use datatypes::series::samples::{ + FieldSpec, SampleRef, Samples, Series, SeriesFields, SeriesSchema, +}; +use datatypes::series::time::{EvalGrid, GridSpan, GridSpec, TimeAxis}; +use datatypes::series::{SeriesLabels, SeriesOrdering}; + +use super::insert_native_merge_inside_current_fused; +use crate::evaluated::EvaluatedChunk; +use crate::histogram_eval::HistogramFunction; +use crate::info::InfoSelector; +use crate::map::MathOp; +use crate::native::execute::{NativeExecutionState, NativeExecutor}; +use crate::native::plan::{ + AggregateKind, AggregateStage, AggregateStateSchema, BoundAggregate, BoundInput, BoundInstant, + BoundMap, BoundScalar, BoundVectorOp, BoundWindow, CanonicalGrouping, CanonicalOwner, Grouping, + InstantValue, MapOp, NativeExpr, NativeNode, NativeProperties, NativeTree, NativeType, +}; +use crate::range_eval::{RangeFunction, RangeWindow}; +use crate::select::SortOrder; +use crate::vector_match::{BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels}; + +fn grid(start: i64, step: i64, len: u64) -> Arc { + Arc::new(EvalGrid::try_new(GridSpec::try_new(start, step, len).unwrap()).unwrap()) +} + +fn fields(count: usize) -> Arc<[FieldSpec]> { + (0..count) + .map(|index| FieldSpec { + name: format!("v{index}").into(), + }) + .collect() +} + +fn evaluated(schema_fields: Arc<[FieldSpec]>, grid: Arc) -> NativeType { + NativeType::Samples(Arc::new(SeriesSchema { + fields: schema_fields, + eval_grid: Some(grid), + })) +} + +/// A scan of raw samples, the leaf every tree here is built over. +fn raw_input(source_index: usize) -> NativeNode { + 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, + } +} + +/// A leaf that hands over samples already evaluated on `out`. +/// +/// This is the boundary a tree split at a repartition starts from: Fused B's `Input` in +/// `A -> Repartition -> B`, whose fragments are the pieces of one canonical label set that the +/// repartition fanned out. It is also the one leaf the runtime does *not* collect for a window +/// above it, which is what the tests built over it are about. +fn evaluated_input(source_index: usize, out: &Arc) -> NativeNode { + let output = evaluated(fields(1), Arc::clone(out)); + NativeNode { + expr: NativeExpr::Input(BoundInput { + output: output.clone(), + source_index, + ordering: SeriesOrdering::KeyThenTime, + }), + output, + } +} + +/// A constant, which a binary operation broadcasts. +fn scalar(value: f64, out: Arc) -> NativeNode { + NativeNode { + expr: NativeExpr::Scalar(BoundScalar { value }), + output: NativeType::Scalar(out), + } +} + +/// `rate(x[5m])` over `inner`, a window that folds it onto `out` and drops the metric name. +/// +/// `input_grid` is the grid `inner` is evaluated on, which is `None` for raw samples: the validator +/// compares it against what the input actually produces. +fn window_on( + inner: NativeNode, + input_grid: Option>, + out: Arc, +) -> NativeNode { + window_function_on(RangeFunction::Rate, inner, input_grid, out) +} + +/// `function(x[5m])` over `inner`, a window on `out` whose input is evaluated on `input_grid`. +fn window_function_on( + function: RangeFunction, + inner: NativeNode, + input_grid: Option>, + out: Arc, +) -> NativeNode { + window_sized_on(function, inner, input_grid, out, 300_000) +} + +/// [`window_function_on`] over a window of `range_ms`, which is what decides how many output points +/// one contribution decides. +/// +/// A range that spans several input steps makes every contribution decide every output point it +/// reaches, so two contributions of one label set collide at each of them and the window refuses the +/// second. A range that holds one input step makes each contribution decide its own output point, so +/// the same two contributions are two *series* instead - which is the gap these tests measure. +fn window_sized_on( + function: RangeFunction, + inner: NativeNode, + input_grid: Option>, + out: Arc, + range_ms: i64, +) -> NativeNode { + NativeNode { + expr: NativeExpr::Window { + input: Box::new(inner), + op: BoundWindow { + function, + window: RangeWindow { + range_ms, + offset_ms: 0, + anchor_ms: None, + modifier: None, + }, + input_grid, + output_grid: Arc::clone(&out), + }, + }, + output: evaluated(fields(1), out), + } +} + +/// [`window_on`] over raw samples, which is `rate(x[5m])`. +fn window_over(inner: NativeNode, out: Arc) -> NativeNode { + window_on(inner, None, out) +} + +/// `abs(inner)`: a point-wise map, which also drops the metric name. +fn map_over(inner: NativeNode) -> NativeNode { + let output = inner.output.clone(); + NativeNode { + expr: NativeExpr::Map { + input: Box::new(inner), + op: BoundMap { + ops: Arc::from([MapOp::Math(MathOp::Abs)]), + }, + }, + output, + } +} + +/// `sum without() (inner)` at `stage`, which folds every series it is handed into one group. +fn aggregate_over(inner: NativeNode, out: Arc, stage: AggregateStage) -> NativeNode { + let output = match stage { + AggregateStage::Partial => NativeType::AggregateState(Arc::new(AggregateStateSchema { + group: Grouping::By(Arc::from([])), + accumulators: fields(1), + grid: Arc::clone(&out), + })), + AggregateStage::Complete | AggregateStage::Final => evaluated(fields(1), out), + }; + NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(inner), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::from([])), + stage, + }, + }, + output, + } +} + +/// A vector operation over `inner`, laid out as `inner` is. +fn vector_over(inner: NativeNode, op: BoundVectorOp) -> NativeNode { + let output = inner.output.clone(); + NativeNode { + expr: NativeExpr::Vector { + input: Box::new(inner), + op, + }, + output, + } +} + +/// An `adding` match with no `on`/`ignoring`, which is `x + y`. +fn add() -> BinaryMatching { + BinaryMatching { + op: BinaryOp::Add, + return_bool: false, + labels: MatchingLabels::Ignoring(Vec::new()), + cardinality: Cardinality::OneToOne, + fill: FillValues::default(), + } +} + +/// A binary operation over `left` and `right` on `out`, which is one value per input field. +fn binary_over(left: NativeNode, right: NativeNode, out: Arc) -> NativeNode { + binary_matched(left, right, out, add()) +} + +/// A binary operation over `left` and `right` on `out` with `matching`, one value per input field. +fn binary_matched( + left: NativeNode, + right: NativeNode, + out: Arc, + matching: BinaryMatching, +) -> NativeNode { + let output = evaluated(fields(1), out); + NativeNode { + expr: NativeExpr::Binary { + left: Box::new(left), + right: Box::new(right), + matching, + }, + output, + } +} + +/// A comparison with no `bool`, which is a **filter**: `x > y` keeps the label set it read. +/// +/// This is the one binary modifier that does not drop the metric name (`broadcast_drops_labels`), +/// so a broadcast bound this way reaches the vector side's series without rewriting their labels. +fn filter() -> BinaryMatching { + BinaryMatching { + op: BinaryOp::Gt, + return_bool: false, + labels: MatchingLabels::Ignoring(Vec::new()), + cardinality: Cardinality::OneToOne, + fill: FillValues::default(), + } +} + +/// `timestamp(inner)`, which reports a position under the metric name dropped. +fn timestamp_over(inner: NativeNode) -> NativeNode { + let output = inner.output.clone(); + NativeNode { + expr: NativeExpr::Timestamp { + input: Box::new(inner), + }, + output, + } +} + +/// `Merge(inner)`: the canonical collection, placed by hand as the pass would place it. +fn merge_over(inner: NativeNode) -> NativeNode { + let output = inner.output.clone(); + NativeNode { + expr: NativeExpr::Merge { + input: Box::new(inner), + }, + output, + } +} + +/// An instant selection over `inner` that reports what `value` says it reports, on `out`. +fn selection(inner: NativeNode, out: Arc, value: InstantValue) -> NativeNode { + NativeNode { + expr: NativeExpr::Instant { + input: Box::new(inner), + op: BoundInstant { + lookback_ms: 300_000, + offset_ms: 0, + at_ms: None, + value, + output_grid: Arc::clone(&out), + }, + }, + output: evaluated(fields(1), out), + } +} + +/// `info(base, {})`: the base series with the data labels of the info series that share theirs. +fn info_over(base: NativeNode, info: NativeNode) -> NativeNode { + let output = base.output.clone(); + NativeNode { + expr: NativeExpr::Info { + base: Box::new(base), + info: Box::new(info), + selector: InfoSelector { + name_matchers: Vec::new(), + data_labels: Vec::new(), + keep_unmatched: true, + }, + }, + output, + } +} + +/// The node for the info side of [`info_over`]: the instant selection that reports sample times. +fn info_selection(out: &Arc) -> NativeNode { + selection(raw_input(1), Arc::clone(out), InstantValue::InfoTimestamp) +} + +/// `window_fn(x[5m])` over the scan, a window that folds raw samples onto `out`. +fn window_of(function: RangeFunction, out: Arc) -> NativeNode { + NativeNode { + expr: NativeExpr::Window { + input: Box::new(raw_input(0)), + op: BoundWindow { + function, + window: RangeWindow { + range_ms: 300_000, + offset_ms: 0, + anchor_ms: None, + modifier: None, + }, + input_grid: None, + output_grid: Arc::clone(&out), + }, + }, + output: evaluated(fields(1), out), + } +} + +/// Runs `root` over one scan and reads its answer as the rows of its one field, or the error the +/// execution reported. +/// +/// `fragments` are `(timestamp, value, is_last)` and are all pushed to the tree's one input under +/// one storage key, which is how one series arriving as several chunks looks: a chunk that does not +/// close the series leaves it open for the next one. +fn runtime_rows( + root: NativeNode, + fragments: &[(i64, f64, bool)], +) -> Result, String> { + let schema = SeriesSchema { + fields: fields(1), + eval_grid: None, + }; + let metric = SeriesLabels::try_new(Some("x".into()), Vec::new()).expect("a metric name"); + let tree = NativeTree::try_new(root).expect("the fixture is a legal tree"); + let pool: Arc = Arc::new(UnboundedMemoryPool::default()); + let state = + NativeExecutionState::try_new(&tree, Arc::clone(&pool)).expect("the runtime runs the tree"); + let ids = state.input_ids(0).collect::>(); + let mut executor = NativeExecutor::new(state); + for &(timestamp, value, is_last) in fragments { + let chunk = RawSeriesChunk::try_new( + &schema, + Buffer::from_slice_ref(b"series-a"), + Some(metric.clone()), + RawTimestamps::try_new(ScalarBuffer::from(vec![timestamp])).expect("one timestamp"), + SeriesFields::one(Samples::from_float(Float64Array::from(vec![value]))), + is_last, + ) + .expect("the chunk is bound to the schema it is built against"); + executor + .push_raw(ids[0], chunk) + .expect("the chunk is pushed to the tree's input"); + } + let chunks = executor.finish().map_err(|error| error.to_string())?; + let mut rows = Vec::new(); + for chunk in &chunks { + let field = chunk + .series + .fields + .as_slice() + .first() + .expect("the result carries the one field it was bound to"); + for (position, row) in chunk.series.time.positions() { + match field.sample(row) { + SampleRef::Float(value) => rows.push((position, value)), + _ => panic!("the row at {row} holds no float"), + } + } + } + Ok(rows) +} + +/// Runs `tree` over closing evaluated fragments of one canonical label set and reads its answer as +/// one entry per series the root handed over, holding that series' `(position, value)` rows. +/// +/// `fragments` are `(position on the input grid, value)`, and each is a chunk that closes the series +/// it carries: this is the shape a repartition's pieces of one label set take, several chunks with +/// one point each, each complete in itself. One entry back is one series, which is what makes the +/// count of them the answer to "did the window answer one series per label set, or one per piece". +fn evaluated_series_rows( + tree: &NativeTree, + input_grid: &Arc, + fragments: &[(usize, f64)], +) -> Result>, String> { + let schema = Arc::new(SeriesSchema { + fields: fields(1), + eval_grid: Some(Arc::clone(input_grid)), + }); + let metric = SeriesLabels::try_new(Some("x".into()), Vec::new()).expect("a metric name"); + let pool: Arc = Arc::new(UnboundedMemoryPool::default()); + let state = NativeExecutionState::try_new(tree, Arc::clone(&pool)) + .map_err(|error| error.to_string())?; + let ids = state.input_ids(0).collect::>(); + let mut executor = NativeExecutor::new(state); + for &(position, value) in fragments { + let chunk = EvaluatedChunk::try_new( + Series::try_new( + &schema, + metric.clone(), + TimeAxis::from_grid_spans(Arc::clone(input_grid), vec![GridSpan::new(position, 1)]) + .expect("a span on the grid"), + SeriesFields::one(Samples::from_float(Float64Array::from(vec![value]))), + ) + .expect("a series on the schema"), + true, + ) + .expect("a closing chunk"); + executor + .push_evaluated(ids[0], chunk) + .map_err(|error| error.to_string())?; + } + let chunks = executor.finish().map_err(|error| error.to_string())?; + let mut series = Vec::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 mut rows = Vec::new(); + for (position, row) in chunk.series.time.positions() { + match field.sample(row) { + SampleRef::Float(value) => rows.push((position, value)), + _ => panic!("the row at {row} holds no float"), + } + } + series.push(rows); + } + Ok(series) +} + +/// What a caller that placed the tree in one execution knows about its scans: this execution holds +/// one scan partition's pieces of the metric, and nothing says which series share a label set. +fn source_pieces() -> NativeProperties { + NativeProperties::source_pieces() +} + +/// The count of merges the pass put into `tree`, which is the number this change is about. +fn merges(tree: &NativeTree) -> usize { + tree.to_string().matches("Merge").count() +} + +/// Passes `root` with one boundary entry per source and returns the rewritten tree. +fn passed(root: NativeNode, sources: usize) -> NativeTree { + let tree = NativeTree::try_new(root).expect("the fixture is a legal tree"); + insert_native_merge_inside_current_fused(&tree, &vec![source_pieces(); sources]) + .expect("the pass keeps the tree legal") +} + +/// A point-wise map reads one series at a time, so it collects nothing and needs no merge. +/// +/// The old pass wrapped every map's input, because a scan's facts are +/// [`CanonicalGrouping::SourcePiecesOnly`] and that was reason enough. What that assertion encoded +/// is exactly the class rule this change removes: "the input is source pieces, so collect it", +/// asked without ever asking whether *this* consumer needed one series. A map does not: it applies +/// its operations to each chunk as it arrives (`map_chunk`), and a label set it makes two series +/// share is a fact about its output, answered at the consumer of that output. +#[test] +fn a_pointwise_map_does_not_collect_its_input() { + let out = grid(0, 60_000, 4); + let tree = passed(map_over(window_over(raw_input(0), Arc::clone(&out))), 1); + assert_eq!(0, merges(&tree), "a map collected its input:\n{tree}"); + let NativeExpr::Map { input, .. } = &tree.root().expr else { + panic!("the map is not the root:\n{tree}"); + }; + assert!( + matches!(input.expr, NativeExpr::Window { .. }), + "the map's input is not the window it was given:\n{tree}" + ); +} + +/// A broadcast reaches each of the vector side's series as it is, so neither side is collected. +#[test] +fn a_scalar_broadcast_does_not_collect_its_vector_side() { + let out = grid(0, 60_000, 4); + let tree = passed( + binary_over( + window_over(raw_input(0), Arc::clone(&out)), + scalar(2.0, Arc::clone(&out)), + Arc::clone(&out), + ), + 1, + ); + assert_eq!(0, merges(&tree), "`x * 2` collected a side:\n{tree}"); + let NativeExpr::Binary { left, right, .. } = &tree.root().expr else { + panic!("the binary operation is not the root:\n{tree}"); + }; + assert!( + matches!(right.expr, NativeExpr::Scalar(_)), + "the constant side changed:\n{tree}" + ); + assert!( + matches!(left.expr, NativeExpr::Window { .. }), + "the vector side was collected anyway:\n{tree}" + ); +} + +/// A `Complete` aggregate folds a group one point at a time, so it does not need its input +/// collected into one series per label set. +/// +/// The kernel, read in `execute.rs` and `aggregate.rs`: `AggregateRuntime::accumulate` keeps a +/// `FloatAggregator` whose states are indexed by `(group, absolute grid position)` and folds every +/// chunk it is handed into those states before dropping the chunk, and `FloatAggregator::finish` +/// then builds one series per group. Several contributions of one group at one position are +/// therefore the same accumulation as one contiguous series would give. What a `Complete` aggregate +/// does need is that every contribution to a group lies in this execution, which +/// `needs_single_execution_in` answers (`AggregateStage::Complete => true`) and the placement +/// enforces - a placement condition, not a merge. +/// +/// The merge the old pass put here was not re-assembling a series for the fold; it was refusing a +/// vector holding two values of one field at one position under one label set +/// (`merge_same_labelset`'s "vector cannot contain metrics with the same labelset"). That check is +/// gone from this path with the merge; the module documentation's duplicate-check section says which +/// paths keep it and which are Part C's to restore. +#[test] +fn a_complete_aggregate_does_not_collect_its_input() { + let out = grid(0, 60_000, 4); + let tree = passed( + aggregate_over( + window_over(raw_input(0), Arc::clone(&out)), + Arc::clone(&out), + AggregateStage::Complete, + ), + 1, + ); + assert_eq!( + 0, + merges(&tree), + "a complete aggregate collected its input:\n{tree}" + ); + let NativeExpr::Aggregate { input, .. } = &tree.root().expr else { + panic!("the aggregate is not the root:\n{tree}"); + }; + assert!( + matches!(input.expr, NativeExpr::Window { .. }), + "the aggregate's input was collected anyway:\n{tree}" + ); +} + +/// A window over fragments is served by the runtime, not by a merge in the tree - for the children +/// whose fragments the runtime is already the collection for. +/// +/// `RuntimeNode::build`'s `Window` arm asks +/// [`unifies_its_own_contributions`](crate::native::plan::unifies_its_own_contributions) about its +/// child and gives the window a `MergeInput` when the answer is `Some(false)` - the child reports +/// its own fragments: they are collected, checked against the stream contract (`MergeInput::check`) +/// and merged by `merge_same_labelset` before the evaluator sees them. A merge node in the tree +/// would turn that collection *off* (a merge answers `Some(true)`), so the window would stream an +/// already-merged series while the check moved to a node without the evaluator's context. The two +/// children below are exactly those: a `Map`, which always reports fragments, and a scan leaf, which +/// hands over one whole storage series at a time and which no merge could collect in any case (a +/// merge binds samples evaluated on a grid, and a scan's are raw). +/// +/// The third child, an evaluated `Input`, is the one this pairing does not cover: `Input` reports +/// nothing (`None`) and nothing below it unified anything, so the pass inserts the merge there and +/// [`a_window_over_a_plain_evaluated_input_collects_aliasing_fragments`] is that test. +#[test] +fn a_window_over_fragments_is_served_by_the_runtime_not_a_merge() { + let out = grid(0, 60_000, 4); + // A window over raw samples: the leaf hands over one whole series at a time, so it streams. + let over_scan = passed(window_over(raw_input(0), Arc::clone(&out)), 1); + assert_eq!( + 0, + merges(&over_scan), + "a window over a scan gained a merge:\n{over_scan}" + ); + // A window over an evaluated child: this is the shape whose runtime arm builds a `MergeInput` + // (`child_shared`), so the collection is the runtime's and not the tree's. + let over_evaluated = passed( + window_on( + map_over(window_over(raw_input(0), Arc::clone(&out))), + Some(Arc::clone(&out)), + Arc::clone(&out), + ), + 1, + ); + assert_eq!( + 0, + merges(&over_evaluated), + "a window over an evaluated child gained a merge as well as the runtime's own:\n{over_evaluated}" + ); +} + +/// A vector/vector match reads both sides whole and files its rows by result label set, so neither +/// side is collected. +/// +/// The kernel, read in `vector_match.rs`: `binary_vectors` groups each operand by the match's labels +/// (`group`), pairs them point by point, and files every pairing's sample in `Results` under the +/// label set the sample ends under; `merge_fields` then hands over one series per set, and +/// `Results::sort_rows` refuses two values of a field at one point under one set. Which series may +/// pair is the placement's question (`matching_co_located`, via `needs_single_execution_in`), not a +/// merge's. +/// +/// The old pass collected both sides of every match, because both sides were "source pieces". What +/// that encoded is that a match cannot match pieces; it can - and the same-label-set refusal a merge +/// would have produced is the kernel's own. +#[test] +fn a_vector_match_does_not_collect_either_side() { + let out = grid(0, 60_000, 4); + let tree = passed( + binary_over( + window_over(raw_input(0), Arc::clone(&out)), + window_over(raw_input(1), Arc::clone(&out)), + Arc::clone(&out), + ), + 2, + ); + assert_eq!(0, merges(&tree), "`x + y` collected a side:\n{tree}"); + let NativeExpr::Binary { left, right, .. } = &tree.root().expr else { + panic!("the match is not the root:\n{tree}"); + }; + for side in [left, right] { + assert!( + matches!(side.expr, NativeExpr::Window { .. }), + "a side of the match was collected:\n{tree}" + ); + } +} + +/// The root is not canonicalized: Prometheus response assembly groups batches by label set and +/// SQL/TQL consumes rows, so neither requires a native whole-series chunk. +#[test] +fn the_root_is_not_canonicalized_by_default() { + let out = grid(0, 60_000, 4); + // A map at the root is the shape `next_chunk` names: it hands over pieces. + let mapped = passed(map_over(window_over(raw_input(0), Arc::clone(&out))), 1); + assert!( + !matches!(mapped.root().expr, NativeExpr::Merge { .. }), + "root was canonicalized:\n{mapped}" + ); + // A bare scan is the other: one whole series at a time, never a collected one. + let scanned = passed(raw_input(0), 1); + assert!( + !matches!(scanned.root().expr, NativeExpr::Merge { .. }), + "a scan root was canonicalized:\n{scanned}" + ); +} + +/// The seven vector operations are decided one at a time, because only one of them needs its input +/// collected into one series per canonical label set. +/// +/// The table below is the finding, one line per kernel: +/// +/// - `topk`: ranks the series holding a sample at each point and keeps the selected points of the +/// series it picked. Pieces of one label set hold their samples at different positions, so at any +/// point at most one of them is a candidate, and `k` counts series. +/// - `limitk`: **the one that needs a merge.** `k` counts *series* in input order (`select::limitk` +/// walks a group's members in that order and stops after `k`), so pieces of one label set would +/// occupy several slots and be counted against `k`. The runtime merges its input only for a child +/// that shares runs (`vector.merge_input = limitk && child_shared`), and a scan leaf reports +/// `shared = false`, so the tree has to carry the merge. +/// - `limit_ratio`: keeps whole series by the hash of their labels, so all pieces of one label set +/// are kept or dropped together and each is rebuilt from the positions it holds. +/// - `sort`: reorders the series it is given; the series keep their labels and their points. +/// - `histogram_quantile`/`histogram_fraction`: the kernel merges label-set pieces itself +/// (`vector_series` calls `merge_same_labelset` before `evaluate_histograms`). +/// - `count_values`: counts samples per group and value at each point, which folds contributions the +/// way an aggregate folds them. +/// - `absent`: reads presence per point over the whole vector, so the union of the contributions is +/// the presence it needs. +#[test] +fn each_vector_operator_is_decided_on_its_own() { + let out = grid(0, 60_000, 4); + let by_job = || Grouping::By(Arc::from([Arc::from("job")])); + for (op, expected) in [ + ( + BoundVectorOp::TopK { + k: 1.0, + top: true, + group: by_job(), + }, + 0, + ), + ( + BoundVectorOp::LimitK { + k: 1.0, + group: by_job(), + }, + 0, + ), + (BoundVectorOp::LimitRatio { ratio: 0.5 }, 0), + (BoundVectorOp::Sort(SortOrder::Asc), 0), + ( + BoundVectorOp::Absent { + labels: SeriesLabels::try_new(None, Vec::new()).expect("no labels"), + }, + 0, + ), + ( + BoundVectorOp::Histogram(HistogramFunction::Quantile(0.9)), + 0, + ), + ( + BoundVectorOp::CountValues { + label: Arc::from("value"), + group: by_job(), + }, + 0, + ), + ] { + let tree = passed( + vector_over(window_over(raw_input(0), Arc::clone(&out)), op.clone()), + 1, + ); + assert_eq!( + expected, + merges(&tree), + "`{op:?}` should have gained {expected} merge(s):\n{tree}" + ); + } +} + +/// The three-condition rule, positively: a consumer that needs one unified series, over an input that +/// does not already provide one, still gets a merge. +/// +/// `limitk` is that consumer (see [`each_vector_operator_is_decided_on_its_own`]); the input is a +/// window over a scan, which proves no grouping. Co-location - condition 3 - is not this pass's to +/// answer: the route only runs it for a placement that reads every input partition, and +/// `merges_are_placeable_in` is where a tree carrying a merge is checked against that execution. +/// +/// The second half is the other side of condition 2: an input that already answers one series per +/// label set - here the merge the first half inserted - gains nothing. +#[test] +fn the_only_consumer_that_needs_a_merge_still_gets_one() { + let out = grid(0, 60_000, 4); + let limitk = || BoundVectorOp::LimitK { + k: 1.0, + group: Grouping::By(Arc::from([])), + }; + let tree = passed( + vector_over(window_over(raw_input(0), Arc::clone(&out)), limitk()), + 0, + ); + assert_eq!( + 0, + merges(&tree), + "runtime LimitK already collects its input:\n{tree}" + ); + let NativeExpr::Vector { input, .. } = &tree.root().expr else { + panic!("the vector operation is not the root:\n{tree}"); + }; + assert!( + matches!(input.expr, NativeExpr::Window { .. }), + "the runtime's collection trigger was wrapped in an explicit merge:\n{tree}" + ); + + // Condition 2 again, one level down: an inner `limitk` already collects its input at runtime, + // so the outer one gains no explicit merge. + let nested = passed( + vector_over( + vector_over(window_over(raw_input(0), Arc::clone(&out)), limitk()), + limitk(), + ), + 1, + ); + assert_eq!( + 0, + merges(&nested), + "limitk inputs collected by their runtime need no explicit merge:\n{nested}" + ); +} + +/// A legal `info` tree survives the pass, and the tree the pass produces is the one its validator +/// accepts. +/// +/// This is the contract the pass once broke: it wrapped every whole-series consumer in a merge, +/// `info`'s info side included, and the validator used to require that side be *directly* an instant +/// selection reporting sample times - so the pass built a tree `NativeTree::try_new` then rejected. +/// `info` needs no merge on either side (the kernel indexes the info side by `(name, identifying +/// values)` and merges the results that share a label set), and the validator still looks through a +/// merge for a tree that carries one. +#[test] +fn a_legal_info_tree_survives_canonicalize_and_validation() { + let out = grid(0, 60_000, 4); + let tree = NativeTree::try_new(info_over( + window_over(raw_input(0), Arc::clone(&out)), + info_selection(&out), + )) + .expect("the compiled tree is legal"); + let boundary = [NativeProperties::source_pieces()]; + let passed = insert_native_merge_inside_current_fused(&tree, &boundary) + .expect("the pass keeps the tree legal"); + + assert_eq!(0, merges(&passed), "`info` collected a side:\n{passed}"); + + // The pass returns a tree it validated, and the same tree validates again from scratch. + NativeTree::try_new(passed.root().clone()).expect("the passed tree is legal on its own"); + + // The pass is idempotent on its own output: the one consumer it collects already carries a + // merge, so a second run adds nothing. + let again = insert_native_merge_inside_current_fused(&passed, &boundary) + .expect("the pass accepts its own output"); + assert_eq!( + passed.to_string(), + again.to_string(), + "the pass changed its own output" + ); +} + +/// The pass is idempotent over the output it actually produces, which is the tree with a merge in +/// it: a second run must not collect an input a merge already made complete. +#[test] +fn the_pass_is_idempotent() { + let out = grid(0, 60_000, 4); + let boundary = [NativeProperties::source_pieces()]; + let limitk = || BoundVectorOp::LimitK { + k: 1.0, + group: Grouping::By(Arc::from([])), + }; + for root in [ + // The consumers that gain a merge: `limitk`, and a window sitting on a fused input. + vector_over(window_over(raw_input(0), Arc::clone(&out)), limitk()), + window_on( + evaluated_input(0, &out), + Some(Arc::clone(&out)), + Arc::clone(&out), + ), + // And shapes that gain none, so "changed nothing" is asserted on both answers. + map_over(window_over(raw_input(0), Arc::clone(&out))), + window_over(raw_input(0), Arc::clone(&out)), + ] { + let once = NativeTree::try_new(root).expect("the fixture is a legal tree"); + let once = insert_native_merge_inside_current_fused(&once, &boundary) + .expect("the pass keeps the tree legal"); + let twice = insert_native_merge_inside_current_fused(&once, &boundary) + .expect("the pass accepts its own output"); + assert_eq!( + once.to_string(), + twice.to_string(), + "the pass changed its own output:\n{once}" + ); + } +} + +/// An instant selection that reports an info series' own time keeps every fact its input had: the +/// runtime files that series under all of its labels, so which series share a label set is +/// unchanged. `timestamp(x)` is the one that does not - it drops the metric name on the way - and +/// the two are asserted together so the distinction cannot be lost. +#[test] +fn info_timestamp_keeps_its_input_facts() { + let out = grid(0, 60_000, 4); + // A boundary that proves something, so a downgrade is visible rather than a coincidence of two + // `Unknown`s. + let boundary = [NativeProperties { + owner: CanonicalOwner::ByCanonicalKey, + grouping: CanonicalGrouping::SourcePiecesOnly, + }]; + let facts = |value| selection(raw_input(0), Arc::clone(&out), value).properties(&boundary); + + assert_eq!( + boundary[0], + facts(InstantValue::InfoTimestamp), + "an info timestamp rewrote the label sets it read" + ); + // The sample itself is answered under the label set it was read from, so it is the same answer. + assert_eq!(boundary[0], facts(InstantValue::Sample)); + assert_eq!( + NativeProperties { + owner: CanonicalOwner::Unknown, + grouping: CanonicalGrouping::SourcePiecesOnly, + }, + facts(InstantValue::Timestamp), + "a timestamp kept facts it rewrote" + ); +} + +/// A window folds the contributions of one canonical label set as **one** logical series, and the +/// `avg_over_time` / `sum_of_avg` pair below is what makes that visible: `sum_over_time` cannot tell +/// the two behaviours apart (`2 + 4 = 6` either way), while an average of the union (`(2 + 4) / 2 = +/// 3`) differs from a sum of separate averages (`2 + 4 = 6`). +/// +/// What performs the joint processing, and therefore why the pass inserts no merge here: the window's +/// own evaluator carries one series' state across its chunks and closes it on the chunk that says +/// `is_last` (`NativeChunk::Raw` -> `RangeEvaluator::push`, whose `SeriesState` keeps the window's +/// samples and its last timestamp for the next chunk). The chunks below are one *storage* series +/// handed over in two pieces - the shape a scan partition's pieces take - and one storage series is +/// one canonical label set, so nothing can alias at the leaf and the evaluator is the unification. A +/// merge could not stand in for it either: one binds samples evaluated on a grid, and this leaf's +/// samples are raw, which is why the pass never collects a scan even though it is an `Input`. +/// +/// The evaluated-input side is the one that splits in two, and both halves are pinned. Where the +/// contributions come from a node that reports its own fragments, the window gets its `MergeInput` +/// from the runtime: [`a_window_over_aliasing_child_collects_its_fragments_at_runtime`]. Where they +/// arrive at a fused input - the split shape `A -> Repartition -> B` - the leaf reports nothing, so +/// the runtime collects nothing either and the tree has to carry the merge: +/// [`a_window_over_a_plain_evaluated_input_collects_aliasing_fragments`] is the tree it carries it +/// in, and [`the_merge_in_front_of_a_window_is_what_collects_the_fragments`] is the `avg_over_time` +/// measurement of the pair above taken on that collected shape, plus the same fixture run without +/// the merge, where it does not answer a wrong number but is refused. +#[test] +fn a_window_folds_the_contributions_of_one_label_set_as_one_series() { + let rows = |function| { + // `sum(avg_over_time(x[5m]))`, so the answer is one number. + runtime_rows( + aggregate_over( + window_of(function, grid(60_000, 60_000, 1)), + grid(60_000, 60_000, 1), + AggregateStage::Complete, + ), + // One series at 0s holding 2, continuing at 20s holding 4; the 300s window over the one + // grid point at 60s covers both. + &[(0, 2.0, false), (20_000, 4.0, true)], + ) + }; + assert_eq!( + Ok(vec![(0, 3.0)]), + rows(RangeFunction::AvgOverTime), + "the window did not average the union of the two contributions: it folded them apart" + ); + // The control: `sum_over_time` answers 6 either way, so it proves nothing on its own - it is + // here to show that the assertion above is the discriminating one. + assert_eq!( + Ok(vec![(0, 6.0)]), + rows(RangeFunction::SumOverTime), + "`sum_over_time` no longer answers the sum of the contributions" + ); +} + +/// The same shape with two contributions holding a value at the **same** position is an error, not a +/// sum: `sum` would hide the conflict by adding the two values. +/// +/// On a streaming window the check is the evaluator's: `RangeEvaluator::push` refuses timestamps that +/// do not increase across the chunks of one series ("raw series timestamps are not increasing across +/// chunks"). It is therefore still made, by the node that folds the contributions. +#[test] +fn a_window_refuses_two_contributions_holding_one_position() { + let error = runtime_rows( + aggregate_over( + window_of(RangeFunction::SumOverTime, grid(60_000, 60_000, 1)), + grid(60_000, 60_000, 1), + AggregateStage::Complete, + ), + &[(0, 2.0, false), (0, 4.0, true)], + ) + .expect_err("two values of one position under one label set were summed"); + assert!( + error.contains("not increasing"), + "the refusal does not name the stream contract it broke: {error}" + ); +} + +/// A window over a node that *can* alias gets its collection from the runtime, and that collection is +/// a `MergeInput`: `RuntimeNode::build`'s `Window` arm asks its child for `shared`, and the nodes that +/// rewrite labels report `true` - a `Map` always, a `Timestamp` always, a `Window` always, a +/// label-dropping broadcast always. The window then collects the fragments, checks the stream contract +/// (`MergeInput::check`) and merges them with `merge_same_labelset` before the evaluator sees them. +/// +/// That merge *is* the joint processing: cross-contribution ordering (it sorts by position), one state +/// per label set (each merged set is handed to the evaluator as one closing series) and the +/// end-of-series decision (read from the `is_last` of the collected fragments, not of each piece). +/// +/// The fixture is two fragments under one label set holding values at *different* positions, which is +/// what a partition's pieces of a series look like. A window that opened one state per fragment would +/// report two series under one label set; one `MergeInput` reports one. The pass' own output is +/// asserted first, because the merge that does it is the runtime's. +/// +/// That is what makes this the pair of +/// [`a_window_over_a_plain_evaluated_input_collects_aliasing_fragments`], where the same fragments +/// arrive at a fused input instead: the pass leaves this tree alone because its window's child is a +/// `Map` - a node whose fragments the runtime reports and collects, and whose collection a +/// tree-level merge would turn off - and it collects that one because its window's child is the +/// boundary itself, which proves nothing at all. The pass decides between the two with +/// [`unifies_its_own_contributions`](crate::native::plan::unifies_its_own_contributions), the same +/// function `RuntimeNode::build` computes its `shared` flag from; the module documentation of the +/// window arm is the full statement. +#[test] +fn a_window_over_aliasing_child_collects_its_fragments_at_runtime() { + // Two grid points, so the two fragments have a position each. + let out = grid(60_000, 60_000, 2); + // The pass leaves the tree alone: the collection below is the runtime's. + let passed = passed( + window_on( + map_over(window_over(raw_input(0), Arc::clone(&out))), + Some(Arc::clone(&out)), + Arc::clone(&out), + ), + 1, + ); + assert_eq!( + 0, + merges(&passed), + "the pass inserted a merge the runtime already performs:\n{passed}" + ); + + // The same shape executed over two complementary fragments of one label set. + let schema = Arc::new(SeriesSchema { + fields: fields(1), + eval_grid: Some(Arc::clone(&out)), + }); + let metric = SeriesLabels::try_new(Some("x".into()), Vec::new()).expect("a metric name"); + let fragment = |position: usize, value: f64| { + EvaluatedChunk::try_new( + Series::try_new( + &schema, + metric.clone(), + TimeAxis::from_grid_spans(Arc::clone(&out), vec![GridSpan::new(position, 1)]) + .expect("a span on the grid"), + SeriesFields::one(Samples::from_float(Float64Array::from(vec![value]))), + ) + .expect("a series on the schema"), + true, + ) + .expect("a closing chunk") + }; + // The input's node output and its binding are one schema, which is what `validate` compares. + let input = NativeNode { + expr: NativeExpr::Input(BoundInput { + output: NativeType::Samples(Arc::clone(&schema)), + source_index: 0, + ordering: SeriesOrdering::KeyThenTime, + }), + output: NativeType::Samples(Arc::clone(&schema)), + }; + let tree = NativeTree::try_new(window_on( + map_over(input), + Some(Arc::clone(&out)), + Arc::clone(&out), + )) + .expect("a legal tree"); + let pool: Arc = Arc::new(UnboundedMemoryPool::default()); + let state = + NativeExecutionState::try_new(&tree, Arc::clone(&pool)).expect("the runtime runs the tree"); + let ids = state.input_ids(0).collect::>(); + let mut executor = NativeExecutor::new(state); + for (position, value) in [(0, 1.0), (1, 2.0)] { + executor + .push_evaluated(ids[0], fragment(position, value)) + .expect("the fragment is pushed"); + } + let chunks = executor.finish().expect("the window folds both fragments"); + assert_eq!( + 1, + chunks.len(), + "the window opened one state per fragment instead of collecting them:\n{tree}" + ); + // One series, holding the positions the window decided for the unified series. Its values are + // the function's (`rate` here), so the positions are what this asserts: a window that opened one + // state per fragment answers one series per fragment, not one series with both fragments' rows. + let positions: Vec = chunks[0] + .series + .time + .positions() + .map(|(position, _)| position) + .collect(); + assert_eq!( + vec![1], + positions, + "the collected series does not hold the position the two fragments decide together" + ); +} + +/// The duplicate check an aliasing node owes is its own, and it fires before a window above it +/// collects anything: `Map` keeps an `Emitted` occupancy table per output label set and refuses a +/// second series writing a position a first one wrote ("no merge resolves two values of one field at +/// one position"). +/// +/// This is the check a merge in the tree would otherwise have made, and on this path it was never the +/// merge's: the node that makes two series one label set is the node that refuses the conflict, which +/// is why removing the merge above it takes no check with it. +#[test] +fn an_aliasing_node_refuses_two_contributions_holding_one_position() { + let out = grid(60_000, 60_000, 1); + let schema = Arc::new(SeriesSchema { + fields: fields(1), + eval_grid: Some(Arc::clone(&out)), + }); + let metric = SeriesLabels::try_new(Some("x".into()), Vec::new()).expect("a metric name"); + let fragment = |value: f64| { + EvaluatedChunk::try_new( + Series::try_new( + &schema, + metric.clone(), + TimeAxis::from_grid_spans(Arc::clone(&out), vec![GridSpan::new(0, 1)]) + .expect("a span on the grid"), + SeriesFields::one(Samples::from_float(Float64Array::from(vec![value]))), + ) + .expect("a series on the schema"), + true, + ) + .expect("a closing chunk") + }; + let input = NativeNode { + expr: NativeExpr::Input(BoundInput { + output: NativeType::Samples(Arc::clone(&schema)), + source_index: 0, + ordering: SeriesOrdering::KeyThenTime, + }), + output: NativeType::Samples(Arc::clone(&schema)), + }; + // A `Map` is the label-rewriting node, and it keeps the table whether or not it rewrites here. + let tree = NativeTree::try_new(map_over(input)).expect("a legal tree"); + let pool: Arc = Arc::new(UnboundedMemoryPool::default()); + let state = + NativeExecutionState::try_new(&tree, Arc::clone(&pool)).expect("the runtime runs the tree"); + let ids = state.input_ids(0).collect::>(); + let mut executor = NativeExecutor::new(state); + executor + .push_evaluated(ids[0], fragment(2.0)) + .expect("the first fragment is pushed"); + executor + .push_evaluated(ids[0], fragment(4.0)) + .expect("the second fragment is pushed"); + let error = executor + .finish() + .expect_err("two values of one position under one label set were folded"); + assert!( + error.to_string().contains("no merge resolves"), + "the aliasing node did not refuse the two values: {error}" + ); +} + +/// A window over a **plain** evaluated input collects the aliasing fragments of one label set into +/// one series. This is the split shape: `A -> Repartition -> B` with `B`'s tree `Window(Input)`. +/// +/// The two fragments below are the pieces of one canonical label set that crossed the repartition - +/// one closing chunk each, holding one point each, at different positions. The pass hands the tree +/// over with a merge between the window and the boundary, because the runtime's own collection is +/// not reachable here: `RuntimeNode::build`'s `Window` arm builds its `MergeInput` only when the +/// child reports its fragments, and `NativeExpr::Input` reports nothing at all - a leaf hands over +/// what its boundary bound it to and claims nothing about the pieces being one series, so +/// [`unifies_its_own_contributions`](crate::native::plan::unifies_its_own_contributions) answers +/// `None` for it and the runtime's flag follows with `false`. Nothing below the window in a split +/// tree unifies them either, so without the merge the window opens one state per fragment and +/// answers two series where one is owed. That is what this test measures, and both halves are +/// asserted: the merge is in the tree, and the answer is one series. +/// +/// [`a_window_over_aliasing_child_collects_its_fragments_at_runtime`] is the other side of the same +/// measurement: a child that *does* report `shared` - a `Map` - gets its `MergeInput` from the +/// runtime, and the pass inserts nothing there. +#[test] +fn a_window_over_a_plain_evaluated_input_collects_aliasing_fragments() { + let out = grid(60_000, 60_000, 2); + let tree = passed( + window_on( + evaluated_input(0, &out), + Some(Arc::clone(&out)), + Arc::clone(&out), + ), + 1, + ); + // The merge is where the collection has to be, since the runtime does not perform it: between + // the window and the input whose fragments are the pieces of one label set. + assert_eq!( + 1, + merges(&tree), + "the pass left the window's fused input uncollected:\n{tree}" + ); + let NativeExpr::Window { input, .. } = &tree.root().expr else { + panic!("the window is not the root:\n{tree}"); + }; + let NativeExpr::Merge { input: collected } = &input.expr else { + panic!("the merge does not sit directly over the window's input:\n{tree}"); + }; + assert!( + matches!(&collected.expr, NativeExpr::Input(_)), + "the merge does not sit over the boundary:\n{tree}" + ); + + // Two fragments of one label set, at different positions, each closing the series it carries. + let series = evaluated_series_rows(&tree, &out, &[(0, 1.0), (1, 2.0)]) + .expect("the runtime runs the tree"); + assert_eq!( + 1, + series.len(), + "the window opened one state per fragment instead of collecting them: {series:?}\n{tree}" + ); +} + +/// The merge the pass inserts is the joint processing, and this fixture is what measures it: +/// `avg_over_time` over two complementary contributions of one label set is their average over the +/// union (`(2 + 4) / 2 = 3`), and a `sum` around it keeps that 3. A window that opened one state per +/// contribution and averaged each on its own would not answer 3 from these two values, so the +/// assertion on the collected shape is one only a collected shape satisfies. +/// +/// `sum_over_time` is kept as the control and is labelled one because it does *not* discriminate: it +/// answers `2 + 4 = 6` whether the contributions arrived collected or apart, so it shows that the +/// average above is the assertion that separates the two behaviours, and proves nothing by itself. +/// +/// The un-collected shape is pinned as well, and it is not merely wrong - it is *refused*. Both +/// contributions decide the same output position, and the window's `Emitted` occupancy table reports +/// two series holding a value of one field at one position under one label set: "no merge resolves +/// two values of one field at one position". That is the check the module documentation's +/// duplicate-check section says a label-rewriting node keeps, and it is why this fixture separates +/// the two shapes by an error as well as by an answer: the merge is what turns two contributions +/// into the one series whose single pass is what the query owes. +#[test] +fn the_merge_in_front_of_a_window_is_what_collects_the_fragments() { + // The window reads a 300s range over an input on a 20s grid, at the one point of a 60s grid, so + // its range covers both contributions. + let input_grid = grid(0, 20_000, 2); + let out = grid(60_000, 60_000, 1); + // The two contributions, as `(position on the input grid, value)`: 2 at 0s and 4 at 20s. + let fragments = [(0, 2.0), (1, 4.0)]; + let window = |function| { + window_function_on( + function, + evaluated_input(0, &input_grid), + Some(Arc::clone(&input_grid)), + Arc::clone(&out), + ) + }; + // `sum(window_fn(x[5m]))`: `sum` folds whatever the window answered, so the answer is one number + // whether the window collected its contributions or not. + let summed = |function| -> Result { + let root = aggregate_over(window(function), Arc::clone(&out), AggregateStage::Complete); + let tree = passed(root, 1); + assert_eq!( + 1, + merges(&tree), + "the pass did not collect the window's fused input:\n{tree}" + ); + let series = evaluated_series_rows(&tree, &input_grid, &fragments)?; + Ok(series.iter().flatten().map(|&(_, value)| value).sum()) + }; + + // The bare window: `avg_over_time` over the two contributions as one series is their average, and + // the collected series holds the one output position with that value. + let collected = passed(window(RangeFunction::AvgOverTime), 1); + assert_eq!( + 1, + merges(&collected), + "the pass did not collect the window's fused input:\n{collected}" + ); + assert_eq!( + Ok(vec![vec![(0, 3.0)]]), + evaluated_series_rows(&collected, &input_grid, &fragments), + "`avg_over_time` did not average the collected contributions into one series" + ); + // An aggregate above it keeps that 3: one series in, one value out. + assert_eq!( + Ok(3.0), + summed(RangeFunction::AvgOverTime), + "`sum(avg_over_time(x[5m]))` did not keep the collected average" + ); + // The control, which answers 6 either way and therefore discriminates nothing: `sum_over_time` + // folds the two contributions into their sum whether they arrived collected or apart. + assert_eq!( + Ok(6.0), + summed(RangeFunction::SumOverTime), + "`sum_over_time` no longer answers the sum of the contributions" + ); + + // Without the pass the same fixture is not answered at all: two states decide the one output + // position, and the window's occupancy table refuses the second. This is the shape the merge is + // there to prevent, so it is asserted rather than described. + let uncollected = NativeTree::try_new(window(RangeFunction::AvgOverTime)) + .expect("the fixture is a legal tree"); + assert_eq!( + 0, + merges(&uncollected), + "the fixture carries a merge of its own:\n{uncollected}" + ); + let error = evaluated_series_rows(&uncollected, &input_grid, &fragments) + .expect_err("two states decided one output position under one label set"); + assert!( + error.contains("no merge resolves"), + "the uncollected window did not report the duplicate it decided: {error}" + ); +} + +/// A `Vector` that hands the series it ranked back unchanged is collected **above** it, because the +/// window over it reads the fragments of one label set the boundary handed over. +/// +/// `topk`, `limit_ratio` and `sort` share one arm of both readers - `unifies_its_own_contributions` +/// answers for all three by passing its input's answer on, and `RuntimeNode::build`'s report follows +/// that answer - so they are asserted together. Each hands back the series it read with their +/// labels, so several contributions of one label set stay several series. Over a boundary all three +/// answer "cannot say", which is what makes [`rewrite`]'s `Window` arm insert the merge. +/// +/// The measurement is the pair the gap is stated as. The window below is one input step wide and the +/// input grid and the output grid are the same two points, so each fragment decides exactly one +/// output position: +/// +/// - **without** the pass: `merges = 0` and the window opens one state per fragment, so it answers +/// **two** series where one is owed; +/// - **with** the pass: `merges = 1` and the fragments are collected first, so the window answers +/// **one** series holding both positions. +/// +/// The merge goes above the operation and not below it: `topk` ranks the series it is handed, so +/// collecting below it would change which series it keeps. What this test pins is the window's +/// series identity; a collection *below* a `topk` is a different question and is not what this pass +/// does. +#[test] +fn a_vector_that_hands_its_series_back_is_collected_above() { + // Two points 20s apart, and a window one step wide, so a fragment at input position `i` decides + // output position `i` and nothing else. + let grid_two = grid(0, 20_000, 2); + let range_ms = 20_000; + let by_job = || Grouping::By(Arc::from([Arc::from("job")])); + for op in [ + BoundVectorOp::TopK { + k: 1.0, + top: true, + group: by_job(), + }, + BoundVectorOp::LimitRatio { ratio: 1.0 }, + BoundVectorOp::Sort(SortOrder::Asc), + ] { + let shape = || { + window_sized_on( + RangeFunction::AvgOverTime, + vector_over(evaluated_input(0, &grid_two), op.clone()), + Some(Arc::clone(&grid_two)), + Arc::clone(&grid_two), + range_ms, + ) + }; + // Without the pass: two fragments, two states, two series under one label set. + let uncollected = NativeTree::try_new(shape()).expect("the fixture is a legal tree"); + assert_eq!( + 0, + merges(&uncollected), + "the fixture carries a merge of its own:\n{uncollected}" + ); + let apart = evaluated_series_rows(&uncollected, &grid_two, &[(0, 1.0), (1, 2.0)]) + .unwrap_or_else(|error| panic!("the runtime runs `{op:?}`: {error}\n{uncollected}")); + assert_eq!( + 2, + apart.len(), + "`{op:?}` answered {} series where the un-collected window owes one per fragment: \ + {apart:?}\n{uncollected}", + apart.len() + ); + + // With the pass: one merge above the operation, and one series holding both positions. + let tree = passed(shape(), 1); + assert_eq!( + 1, + merges(&tree), + "`{op:?}` over a boundary was not collected:\n{tree}" + ); + let NativeExpr::Window { input, .. } = &tree.root().expr else { + panic!("the window is not the root:\n{tree}"); + }; + let NativeExpr::Merge { input: collected } = &input.expr else { + panic!("the merge does not sit over the window's input:\n{tree}"); + }; + assert!( + matches!(&collected.expr, NativeExpr::Vector { .. }), + "the merge does not sit over the vector operation:\n{tree}" + ); + let series = evaluated_series_rows(&tree, &grid_two, &[(0, 1.0), (1, 2.0)]) + .unwrap_or_else(|error| panic!("the runtime runs `{op:?}`: {error}\n{tree}")); + assert_eq!( + vec![vec![(0, 1.0), (1, 2.0)]], + series, + "`{op:?}` did not hand the window one folded series: {series:?}\n{tree}" + ); + } +} + +/// A scalar `Broadcast` whose comparison keeps the label set it read is collected **above** it, for +/// the same reason a `Vector` that keeps its series is. +/// +/// `x > 0` is bound as a filter, so `BroadcastRuntime::drops_labels` is `false` and the broadcast +/// reaches each series of its vector side as it is, keeping its labels: what arrives at the window is +/// the boundary's fragments, still several series of one label set. +/// `unifies_its_own_contributions` passes the vector side's "cannot say" on for exactly the +/// operations that keep the labels, which is what makes the window collect them. +/// +/// The measurement is the same pair as [`a_vector_that_hands_its_series_back_is_collected_above`]'s, +/// on the same one-step window: **without** the pass `merges = 0` and the window answers **two** +/// series where one is owed, and **with** it `merges = 1` and the answer is **one** series holding +/// both positions. +/// +/// The control is the other label policy on the same shape: `x + 0` drops the metric name, the +/// runtime reports its fragments itself, and the tree gains no merge. +#[test] +fn a_filtering_comparison_that_keeps_its_labels_is_collected_above() { + let grid_two = grid(0, 20_000, 2); + let range_ms = 20_000; + let shape = |matching: BinaryMatching| { + window_sized_on( + RangeFunction::AvgOverTime, + binary_matched( + evaluated_input(0, &grid_two), + scalar(0.0, Arc::clone(&grid_two)), + Arc::clone(&grid_two), + matching, + ), + Some(Arc::clone(&grid_two)), + Arc::clone(&grid_two), + range_ms, + ) + }; + + // Without the pass: `avg_over_time((x > 0)[5m:1m])` answers one series per fragment. + let uncollected = NativeTree::try_new(shape(filter())).expect("the fixture is a legal tree"); + assert_eq!( + 0, + merges(&uncollected), + "the fixture carries a merge of its own:\n{uncollected}" + ); + let apart = evaluated_series_rows(&uncollected, &grid_two, &[(0, 1.0), (1, 2.0)]) + .expect("the runtime runs the filtering comparison"); + assert_eq!( + 2, + apart.len(), + "`x > 0` answered {} series where the un-collected window owes one per fragment: \ + {apart:?}\n{uncollected}", + apart.len() + ); + + // With the pass: the merge sits over the comparison and the window answers one series. + let tree = passed(shape(filter()), 1); + assert_eq!( + 1, + merges(&tree), + "the filtering comparison was not collected:\n{tree}" + ); + let NativeExpr::Window { input, .. } = &tree.root().expr else { + panic!("the window is not the root:\n{tree}"); + }; + let NativeExpr::Merge { input: collected } = &input.expr else { + panic!("the merge does not sit over the window's input:\n{tree}"); + }; + assert!( + matches!(&collected.expr, NativeExpr::Binary { .. }), + "the merge does not sit over the comparison:\n{tree}" + ); + // Both fragments pass the filter (1 > 0 and 2 > 0), so both positions survive the comparison; + // the merge makes them one series and the window answers one series covering both. + let series = evaluated_series_rows(&tree, &grid_two, &[(0, 1.0), (1, 2.0)]) + .expect("the runtime runs the filtering comparison"); + assert_eq!( + vec![vec![(0, 1.0), (1, 2.0)]], + series, + "the filtering comparison did not hand the window one folded series: {series:?}\n{tree}" + ); + + // The control: the same shape under an operation that drops the metric name reports its own + // fragments, so the runtime collects them and the tree gains nothing. + let dropping = passed(shape(add()), 1); + assert_eq!( + 0, + merges(&dropping), + "a label-dropping broadcast was collected as well as announced:\n{dropping}" + ); +} + +/// The merge-count table: one row per consumer, and the number of merges the pass leaves for it. +/// +/// The counts are what `unifies_its_own_contributions` decides, so this table is the whole answer in +/// one place. A child that reports its fragments itself (`Map`, `Timestamp`, a label-dropping +/// broadcast, a match of two vectors, a `Window` over one of those) is collected by the runtime's +/// `MergeInput`, and a child that has already unified (`Merge`, an `Aggregate`, two vectors matched +/// by the kernel) needs no collection at all: both are `0`. The children that answer "cannot say" +/// are the boundary itself and the nodes that hand its series back unchanged - `topk`/`limit_ratio`/ +/// `sort`, a filtering comparison - and those are the `1`s. +/// +/// `limitk` is the one consumer whose merge is *below* the operation rather than above it, because +/// `k` counts series: its row is a `1` that the `Vector` arm of [`rewrite`] inserts, not this +/// table's window. It is also the one row that gained no change from this work - the `Vector` arm +/// already collected it - which is why it is in the table: the two rows this change added are the +/// `topk`/`limit_ratio`/`sort` ones and `x > 0`, and the `0` rows above them are the ones that must +/// stay `0`. +#[test] +fn the_merge_count_table() { + let out = grid(0, 60_000, 4); + let grid_of_input = || Arc::clone(&out); + let by_job = || Grouping::By(Arc::from([Arc::from("job")])); + let window_on_input = |inner: NativeNode| { + window_function_on( + RangeFunction::AvgOverTime, + inner, + Some(grid_of_input()), + Arc::clone(&out), + ) + }; + let rows: Vec<(&str, NativeNode, usize, usize)> = vec![ + ( + "`rate(x[5m])`: a window over a raw scan", + window_over(raw_input(0), Arc::clone(&out)), + 1, + 0, + ), + ( + "`avg_over_time(abs(rate(x[5m]))[5m:1m])`: a map reports its fragments", + window_on_input(map_over(window_over(raw_input(0), Arc::clone(&out)))), + 1, + 0, + ), + ( + "`avg_over_time(timestamp(rate(x[5m]))[5m:1m])`: a timestamp reports its fragments", + window_on_input(timestamp_over(window_over(raw_input(0), Arc::clone(&out)))), + 1, + 0, + ), + ( + "`avg_over_time(rate(x[5m])[5m:1m])`: the inner window reports the fragments it collects", + window_on_input(window_over(raw_input(0), Arc::clone(&out))), + 1, + 0, + ), + ( + "a window over a merge: it has already unified", + window_on_input(merge_over(evaluated_input(0, &out))), + 1, + 0, + ), + ( + "`sum(sum_over_time(x[5m]))`: a finished aggregate has", + window_on_input(aggregate_over( + window_over(raw_input(0), Arc::clone(&out)), + Arc::clone(&out), + AggregateStage::Complete, + )), + 1, + 0, + ), + ( + "a window over `x + y`: the kernel files rows by result label set", + window_on_input(binary_over( + window_over(raw_input(0), Arc::clone(&out)), + window_over(raw_input(1), Arc::clone(&out)), + Arc::clone(&out), + )), + 2, + 0, + ), + ( + "a window over `x * 2`: a label-dropping broadcast reports its fragments", + window_on_input(binary_over( + window_over(raw_input(0), Arc::clone(&out)), + scalar(2.0, Arc::clone(&out)), + Arc::clone(&out), + )), + 1, + 0, + ), + ( + "`avg_over_time(x[5m:1m])`: a window over a fused input", + window_on_input(evaluated_input(0, &out)), + 1, + 1, + ), + ( + "`avg_over_time(topk(1, x)[5m:1m])`: a `Vector` hands its series back", + window_on_input(vector_over( + evaluated_input(0, &out), + BoundVectorOp::TopK { + k: 1.0, + top: true, + group: by_job(), + }, + )), + 1, + 1, + ), + ( + "`avg_over_time(limit_ratio(0.5, x)[5m:1m])`", + window_on_input(vector_over( + evaluated_input(0, &out), + BoundVectorOp::LimitRatio { ratio: 0.5 }, + )), + 1, + 1, + ), + ( + "`avg_over_time(sort(x)[5m:1m])`", + window_on_input(vector_over( + evaluated_input(0, &out), + BoundVectorOp::Sort(SortOrder::Asc), + )), + 1, + 1, + ), + ( + "`avg_over_time(limitk(1, x)[5m:1m])`: the merge is below the operation", + window_on_input(vector_over( + evaluated_input(0, &out), + BoundVectorOp::LimitK { + k: 1.0, + group: by_job(), + }, + )), + 1, + 1, + ), + ( + "`avg_over_time((x > 0)[5m:1m])`: a filtering comparison keeps its labels", + window_on_input(binary_matched( + evaluated_input(0, &out), + scalar(0.0, Arc::clone(&out)), + Arc::clone(&out), + filter(), + )), + 1, + 1, + ), + ]; + for (shape, root, sources, expected) in rows { + // The count is what the *pass* inserted, so a fixture that carries a merge of its own (the + // "already unified" row, whose merge is placed by hand to be the input) does not have that + // merge counted against it. + let compiled = NativeTree::try_new(root).expect("the fixture is a legal tree"); + let before = merges(&compiled); + let tree = passed(compiled.root().clone(), sources); + let inserted = merges(&tree) - before; + assert_eq!( + expected, inserted, + "{shape}: expected {expected} inserted merge(s) in\n{tree}" + ); + } +} + +#[test] +fn unchanged_complete_tree_is_a_noop() { + let out = grid(0, 60_000, 4); + let tree = NativeTree::try_new(aggregate_over( + window_over(raw_input(0), Arc::clone(&out)), + Arc::clone(&out), + AggregateStage::Complete, + )) + .unwrap(); + let result = insert_native_merge_inside_current_fused(&tree, &[source_pieces()]).unwrap(); + assert!(matches!(result.root().expr, NativeExpr::Aggregate { .. })); + assert_eq!(0, merges(&result)); +} + +#[test] +fn eval_time_has_no_label_sets() { + let out = grid(0, 60_000, 4); + let node = NativeNode { + expr: NativeExpr::EvalTime { date: None }, + output: NativeType::Scalar(Arc::clone(&out)), + }; + let props = node.properties(&[]); + assert_eq!(CanonicalGrouping::CanonicalComplete, props.grouping); + assert_eq!(CanonicalOwner::SinglePartition, props.owner); +} + +#[test] +fn vector_of_is_complete_by_construction() { + let out = grid(0, 60_000, 4); + let scalar = scalar(1.0, Arc::clone(&out)); + let output = evaluated(fields(1), Arc::clone(&out)); + let node = NativeNode { + expr: NativeExpr::VectorOf { + input: Box::new(scalar), + }, + output, + }; + assert_eq!( + CanonicalGrouping::CanonicalComplete, + node.properties(&[]).grouping + ); +} + +#[test] +fn parameterized_map_preserves_input_properties() { + let out = grid(0, 60_000, 4); + let input = window_over(raw_input(0), Arc::clone(&out)); + let output = input.output.clone(); + let node = NativeNode { + expr: NativeExpr::ParameterizedMap { + input: Box::new(input.clone()), + op: crate::native::plan::ParameterizedOp::Round, + parameters: vec![], + }, + output, + }; + assert_eq!( + input.properties(&[source_pieces()]), + node.properties(&[source_pieces()]) + ); +} diff --git a/src/promql/src/native/mod.rs b/src/promql/src/native/mod.rs index 4fb62838b73..833e6e5b5ed 100644 --- a/src/promql/src/native/mod.rs +++ b/src/promql/src/native/mod.rs @@ -29,6 +29,7 @@ mod aggregate; pub mod batch; +pub mod canonicalize; mod display; pub mod execute; pub mod metrics; diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs index 6eaad14424a..6904b450ec2 100644 --- a/src/promql/src/native/plan.rs +++ b/src/promql/src/native/plan.rs @@ -23,8 +23,9 @@ //! operation states about its input, which is what lets a new combination of supported operations //! be an ordinary tree instead of another hand-written shape. //! -//! What the tree cannot express yet is deliberate for this stage: `scalar(v)` has no node. The -//! legality below is therefore about the operations that exist, not about PromQL's whole surface. +//! What the tree cannot express yet is deliberate for this stage: [`NativeExpr::Scalar`] holds a +//! constant, so `time()` and `scalar(v)` have no node. The legality below is therefore about the +//! operations that exist, not about PromQL's whole surface. use std::sync::Arc; @@ -40,7 +41,7 @@ use crate::map_label::LabelRewrite; use crate::map_time::DateFunction; use crate::range_eval::{RangeFunction, RangeWindow}; use crate::select::SortOrder; -use crate::vector_match::{BinaryMatching, MatchingLabels, SetOp}; +use crate::vector_match::{BinaryMatching, BinaryOp, MatchingLabels, SetOp}; /// What one node produces. /// @@ -141,11 +142,27 @@ pub struct BoundInput { pub ordering: SeriesOrdering, } +impl BoundInput { + /// What the caller stated about the boundary this leaf was bound to. + /// + /// The facts travel beside the tree, one entry per source, and `source_index` is what names the + /// entry: two scans are two sources with their own placement, so reading the facts by position + /// in anything but the boundary would judge one source by another's proof. An entry the caller + /// did not supply proves nothing ([`NativeProperties::not_known`]) rather than falling back to a + /// neighbour's. + fn facts_in(&self, boundary: &[NativeProperties]) -> NativeProperties { + boundary + .get(self.source_index) + .copied() + .unwrap_or_else(NativeProperties::not_known) + } +} + /// A constant scalar bound for broadcast or a statement root. /// /// A scalar that varies per grid point is the same [`NativeType::Scalar`] as this constant: what -/// keeps a binary operation from label-matching it is its type, not its constness. The varying -/// ones are [`NativeExpr::EvalTime`] and a binary operation between scalars. +/// keeps a binary operation from label-matching it is its type, not its constness. This binding +/// only holds the constant case. #[derive(Clone, Debug)] pub struct BoundScalar { /// The constant value. @@ -202,31 +219,6 @@ pub struct BoundMap { pub ops: Arc<[MapOp]>, } -/// A math function whose scalar parameters are bound as nodes, because at least one of them varies -/// per grid point; Prometheus reads each parameter at every step. With constant parameters the same -/// function is a [`MathOp`] in a [`BoundMap`]. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -pub enum ParameterizedOp { - /// `round(v, to_nearest)`. - Round, - /// `clamp(v, min, max)`. - Clamp, - /// `clamp_min(v, min)`. - ClampMin, - /// `clamp_max(v, max)`. - ClampMax, -} - -impl ParameterizedOp { - /// How many parameters the function reads after its vector. - pub fn arity(self) -> usize { - match self { - Self::Clamp => 2, - Self::Round | Self::ClampMin | Self::ClampMax => 1, - } - } -} - /// One operation of a map, applied to each series on its own: math over its float values, a /// function of its native histograms, or a change of its labels. Each can give two series one label /// set, which is why all live in a map. @@ -237,6 +229,24 @@ pub enum MapOp { Label(Arc), } +/// A parameterized scalar math function. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ParameterizedOp { + Round, + Clamp, + ClampMin, + ClampMax, +} + +impl ParameterizedOp { + pub fn arity(self) -> usize { + match self { + Self::Clamp => 2, + _ => 1, + } + } +} + /// Which aggregate a node computes. /// /// Every kind here folds a group's values into one value per grid point. The aggregates that @@ -364,14 +374,9 @@ pub enum NativeExpr { Input(BoundInput), /// A constant scalar, evaluated on the statement's grid. Scalar(BoundScalar), - /// The evaluation time of each grid point, as `time()` reads it in seconds, or a date function - /// of it, as a date function called without an argument reads it. - /// - /// A date function is bound here rather than over `time()`: Prometheus reads the calendar from - /// the evaluation timestamp truncated to whole seconds, which the float `time()` does not carry - /// exactly for every timestamp. + /// Evaluation time or a date function evaluated directly from it. EvalTime { date: Option }, - /// `vector(s)`: the scalar as one series without labels. + /// A scalar materialized as one unlabeled series. VectorOf { input: Box }, /// An instant selection over its input. Instant { @@ -390,8 +395,7 @@ pub enum NativeExpr { input: Box, op: BoundMap, }, - /// A point-wise math function over its input whose parameters are scalar nodes, in argument - /// order. + /// Point-wise math with scalar node parameters. ParameterizedMap { input: Box, op: ParameterizedOp, @@ -442,6 +446,379 @@ pub struct NativeNode { pub output: NativeType, } +/// Where the fragments of one canonical label set live. +/// +/// A canonical label set is the label set a [`NativeExpr::Merge`] groups fragments by: `__name__` +/// plus the ordinary labels, which is exactly what [`SeriesLabels`] holds. The owner says how much +/// of one label set a single execution sees, which is what decides whether an answer computed from +/// a node's output can be the whole answer. +/// +/// No node carries this fact. It follows from how the query was placed, so it is derived by +/// [`NativeNode::properties`] from the facts a caller supplies for the tree's inputs. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CanonicalOwner { + /// Every series of every label set the node can produce lies in this execution, so each label + /// set it reads is read whole. + SinglePartition, + /// The execution holds the label sets its own canonical key names, and no other execution holds + /// a fragment of one of them. + ByCanonicalKey, + /// How the output was placed against the label sets it produces is not known. + Unknown, +} + +/// Whether a node answers at most one series per canonical label set. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum CanonicalGrouping { + /// One series per canonical label set, so a consumer that needs the whole series already has + /// it. + CanonicalComplete, + /// Pieces of label sets, which is what one scan partition holds of a metric: a consumer of + /// these answers a piece of the result. + SourcePiecesOnly, +} + +/// The static facts a node's output answers about canonical series. +/// +/// These are what a tree rewrite needs and a runtime does not: whether the output already holds one +/// series per canonical label set ([`CanonicalGrouping`]) and how much of each label set this +/// execution holds ([`CanonicalOwner`]). +/// +/// They are derived, never stored on a node: [`NativeNode::properties`] recomputes them from the +/// tree, so a rewritten tree cannot keep a fact that no longer holds. What a compiler knows about a +/// boundary it built itself travels as the [`NativeProperties`] it hands +/// [`NativeNode::properties`] for an [`NativeExpr::Input`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct NativeProperties { + /// How much of each canonical label set this execution holds. + pub owner: CanonicalOwner, + /// Whether the output holds at most one series per canonical label set. + pub grouping: CanonicalGrouping, +} + +impl NativeProperties { + /// What an input's own fragments prove: nothing about which series share a label set. + /// + /// A scan hands over the pieces of a metric that one partition holds. Nothing has grouped them + /// by label set, so the fact is not claimed; a caller that knows better says so instead. + pub fn source_pieces() -> Self { + Self { + owner: CanonicalOwner::Unknown, + grouping: CanonicalGrouping::SourcePiecesOnly, + } + } + + /// What a node the caller said nothing about proves: the same as [`Self::source_pieces`], + /// because a boundary the caller left unstated proves no more than a plain scan. + pub fn not_known() -> Self { + Self::source_pieces() + } + + /// What a node whose output label sets are its own still proves about *ownership*. + /// + /// Ownership is the first question: does this execution hold every piece of the label sets the + /// output names? An operator whose output identity is not the input's keys cannot carry the + /// input's answer over. A rewrite can give two series one label set - `rate(x)` and `rate(y)` + /// coincide on `{job="z"}` where `x` and `y` did not - and an aggregate's output identity is its + /// group key while a match's is its modifiers, so a co-location proof about the *input* keys says + /// nothing about the output keys: two input label sets this execution holds and one it does not + /// can fold into one group here, and that group is then not whole. + /// + /// `SinglePartition` survives because it is not a claim about keys at all - it holds every series + /// there is - and `ByCanonicalKey` becomes `Unknown`, because the keys it named are not the ones + /// the output answers under. + /// + /// This answers ownership only. What a node's *output* is grouped like is a separate question, + /// answered where that node is derived: an operator that has collapsed to one series per key + /// says so itself and is not downgraded by this. + fn owner_of_its_own_labels(self) -> CanonicalOwner { + match self.owner { + CanonicalOwner::ByCanonicalKey => CanonicalOwner::Unknown, + other => other, + } + } + + /// What a node whose output label sets are its own proves: neither question carries over. + /// + /// The ownership is [`Self::owner_of_its_own_labels`] and the grouping falls back to + /// [`CanonicalGrouping::SourcePiecesOnly`], which is the honest answer for a node that has not + /// established that its output is one series per key. A node that *has* established it - a + /// finished aggregate, a match, a set operation - derives its grouping itself instead of calling + /// this. + fn label_sets_of_its_own(self) -> Self { + Self { + owner: self.owner_of_its_own_labels(), + grouping: CanonicalGrouping::SourcePiecesOnly, + } + } + + /// What a merge proves. + /// + /// A merge groups the fragments it received by canonical label set and hands over one series per + /// label set, so it is the node that establishes that fact for what it read: what it emits is + /// one series per label set among the fragments that arrived. That is what a consumer folding a + /// series needs of them. + /// + /// What it read says nothing about what it emits past that, and it cannot collect a fragment that + /// never arrived: the placement of what it read is still the placement of what it wrote, so the + /// ownership is kept unchanged and never upgraded. A merge running inside a partition that holds + /// pieces of several label sets therefore still leaves the ownership `Unknown`, even though it + /// grouped every fragment it was given. + fn merged(self) -> Self { + Self { + grouping: CanonicalGrouping::CanonicalComplete, + ..self + } + } + + /// What two inputs' facts prove together. + /// + /// A binary operation over two vectors, and a set operation, answers series of both of its sides, + /// so its output holds every label set either side produced: the ownership is only what both + /// sides agree on. A side that is a constant is never united in, because it holds no label set: + /// the vector side's facts are the broadcast's alone. + fn united(self, other: Self) -> Self { + Self { + owner: if self.owner == other.owner { + self.owner + } else { + CanonicalOwner::Unknown + }, + ..self + } + } +} + +impl NativeNode { + /// The static properties of this node, given what the caller knows about what is below it. + /// + /// `boundary` is the facts of the fused inputs this tree was compiled over, indexed by the + /// [`BoundInput::source_index`] each leaf was bound to: it is the *only* thing this derivation + /// is handed. A leaf answers from its own entry and from nothing else, and every other node's + /// facts are its children's, in child order, which the recursion below asks for one child at a + /// time. There is hence no place a node could read one child's facts for another's — the mistake + /// that made a tree with two sources judge every source by the first one's proof, and a scan's + /// pieces canonical by a neighbour's. + /// + /// Nothing is stored: the facts are derived from the tree and the boundary every time, so a + /// rewritten tree cannot keep a fact the rewrite invalidated. Asking for a node's facts walks the + /// subtree below it, so facts for every node of a tree cost one walk per consumer; a tree is a + /// few dozen nodes and this is compile-time work, which is worth a derivation with no second + /// representation to keep in step. + pub fn properties(&self, boundary: &[NativeProperties]) -> NativeProperties { + match &self.expr { + // A leaf *is* the boundary it was bound to, and the facts are its own entry's. + NativeExpr::Input(binding) => binding.facts_in(boundary), + // A constant is one value everywhere and holds no label set at all, so it is whole as it + // stands: a merge has nothing to canonicalize. + NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } => no_label_sets(), + NativeExpr::VectorOf { .. } => NativeProperties { + owner: CanonicalOwner::SinglePartition, + grouping: CanonicalGrouping::CanonicalComplete, + }, + // A selection answers one series from one series' rows and keeps the label set it read, + // so the whole series stays one stream under that label set. + // + // The selection that reports the sample's *time* keeps the label set it read only when + // the time it reports is the info series' own (`InfoTimestamp`, which `info` reads): the + // runtime files that under every label of the series. `Timestamp` (`timestamp(x)`) + // drops the metric name before it builds the series, so `a{job="x"}` and `b{job="x"}` + // both arrive as `{job="x"}` and nothing the input proved about which series share a + // label set survives that. + NativeExpr::Instant { input, op } => { + let input = input.properties(boundary); + // `Sample` answers the label set the selection read, and so does `InfoTimestamp`. + // `Timestamp` reports a time instead of the value and rewrites the labels on the + // way, so its facts are a rewrite's. + match op.value { + InstantValue::Sample | InstantValue::InfoTimestamp => input, + InstantValue::Timestamp => input.label_sets_of_its_own(), + } + } + // A window folds a range of one series' samples and hands the answer over as that + // series, so the input's grouping is the output's — unless the function drops the metric + // name, which `RangeFunction::drops_metric_name` reports for every function but + // `last_over_time` and `first_over_time`. A dropping function rewrites the labels in + // `RangeEvaluator::emit`/`empty_chunk`, so `rate(a)` and `rate(b)` can coincide on + // `{job="x"}` where `a` and `b` did not: a proof about the input's label sets says + // nothing about the output's. The merge the window runtime can run itself + // (`RuntimeOp::Window`'s `merge`) joins the chunks it *reads*, so it cannot collect the + // aliases its own rewrite creates afterwards: those are a consumer's to merge, which is + // a merge above the window and not one below it. + NativeExpr::Window { input, op } => { + let input_properties = input.properties(boundary); + if op.function.drops_metric_name() { + input_properties.label_sets_of_its_own() + } else if matches!(unifies_its_own_contributions(input), Some(false)) { + input_properties.label_sets_of_its_own().merged() + } else { + input_properties + } + } + // A timestamp replaces each value with its sample's time, and the labels with + // `drop_metric_name`'s (`NativeNode::timestamp` in + // [`execute`](crate::native::execute)), so the identity it answers under is not the one + // it read. + NativeExpr::Timestamp { input } => input.properties(boundary).label_sets_of_its_own(), + // A map is point-wise over one series at a time, so it reads its input as it is: what it + // does to the label set is a fact about its output. + NativeExpr::Map { input, op } => { + let input = input.properties(boundary); + if op.ops.iter().all(preserves_canonical_identity) { + input + } else { + input.label_sets_of_its_own() + } + } + NativeExpr::ParameterizedMap { input, .. } => input.properties(boundary), + // A merge is the canonicalizer: it answers one series per label set whatever pieces it + // read. + // `info` enriches each base series with the info series that share its identifying + // labels, which is a read across both sides' whole vectors: what it answers under is + // decided per series against the other side, so neither side's grouping is the output's. + NativeExpr::Info { base, info, .. } => base + .properties(boundary) + .united(info.properties(boundary)) + .label_sets_of_its_own(), + NativeExpr::Merge { input } => input.properties(boundary).merged(), + // An aggregate answers two questions with two answers, and they are not the same one. + // + // *Grouping*: a finished fold is one series per group (`FloatAggregator::finish` builds + // one `Series` per group, and `AggregateRuntime::compute` hands over exactly that), and + // two groups are two label sets, so the output already holds at most one series per + // canonical label set. Nothing above it has a series to collect, whatever the labels it + // ended up with. `Partial` answers aggregate state instead, which is not an evaluated + // series at all: it carries no series per label set, so it makes no completeness claim. + // + // *Ownership*: the output's identity is the group key, so a co-location proof about the + // input's canonical keys does not name the groups this node invents - two input label + // sets this execution holds and one it does not can fold into one group here - and the + // owner is re-derived rather than carried over. + NativeExpr::Aggregate { input, .. } => { + let input = input.properties(boundary); + NativeProperties { + owner: input.owner_of_its_own_labels(), + grouping: match &self.output { + NativeType::Samples(_) => CanonicalGrouping::CanonicalComplete, + NativeType::Scalar(_) | NativeType::AggregateState(_) => { + CanonicalGrouping::SourcePiecesOnly + } + }, + } + } + // A vector operation is decided one operator at a time, because the seven do not answer + // the same shape. + // + // *Grouping*: `topk`, `limitk`, `limit_ratio` and `sort` hand back series of their input + // with their labels unchanged - `topk` rebuilds the selected points of the series it + // picked, `limitk` the same over its merged input, `limit_ratio` keeps whole series and + // `sort` reorders them - so what the input held about label sets is what the output + // holds. `limitk` reads one series per label set only because the insertion pass (or the + // runtime, when the child shares runs) merges its input, which is why its grouping is + // the input's and cannot be claimed on its own. `absent` answers one series of its own, + // `histogram_quantile`/`histogram_fraction` merge label-set pieces before the kernel + // (`vector_series` calls `merge_same_labelset`) and answer one series per set, and + // `count_values` files one series per group and value: all three are complete. + // + // *Ownership*: the same four that keep their labels keep the ownership too, while the + // three that invent a label set - `absent`'s matcher labels, a histogram's shared bucket + // labels, `count_values`' value label - re-derive it from a proof about keys they no + // longer name. + NativeExpr::Vector { input, op } => { + let limitk_merges = matches!(op, BoundVectorOp::LimitK { .. }) + && matches!(unifies_its_own_contributions(input), Some(false)); + let input = input.properties(boundary); + match op { + BoundVectorOp::LimitK { .. } if limitk_merges => input.merged(), + BoundVectorOp::TopK { .. } + | BoundVectorOp::LimitK { .. } + | BoundVectorOp::LimitRatio { .. } + | BoundVectorOp::Sort(_) => input, + BoundVectorOp::Absent { .. } + | BoundVectorOp::Histogram(_) + | BoundVectorOp::CountValues { .. } => NativeProperties { + owner: input.owner_of_its_own_labels(), + grouping: CanonicalGrouping::CanonicalComplete, + }, + } + } + NativeExpr::Binary { + left, + right, + matching, + } => match (&left.output, &right.output) { + // Two constants make a constant: there is no label set to canonicalize. + (NativeType::Scalar(_), NativeType::Scalar(_)) => no_label_sets(), + // A scalar broadcast reaches each series of its vector side as it is, unless the + // operation drops the metric name on the way: arithmetic and a `bool` comparison do, + // a filtering comparison keeps the label set it read. + (NativeType::Samples(_), NativeType::Scalar(_)) + | (NativeType::Scalar(_), NativeType::Samples(_)) => { + // Only the vector side's facts are the output's, because a constant contributes + // no label set at all: the broadcast answers the vector side's series under the + // label set it read, or under that label set rewritten. + // + // A `united` of the two sides cannot express that, and answers wrong in both + // directions. It keeps the left side's grouping, so with the constant on the + // left the output would claim the constant's `CanonicalComplete` for a vector + // side that is still only source pieces - the alias a merge above has to + // collect would then never be collected. And it keeps an ownership only the two + // sides agree on, so with the constant on the right a vector side placed by + // canonical key would be downgraded to `Unknown`, which collapses the owner + // partitions a merge above it could have stayed within. + // + // The side that holds the samples is therefore the base either way, so `x > 0` + // and `0 < x` derive the same facts. + let vector = if left.output.samples().is_some() { + left + } else { + right + }; + let vector = vector.properties(boundary); + if broadcast_drops_labels(matching) { + vector.label_sets_of_its_own() + } else { + vector + } + } + // Two vectors are matched by label. The two questions are answered apart. + // + // *Grouping*: the kernel files its rows by *result label set* (`Results` is keyed by + // the label set a pairing's sample ends under) and hands over one series per set, + // refusing two values of a field at one point (`Results::sort_rows` -> "vector + // cannot contain metrics with the same labelset"). What it answers is therefore one + // series per canonical label set however the operands arrived. + // + // *Ownership*: the modifiers decide what label set a sample ends under, so the + // match's keys are not the keys a co-location proof about either side names; a match + // answers series of both sides, so its ownership is only what both agree on and + // `label_sets_of_its_own` re-derives it for the output's keys. + (_, _) => NativeProperties { + owner: left + .properties(boundary) + .united(right.properties(boundary)) + .label_sets_of_its_own() + .owner, + grouping: CanonicalGrouping::CanonicalComplete, + }, + }, + // A set operation keeps each side's samples as they are, and its kernel files them by + // label set as a match does, refusing two values of a field at one point + // (`set_vectors` -> `Results::sort_rows`). *Grouping*: one series per label set, so + // nothing above it has a series to collect. *Ownership*: the samples keep the labels + // they arrived with, so the label sets the output names are the ones the sides' keys + // named and the two proofs unite without being re-derived. + NativeExpr::Set { left, right, .. } => NativeProperties { + owner: left + .properties(boundary) + .united(right.properties(boundary)) + .owner, + grouping: CanonicalGrouping::CanonicalComplete, + }, + } + } +} + /// Whether `node` reads the whole input vector at a point rather than one series at a time. /// /// An `Aggregate` folds every series of its group, a `Binary` and a `Set` pair two vectors' series @@ -452,9 +829,9 @@ pub struct NativeNode { /// `Merge` answers one series per label set, and the label sets it answers span every input series: /// two of them can collapse into the one it emits. It is therefore whole-vector too, which is why it /// also cannot be moved below a boundary that holds only part of a metric. Everything else - an -/// input, a constant or `time()`, a selection, a window, a map, `vector(s)`, and a binary operation -/// with a scalar on one side, whose broadcast reaches each series the other side answers - answers -/// one series from one series' rows, as long as its own operands do. +/// input, a scalar, a selection, a window, a map, and a binary operation with a constant on one +/// side, whose broadcast reaches each series the other side answers - answers one series from one +/// series' rows. /// /// Both a region-side placement and the native route's admission ask this, and they must agree: a /// tree admitted here but refused there becomes a node the physical layer cannot build, and one @@ -462,25 +839,25 @@ pub struct NativeNode { pub fn reads_across_series(node: &NativeNode) -> bool { match &node.expr { NativeExpr::Input(_) | NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } => false, - NativeExpr::Instant { input, .. } - | NativeExpr::Window { input, .. } - | NativeExpr::Map { input, .. } - | NativeExpr::Timestamp { input } - | NativeExpr::VectorOf { input } => reads_across_series(input), - NativeExpr::Binary { left, right, .. } => { - // A scalar side has no labels and is broadcast over the other side's series unchanged, - // so the operation reads what its two sides read. Two vectors are matched by label, - // which pairs series of the whole vector. - let scalar = |side: &NativeNode| matches!(side.output, NativeType::Scalar(_)); - if scalar(left) || scalar(right) { - reads_across_series(left) || reads_across_series(right) - } else { - true - } + NativeExpr::Window { input, .. } => { + matches!(unifies_its_own_contributions(input), Some(false)) + || reads_across_series(input) } - NativeExpr::ParameterizedMap { - input, parameters, .. - } => reads_across_series(input) || parameters.iter().any(reads_across_series), + NativeExpr::Instant { input, .. } + | NativeExpr::VectorOf { input } + | NativeExpr::Map { input, .. } + | NativeExpr::ParameterizedMap { input, .. } + | NativeExpr::Timestamp { input } => reads_across_series(input), + NativeExpr::Binary { left, right, .. } => match (&left.expr, &right.expr) { + // Two constants answer the same value for every series there is, so the operation reads + // none of them; one constant is broadcast over the other side's series unchanged, so what + // the operation reads is what that side reads. + (NativeExpr::Scalar(_), NativeExpr::Scalar(_)) => false, + (NativeExpr::Scalar(_), _) => reads_across_series(right), + (_, NativeExpr::Scalar(_)) => reads_across_series(left), + // Two vectors are matched by label, which pairs series of the whole vector. + (_, _) => true, + }, NativeExpr::Aggregate { .. } | NativeExpr::Vector { .. } | NativeExpr::Set { .. } @@ -489,17 +866,239 @@ pub fn reads_across_series(node: &NativeNode) -> bool { } } -/// Whether `node` is a supported child of Merge: only an input with optional windows and maps. -fn merge_shape_supported(node: &NativeNode) -> bool { +/// Whether `node` answers a value for the whole vector, so a tree holding it has to see every series +/// in one execution rather than one partition's worth at a time. +/// +/// A `Vector` operation (`topk`, `bottomk`, `limitk`, `limit_ratio`, `sort`, `absent`, +/// `histogram_quantile`, `histogram_fraction`, `count_values`) picks, orders or reads its input's +/// series against each other, and a `Complete` aggregate folds every contribution to a group: given +/// part of the metric, either one answers part of the result instead of the result. A `Partial` or +/// `Final` aggregate's own stage is not: its contract is exactly the part it was handed, so only what +/// sits below it is followed. +/// +/// This form passes an empty boundary, so what it asks of a merge is what +/// [`needs_single_execution_in`] answers for one: the merge's own need, which is its input's. Whether +/// a merge can be placed at all is [`merges_are_placeable_in`]'s question, and it is not this one - +/// it is asked by a caller that knows which execution it is placing the tree in. The native route's +/// placement (`NativePlacement::of` in `promql::planner`) and the region-side admission ask this and +/// both have no boundary facts at hand; a caller that has them, as the canonical-merge rewrite does, +/// asks [`needs_single_execution_in`]. +/// +/// Everything else recurses into its children. A window (`rate`, `sum_over_time`, ...), a map, an +/// instant selection and a timestamp compute one series from one series' rows, and a binary +/// operation that broadcasts a constant is decided by its vector side alone, so a tree built from +/// those still answers each of its series on its own: splitting it across scan partitions splits the +/// answer rather than an operator's view. +/// +/// A binary or set operation that *matches* two vectors is not decided by its sides alone, though. +/// It pairs the series of one side with the series of the other that shares the match's label set, +/// which is a read across the whole vector on both sides, and with no boundary facts at hand a match +/// has no proof that one execution holds both. So it needs one execution. +/// +/// This is about placement, not admission: it says how many executions the tree needs, not whether +/// it may run over part of the metric ([`reads_across_series`] answers that). +pub fn needs_single_execution(node: &NativeNode) -> bool { + needs_single_execution_in(node, &[]) +} + +/// Whether `node` answers a value for the whole vector, given what the caller knows about where the +/// sources below it sit. +/// +/// `boundary` is the same slice [`NativeNode::properties`] takes: the facts of the fused inputs the +/// tree was compiled over, indexed by `source_index`. +/// +/// A `Merge` needs every piece of ONE canonical label set, and one execution is one way of having +/// them - but which way applies is not this predicate's question. A merge's own need is therefore +/// its input's ([`NativeExpr::Merge`] below), and whether the execution this tree is placed in can +/// answer a merge at all is [`merges_are_placeable_in`], asked by the caller that knows the +/// placement: a canonical merge is right in a single execution, and in a per-partition one only +/// where the boundary proves the pieces already arrive together ([`pieces_co_located`]). +/// +/// A *match* needs something else. A `Binary` over two vectors and a `Set` pair each series of one +/// side with the series of the other that shares the label set the modifiers name, and a whole match +/// owes an error when one side holds two series of that label set. Canonical-key co-location does +/// not prove that: the keys it proves whole are the canonical label sets, and the match's label set +/// is the caller's. Two series that agree on `on(host)` can differ in every other label and sit +/// under two canonical keys, so two owner partitions can each hold one series of that match group, +/// pair them locally, and never run the duplicate-series check over the real group: the error a +/// whole match reports is silently not reported. A match therefore needs one execution unless both +/// sides are proven to sit in it ([`matching_co_located`]). A broadcast is not a match, so a +/// constant's side is not asked for that proof. +pub fn needs_single_execution_in(node: &NativeNode, boundary: &[NativeProperties]) -> bool { match &node.expr { - NativeExpr::Input(_) => true, - NativeExpr::Window { input, .. } | NativeExpr::Map { input, .. } => { - merge_shape_supported(input) + NativeExpr::Input(_) | NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } => false, + NativeExpr::Instant { input, .. } + | NativeExpr::Window { input, .. } + | NativeExpr::VectorOf { input } + | NativeExpr::Map { input, .. } + | NativeExpr::ParameterizedMap { input, .. } + | NativeExpr::Timestamp { input } => needs_single_execution_in(input, boundary), + // `info` pairs each base series with the info series that share its identifying labels, + // which is a match by a label set of the caller's choosing rather than by a canonical key - + // the same reason a two-vector `Binary` needs one execution unless both sides are proven to + // sit in it. + NativeExpr::Info { base, info, .. } => { + needs_single_execution_in(base, boundary) + || needs_single_execution_in(info, boundary) + || !(matching_co_located(base, boundary) && matching_co_located(info, boundary)) } - _ => false, + NativeExpr::Aggregate { input, op } => { + op.stage == AggregateStage::Complete || needs_single_execution_in(input, boundary) + } + // A match pairs the series of both sides by the label set its modifiers name, which is not + // the canonical key a placement proves whole ([`matching_co_located`]), so it needs one + // execution unless both sides are already in it. A broadcast is not a match: it pairs a + // vector with one value, so the constant's side holds no series to pair and proves nothing it + // was asked for - `x > 0` and `x * 2` stay as parallel as the vector side. + NativeExpr::Binary { left, right, .. } => { + let both_vectors = matches!( + (&left.output, &right.output), + (NativeType::Samples(_), NativeType::Samples(_)) + ); + needs_single_execution_in(left, boundary) + || needs_single_execution_in(right, boundary) + || (both_vectors + && !(matching_co_located(left, boundary) + && matching_co_located(right, boundary))) + } + // A set operation keeps samples of two evaluated vectors, so both sides are always matched: + // `validate` refuses a scalar on either side and there is no broadcast form of it. + NativeExpr::Set { left, right, .. } => { + needs_single_execution_in(left, boundary) + || needs_single_execution_in(right, boundary) + || !(matching_co_located(left, boundary) && matching_co_located(right, boundary)) + } + NativeExpr::Vector { .. } => true, + // A merge needs every piece of one canonical label set, and not every label set in one + // execution: given the pieces it answers one series per label set. One execution provides + // that for any input; a per-partition execution provides it only when the pieces are + // co-located. Which of the two applies is not this predicate's question: it answers what the + // tree needs from an execution, and a merge's own need is its input's. Whether the merge can + // be *placed* here is `merges_are_placeable_in`, asked by the caller that knows the + // placement. + NativeExpr::Merge { input } => needs_single_execution_in(input, boundary), } } +/// Whether every merge in `node` can be answered by an execution placed as `co_located` says. +/// +/// `co_located` is what the caller already decided about the execution this tree will run in: +/// `true` when the execution reads every input partition, so every piece of every label set is +/// present and a merge is always placeable; `false` when it reads one partition, so a merge is +/// placeable only where [`pieces_co_located`] proves the pieces arrive together. +/// +/// This is deliberately not part of [`needs_single_execution_in`]: a merge that cannot be placed +/// is a tree this route does not support here, not a tree that needs one execution. Collapsing the +/// partitions to make it work would take the caller's parallelism away to serve a shape nobody +/// proved was placeable. +pub fn merges_are_placeable_in( + node: &NativeNode, + boundary: &[NativeProperties], + co_located: bool, +) -> bool { + match &node.expr { + NativeExpr::Input(_) | NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } => true, + NativeExpr::Instant { input, .. } + | NativeExpr::VectorOf { input } + | NativeExpr::Map { input, .. } + | NativeExpr::ParameterizedMap { input, .. } + | NativeExpr::Timestamp { input } => merges_are_placeable_in(input, boundary, co_located), + NativeExpr::Window { input, .. } => { + (!matches!(unifies_its_own_contributions(input), Some(false)) + || co_located + || pieces_co_located(input, boundary)) + && merges_are_placeable_in(input, boundary, co_located) + } + // A merge reads the pieces of one canonical label set and answers one series of it, so it is + // placeable where the execution holds every piece. The recursion does not stop here: a merge + // of a merge has to be placeable at every level, and a label-rewriting node between two of + // them rewrites the identity the inner proof was about, so the inner merge can be the one + // that has no proof left. + NativeExpr::Merge { input } => { + (co_located || pieces_co_located(input, boundary)) + && merges_are_placeable_in(input, boundary, co_located) + } + NativeExpr::Info { base, info, .. } => { + merges_are_placeable_in(base, boundary, co_located) + && merges_are_placeable_in(info, boundary, co_located) + } + NativeExpr::Aggregate { input, .. } | NativeExpr::Vector { input, .. } => { + merges_are_placeable_in(input, boundary, co_located) + } + NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => { + merges_are_placeable_in(left, boundary, co_located) + && merges_are_placeable_in(right, boundary, co_located) + } + } +} + +/// Whether the boundary proves that the pieces of every canonical label set below `node` arrive in +/// one execution, so a merge over them reads each label set whole. +/// +/// The proof is ownership, and it is the ownership that survives to `node` that counts: an operator +/// that rewrites the identity - a name-dropping window, an aggregate, a map - downgrades a per-key +/// ownership to `Unknown`, because the label sets it invents are not the ones the boundary's +/// canonical keys name. So `SinglePartition` ("this execution holds every series there is") or +/// `ByCanonicalKey` ("it holds the label sets its own key names and no other execution holds a +/// fragment of one of them") is what a merge needs, and anything else - `Unknown` in particular, +/// which is all a raw scan proves, since it partitions by storage `read_key` and not by canonical +/// key - leaves the pieces spread over partitions. +/// +/// This is a merge's proof and only a merge's, because a merge groups by exactly the key it names: a +/// canonical label set. It proves nothing about the label set a *match* pairs by +/// ([`matching_co_located`]), which is the modifiers' and not the canonical key's, so a match that +/// read this answer would take a per-key placement for a whole one. +fn pieces_co_located(node: &NativeNode, boundary: &[NativeProperties]) -> bool { + matches!( + node.properties(boundary).owner, + CanonicalOwner::SinglePartition | CanonicalOwner::ByCanonicalKey + ) +} + +/// Whether `node`'s series all sit in this one execution, so a label match over them pairs every +/// series it should. +/// +/// A match pairs the series of two sides by the label set its own modifiers name (`on(host)`, +/// `ignoring(zone)`, ...), and a whole match owes an error when one side holds two series of that +/// label set. Only an execution holding every series it could pair can answer that, which is what +/// `SinglePartition` states and what nothing weaker states. +/// +/// `ByCanonicalKey` in particular does not, and that is the whole point of this predicate being +/// separate from [`pieces_co_located`]: it proves the canonical label sets its keys name are whole, +/// and a match does not read those keys. `{host="h", zone="a"}` and `{host="h", zone="b"}` are two +/// canonical label sets owned by two partitions, they are one `on(host)` match group, and each owner +/// would pair its own series against its own and report no duplicate - the answer a whole match +/// refuses. +/// +/// The facts are read off `node` rather than off `node`'s children because what has to be whole is +/// what the match pairs, which is what `node` answers: the same derivation every merge proof uses. +fn matching_co_located(node: &NativeNode, boundary: &[NativeProperties]) -> bool { + node.properties(boundary).owner == CanonicalOwner::SinglePartition +} + +/// Whether `node` is a supported child of a [`NativeExpr::Merge`], and the evaluated vector it hands +/// over. +/// +/// A merge receives fragments and groups them by canonical label set, combining the ones a label set +/// shares by grid position; it names neither the tree nor the operator that produced a fragment. What +/// it reads is therefore the whole of its requirement: samples evaluated on one grid in one field +/// layout, which is an evaluated vector and nothing else. This is that type check and no more, so no +/// node kind is refused for what it computes: a scan's samples, a window's rate, an aggregate's fold +/// and a binary operation's match all hand over the same thing, and which one of them did says +/// nothing about whether the fragments merge. +/// +/// The collection does rely on the shape of the stream - a series arrives whole, its chunks +/// contiguous and its last one marked as closing it - and every node that answers an evaluated vector +/// keeps it. A window, a map, an instant selection, a timestamp and a scalar broadcast hand their +/// input's closing on, and an aggregate, a vector operation, a binary or set operation and a merge +/// itself close each series they answer, because each decides a whole series at once. There is hence +/// no operator above a merge that could break the collection, and none has to be listed here; the +/// collector checks the shape chunk by chunk before it keeps a fragment, which is where a stream that +/// broke it is refused. +fn merge_shape_supported(node: &NativeNode) -> Option<(&Arc, &Arc)> { + evaluated_vector(&node.output) +} + /// A bound evaluation tree. /// /// The tree itself carries no run-time state: every execution builds its own, so one compiled plan @@ -540,22 +1139,19 @@ impl NativeNode { } } NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } => { - // A scalar's output is the scalar type: one value per grid point and no labels. if !matches!(self.output, NativeType::Scalar(_)) { return Err(invalid("a scalar node does not produce a scalar")); } } NativeExpr::VectorOf { input } => { input.validate()?; - let NativeType::Scalar(input_grid) = &input.output else { - return Err(invalid(&format!( - "vector() consumes a scalar, not {}", - input.output.describe() - ))); - }; - let output = ensure_samples(&self.output, "vector()")?; - ensure_field_count(output.fields.len(), 1, "vector()")?; - ensure_grid(&self.output, input_grid, "vector()")?; + if !matches!(input.output, NativeType::Scalar(_)) { + return Err(invalid("vector() consumes a scalar")); + } + ensure_samples(&self.output, "vector()")?; + if !self.output.samples().is_some_and(|s| s.fields.len() == 1) { + return Err(invalid("vector() produces one field")); + } } NativeExpr::Instant { input, op } => { input.validate()?; @@ -583,13 +1179,14 @@ impl NativeNode { } NativeExpr::Merge { input } => { input.validate()?; - if !merge_shape_supported(input) { - return Err(invalid( - "a merge is only supported over input/window/map trees", - )); - } - let Some((_input_schema, input_grid)) = evaluated_vector(&input.output) else { - return Err(invalid("a merge consumes an evaluated vector")); + // A merge canonicalizes evaluated fragments, so an evaluated vector is the whole of + // what it needs of its input: [`merge_shape_supported`] is that check, and its + // documentation says why no shape of tree above those fragments matters. + let Some((_input_schema, input_grid)) = merge_shape_supported(input) else { + return Err(invalid(&format!( + "a merge consumes an evaluated vector, not {}", + input.output.describe() + ))); }; ensure_samples(&self.output, "a merge")?; ensure_same_fields(&input.output, &self.output, "a merge")?; @@ -649,36 +1246,18 @@ impl NativeNode { ensure_grid(&self.output, input_grid, "a map")?; } NativeExpr::ParameterizedMap { - input, - op, - parameters, + input, parameters, .. } => { input.validate()?; - let Some((_, input_grid)) = evaluated_vector(&input.output) else { - return Err(invalid(&format!( - "a map consumes an evaluated vector, not {}", - input.output.describe() - ))); - }; - if parameters.len() != op.arity() { - return Err(invalid(&format!( - "{op:?} reads {} parameter(s), but {} are bound", - op.arity(), - parameters.len() - ))); - } for parameter in parameters { parameter.validate()?; - if !matches!(¶meter.output, NativeType::Scalar(grid) if grid.spec() == input_grid.spec()) - { - return Err(invalid( - "a map parameter is not a scalar on its input's grid", - )); - } } - ensure_samples(&self.output, "a map")?; - ensure_same_fields(&input.output, &self.output, "a map")?; - ensure_grid(&self.output, input_grid, "a map")?; + let Some((_, input_grid)) = evaluated_vector(&input.output) else { + return Err(invalid("a parameterized map consumes an evaluated vector")); + }; + ensure_samples(&self.output, "a parameterized map")?; + ensure_same_fields(&input.output, &self.output, "a parameterized map")?; + ensure_grid(&self.output, input_grid, "a parameterized map")?; } NativeExpr::Vector { input, .. } => { input.validate()?; @@ -902,8 +1481,16 @@ impl NativeNode { ))); }; // The kernel reads the info side's values as sample times, which only an instant - // selection that reports them produces. - let NativeExpr::Instant { op, .. } = &info.expr else { + // selection that reports them produces. A canonical merge above that selection only + // collects the pieces of one label set into one series, so it hands the times on + // unchanged and the check looks through it. Nothing else is looked through: an + // evaluated value vector is not a set of sample times, and reading one as times + // would compare a sample's value against another's. + let mut source = info.as_ref(); + while let NativeExpr::Merge { input } = &source.expr { + source = input; + } + let NativeExpr::Instant { op, .. } = &source.expr else { return Err(invalid( "info reads its info series from an instant selection", )); @@ -935,6 +1522,191 @@ impl NativeNode { } } +/// The properties of a value that holds no label set: a constant, or a binary operation of two of +/// them. +/// +/// There is nothing to canonicalize, and a constant is the same value in every execution. +fn no_label_sets() -> NativeProperties { + NativeProperties { + owner: CanonicalOwner::SinglePartition, + grouping: CanonicalGrouping::CanonicalComplete, + } +} + +/// Whether a map operation leaves the label set it reads as it is. +/// +/// This is the rule the RFC states as "a pure numeric map inherits grouping", answered by the +/// kernels instead: none of the three operations leaves a label set alone, so no map preserves +/// grouping. +/// +/// The per-variant answer is the decision point: a new [`MapOp`] has to say here what it writes. +fn preserves_canonical_identity(op: &MapOp) -> bool { + match op { + // `map_math`, `map_negate` and `map_arithmetic` all end with `drop_metric_name`, which + // removes `__name__` together with the type and unit labels: `abs(x)` and `x` no longer hold + // one label set. + MapOp::Math(_) => false, + // `map_native_histogram` follows Prometheus' `simpleHistogramFunc`: it drops `__name__` too + // (keeping the type and unit), so a histogram function's output is not its input's label + // set. + MapOp::Histogram(_) => false, + // A rewrite writes a caller-chosen destination label, `__name__` included, and can empty it, + // so a series ends under a label set the rewrite built. + MapOp::Label(_) => false, + } +} + +/// Whether a binary operation drops the metric name when one side is a scalar broadcast. +/// +/// `BroadcastRuntime::drops_labels` in [`execute`](crate::native::execute) is the rule the runtime +/// applies: arithmetic and a `bool` comparison answer a series under the metric name dropped, and a +/// filtering comparison keeps the label set it read. +fn broadcast_drops_labels(matching: &BinaryMatching) -> bool { + !matches!( + matching.op, + BinaryOp::Eq | BinaryOp::Ne | BinaryOp::Gt | BinaryOp::Lt | BinaryOp::Ge | BinaryOp::Le + ) || matching.return_bool +} + +/// Whether `node` hands over one logical series per canonical label set, or several. +/// +/// This is the rule the runtime applies when it decides whether a consumer above `node` has to +/// collect what `node` emits, asked of the plan node rather than of the runtime node. It lives here +/// so that `RuntimeNode::build` in [`execute`](crate::native::execute) and the insertion pass +/// ([`insert_native_merge_inside_current_fused`]) read one rule: a second copy of it would be a +/// rule that can silently disagree with this one about the same tree. +/// +/// Three answers, because the flag the runtime computes carries two facts and a boundary answers +/// neither: +/// +/// - `Some(true)`: the output is one series per canonical label set, established here. The runtime +/// marks such a node `shared = false` and nothing above it collects. +/// - `Some(false)`: the output may hold several series of one label set, and this node *says so*. +/// The runtime marks it `shared = true`, so a consumer that needs one unified series collects it +/// through `MergeInput`. A [`NativeExpr::Merge`] inserted above such a node would turn that +/// collection off - a merge answers `Some(true)`, so the consumer would stream an already-merged +/// series - and the pass inserts none there. +/// - `None`: the output may hold several, and this node says nothing: it has neither a proof to +/// offer nor a report to make. Only a boundary ([`NativeExpr::Input`]) answers this, and a node +/// that hands its input's series back unchanged passes the answer on. The runtime's flag is +/// `false` here too, so the flag alone cannot tell this case from a proof - and `None` is exactly +/// the case the pass exists for: the one state in which neither the runtime nor the tree collects +/// anything. +/// +/// `matches!(answer, Some(false))` is the runtime's `shared` flag bit for bit, which is what makes +/// the two readers' agreement checkable: `the_predicate_and_the_runtime_agree` in +/// [`execute`](crate::native::execute)'s tests asserts it node for node on +/// `RuntimeNode::build`'s own return value. +/// +/// [`insert_native_merge_inside_current_fused`]: +/// crate::native::canonicalize::insert_native_merge_inside_current_fused +#[allow(clippy::match_same_arms)] // Each arm is one line of `RuntimeNode::build`, and the reason each answers as it does is written there rather than merged into a bucket. +pub fn unifies_its_own_contributions(node: &NativeNode) -> Option { + match &node.expr { + // A boundary hands over the pieces some upstream execution produced: what it holds is one + // storage series at a time for a scan and one partition's fragments of a label set for an + // exchange, and nothing in the tree can tell which series those fragments belong to. The + // runtime reports `shared = false` for it because a merge binds samples evaluated on a + // grid, not because the pieces are one series per label set, so a proof is not this + // function's to give; it answers `None`, and the flag follows with `false`. + NativeExpr::Input(_) => None, + // A constant holds no label set at all, so it has nothing that could arrive twice. + NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } | NativeExpr::VectorOf { .. } => { + Some(true) + } + // A selection reads raw samples - `validate` refuses an evaluated input - and raw samples + // arrive one whole storage series at a time, so it answers one series per series it read, + // under the label set it read. `timestamp(x)` is the exception: it reports the sample's own + // time under the metric name dropped, which can give two series one label set. + NativeExpr::Instant { op, .. } => Some(op.value != InstantValue::Timestamp), + // The merge is the collection: whatever its input handed over, it groups the fragments by + // canonical label set and emits one series per set, so what it emits shares none. + NativeExpr::Merge { .. } => Some(true), + // The runtime's rule for a window is `drops_metric_name() || !child_shared`, so a window is + // *not* shared - and this arm answers `Some(true)` - exactly when both hold: the function + // keeps the metric name, and the child reports its own fragments. The window then collects + // them (`MergeInput`) and emits one series per label set. + // + // Every other window answers `Some(false)`, and for two different reasons that the runtime's + // formula does not separate: + // + // - the function drops the metric name, so what it emits can coincide with another series' + // label set whatever its input held (this is a real report, and the case `Map`, `Timestamp` + // and a label-dropping broadcast are in); + // - or the child does not report fragments at all - `None` for a boundary, or `Some(true)` + // for a child that has already unified. With a `None` child the window folds each + // contribution on its own, so what it emits does alias its input's fragments and answering + // `Some(false)` is exact. With an already-unified child there is only one series per label + // set to begin with, so `Some(false)` is conservative rather than exact - it makes a + // consumer above collect a stream that needs no collecting, which is what the runtime does + // too, and is the safe direction: nothing is left uncollected. + NativeExpr::Window { input, op } => Some( + !op.function.drops_metric_name() + && matches!(unifies_its_own_contributions(input), Some(false)), + ), + // A map rewrites labels: `preserves_canonical_identity` is `false` for every [`MapOp`], so + // two series can end under one label set and the runtime records what it wrote in an + // `Emitted` table. + NativeExpr::Map { .. } | NativeExpr::ParameterizedMap { .. } => Some(false), + // A timestamp reports a position under the metric name dropped. + NativeExpr::Timestamp { .. } => Some(false), + NativeExpr::Binary { + left, + right, + matching, + } => { + if evaluated_vector(&left.output).is_some() && evaluated_vector(&right.output).is_some() + { + // Two vectors are matched by the kernel, which files its rows by the label set a + // sample ends under and emits one series per set. + Some(true) + } else if broadcast_drops_labels(matching) { + // A broadcast whose operation drops the metric name can give two series of its + // vector side one label set. + Some(false) + } else { + // A filtering comparison keeps the label set it read, so the vector side's answer + // is this node's. + let vector = if left.output.samples().is_some() { + left + } else { + right + }; + unifies_its_own_contributions(vector) + } + } + // A set operation keeps each side's samples and files them by label set, one series per + // set. + NativeExpr::Set { .. } => Some(true), + // `info` indexes the info side by identifying labels and merges the results that share a + // label set. + NativeExpr::Info { .. } => Some(true), + // A finished aggregate emits one series per group, and two groups are two label sets. + // `Partial` state is not a series at all, which is not several either. + NativeExpr::Aggregate { .. } => Some(true), + // `topk`, `limit_ratio` and `sort` hand back the series they read with their labels, so the + // input's answer is theirs. The four that read the whole vector - `limitk`, `absent`, + // `histogram_*` and `count_values` - emit one series per label set: `limitk` counts series + // in input order, so its input carries the merge the pass inserts below it (or the one the + // runtime performs when that input reports fragments), and the other three group their rows + // by label set themselves. + NativeExpr::Vector { input, op } => { + let whole = matches!( + op, + BoundVectorOp::LimitK { .. } + | BoundVectorOp::Absent { .. } + | BoundVectorOp::Histogram(_) + | BoundVectorOp::CountValues { .. } + ); + if whole { + Some(true) + } else { + unifies_its_own_contributions(input) + } + } + } +} + /// The schema and grid of an evaluated vector, or `None` when `ty` is raw samples, a scalar or /// state. /// @@ -1023,5 +1795,7 @@ fn invalid(reason: &str) -> DataFusionError { DataFusionError::Internal(reason.to_string()) } +// Shared with the rewrite's tests in `canonicalize`, which prove a merge lands where these trees +// say it should. #[cfg(test)] -mod tests; +pub(crate) mod tests; diff --git a/src/query/src/promql/planner/native.rs b/src/query/src/promql/planner/native.rs index b24a14002a3..6acc45502bf 100644 --- a/src/query/src/promql/planner/native.rs +++ b/src/query/src/promql/planner/native.rs @@ -49,10 +49,11 @@ use datafusion_expr::Expr; use datatypes::series::SeriesOrdering; use datatypes::series::samples::{FieldSpec, SeriesSchema}; use promql::extension_plan::{AggregateFinalize, FusedSeries, FusedSeriesBinding, RawSeriesSource}; +use promql::native::canonicalize::insert_native_merge_inside_current_fused; use promql::native::execute::NativeExecutionState; use promql::native::plan::{ - BoundAggregate, BoundVectorOp, Grouping, MapOp, NativeExpr, NativeNode, NativeType, - reads_across_series, + BoundAggregate, BoundVectorOp, Grouping, MapOp, NativeExpr, NativeNode, NativeProperties, + NativeType, merges_are_placeable_in, reads_across_series, }; use promql::vector_match::{Cardinality, MatchingLabels, SetOp}; use promql_parser::label::{METRIC_NAME, Matchers}; @@ -705,7 +706,27 @@ impl PromPlanner { single_input_context, ))); } - let tree = compiled.tree; + // RawSeriesSource inputs can read a whole region, but this planner has no proof that + // their canonical keys are globally unique; keep the boundary conservative (`Unknown`). + let boundary = vec![NativeProperties::not_known(); selectors.len()]; + // Every single-input non-per-partition route returned above through AggregateFinalize. + // Therefore this rewrite branch is multi-input only. FusedSeries::execute reads 0..count + // for every input when inputs.len() > 1 (fused_series.rs:1407-1411), proving all pieces + // arrive in this execution. Keep the route fact explicit rather than passing a bare claim. + let co_located = multi_input; + let tree = if per_partition && !multi_input { + compiled.tree + } else { + if !co_located { + return Ok(None); + } + let rewritten = insert_native_merge_inside_current_fused(&compiled.tree, &boundary) + .context(DataFusionPlanningSnafu)?; + if !merges_are_placeable_in(rewritten.root(), &boundary, co_located) { + return Ok(None); + } + rewritten + }; // The tags are asked for before the fields, because the walk answers for a tree the // compiler produced: a scalar root has no labels at all, and a sample root's are the ones // the walk derives or - where it derives none - the ones the existing planner names. @@ -1399,7 +1420,8 @@ mod tests { use datatypes::series::SeriesOrdering; use datatypes::series::samples::{FieldSpec, SeriesSchema}; - use promql::native::plan::{NativeExpr, NativeNode, NativeTree}; + use promql::native::canonicalize::insert_native_merge_inside_current_fused; + use promql::native::plan::{NativeExpr, NativeNode, NativeProperties, NativeTree}; use session::context::QueryContextBuilder; use super::{ @@ -1429,6 +1451,27 @@ mod tests { .tree } + fn contains_merge(node: &NativeNode) -> bool { + match &node.expr { + NativeExpr::Merge { .. } => true, + NativeExpr::Window { input, .. } + | NativeExpr::Map { input, .. } + | NativeExpr::Instant { input, .. } + | NativeExpr::Timestamp { input } + | NativeExpr::VectorOf { input } + | NativeExpr::Aggregate { input, .. } + | NativeExpr::Vector { input, .. } => contains_merge(input), + NativeExpr::Binary { left, right, .. } | NativeExpr::Set { left, right, .. } => { + contains_merge(left) || contains_merge(right) + } + NativeExpr::Info { base, info, .. } => contains_merge(base) || contains_merge(info), + NativeExpr::ParameterizedMap { + input, parameters, .. + } => contains_merge(input) || parameters.iter().any(contains_merge), + NativeExpr::Input(_) | NativeExpr::Scalar(_) | NativeExpr::EvalTime { .. } => false, + } + } + fn input_sources(root: &NativeNode) -> Vec { fn collect(node: &NativeNode, sources: &mut Vec) { match &node.expr { @@ -1480,6 +1523,20 @@ mod tests { .collect() } + #[test] + fn planner_boundary_keeps_window_over_raw_input_unmerged() { + // This is the multi-input planner route's conservative boundary: one unknown property + // per selector occurrence. A range selector compiles directly to Window(Input), but the + // pass cannot wrap raw samples in Merge: merges only bind evaluated vectors. + let tree = compiled_test_tree("rate(some_metric[5m]) + other_metric"); + let boundary = vec![NativeProperties::not_known(); 2]; + let rewritten = insert_native_merge_inside_current_fused(&tree, &boundary).unwrap(); + assert!( + !contains_merge(rewritten.root()), + "the raw Input boundary unexpectedly received a merge:\n{rewritten}" + ); + } + #[test] fn selector_walk_keeps_plain_selector_identity() { for (query, expected) in [ diff --git a/src/query/src/promql/planner/test.rs b/src/query/src/promql/planner/test.rs index a54e4d4bf44..8e654b9eedc 100644 --- a/src/query/src/promql/planner/test.rs +++ b/src/query/src/promql/planner/test.rs @@ -1998,7 +1998,8 @@ async fn native_fused_rate_plans_to_a_fused_series_node() { ); // The scan below is chosen during physical planning, so the route cannot know whether it is // the legacy scan (`KeyThenTime`) or the two-phase one (`Contiguous`) and must bind the weaker - // guarantee both provide. `rate` compiles to one window over that raw input. + // guarantee both provide. `rate` compiles to one window over that raw input. The canonical + // merge planner test uses an evaluated subquery window. let NativeExpr::Window { input, .. } = &fused.tree().root().expr else { panic!("the root is not a window: {:?}", fused.tree().root().expr); };