From f2f8043fdd4b158a63053fc097607da89eb1be4c Mon Sep 17 00:00:00 2001 From: discord9 Date: Tue, 20 Oct 2026 18:58:07 +0800 Subject: [PATCH] feat(promql): add the bound native tree and the metric surface it reports through (#9723) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(promql): add the bound tree and batch protocol for native execution A native PromQL evaluation needs a compiled form whose operations are already bound to kernels, fields and grids. Recognising an opportunity from a finished physical plan cannot provide that: the opportunity then depends on how the optimizer happened to arrange a subtree, and every new shape widens a fixed template. `NativeExpr` instead covers input, scalar, instant, window, map, aggregate and binary nodes, each carrying a bound operation and a declared output type. Legality becomes a question about types and data dependencies, which `NativeNode::validate` answers, so a new combination of supported operations is an ordinary tree rather than another matcher. Two properties are checked there rather than assumed: a window's declared input grid must be the grid its input produces, so a subquery's inner grid is consumed as it is instead of being resampled, and an aggregate's output type must match its stage, so intermediate state cannot pass for a query value. `EvaluatedChunk` is the batch a node hands to the next one: an ordinary Rust value, addressed on absolute grid positions, closing a series exactly once. An empty final chunk is still meaningful — it is how a closed prefix is reported, which a consumer needs in order to tell "nothing more here" from "no more input". Nodes are identified by their position in the tree rather than by an id. The position is derived by the one traversal both `DisplayAs` and run-time construction use, so what is shown and what is registered cannot disagree. It stays in the metric name because the collector aggregates by name first, which would otherwise merge two nodes of the same operation. This adds the tree and the batch protocol only. The evaluator's chunked interface, memory reservation and execution are not part of it. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): make the bound tree's type contracts match what each stage consumes Four review findings on the bound tree, all about checks that passed the wrong things. An aggregate's stage decided its output but not its input. Every stage first required its input to be evaluated, which aggregate state is not, so `Final(Partial(...))` was rejected and `Final(Window(...))` was accepted. Each stage now states its pair: `Complete` takes a vector and produces one on the same grid, `Partial` takes a vector and produces state on that grid with the bound grouping, `Final` takes state with the bound grouping and produces a vector on the state's grid. Window and instant selections accepted aggregate state. The cause was asking `is_evaluated()`, which is false both for raw samples and for state, since state has no grid: `Instant(Partial(...))` and a window bound to `input_grid: None` over `Partial(...)` both passed and would have read state as raw samples. Both now match on `NativeType::Samples` first and only then on the grid, so state is rejected by what it is rather than by lacking a grid. Scalars were modelled as a grid-less evaluated sample, and `BoundScalar`'s doc claimed a value varying per grid point is not a scalar. That is wrong for PromQL: `time()` and `scalar(v)` produce a scalar per evaluation time. A scalar is now its own `NativeType::Scalar(grid)`, so a binary operation can tell a broadcast from label matching -- `sum by(job)(rate(x[5m])) + time()` broadcasts rather than matching on the empty label set -- and `Map`, `Aggregate`, `Instant` and `Window` reject it by type. The `time()` kernel itself is not in scope. An input could bind only samples, so an exchange boundary could not bind the state a `Final` aggregate consumes. `BoundInput` holds a `NativeType`, and input nodes check that their output is the same binding rather than the same schema. A peak is the one figure the collector must not add up: it groups metrics by name, clears the partition and sums gauges, so partition peaks of 1024 and 2048 reported 3072, which is neither peak. `PEAK_RESERVED_BYTES` now carries the partition in its name (`native.p0.root.peak_reserved_bytes`), while counts and the other gauges still sum across partitions, where a sum is meaningful. The test that claimed the peak kept its partition only checked a name prefix; it now registers two partitions, runs the real `aggregate_by_name`, and asserts both peaks survive with their own values. The walkthrough's `+1` test was wrong twice over: it took a `Final` aggregate directly over a `Window` -- the case the first finding now rejects -- and applied `MathOp::Abs`, which is not `+1`. It now builds a real binary `+` with a scalar operand. Signed-off-by: discord9 Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * refactor(promql): use the evaluator's chunk type for native batches `native/batch.rs` defined its own `EvaluatedChunk` with the same two fields as `promql::evaluated::EvaluatedChunk`. Two definitions of one protocol drift: the evaluator's chunk could not be handed to a native node without translation, and a change to either would have to be mirrored. Use `crate::evaluated::EvaluatedChunk` instead. The type moves here with the evaluator's documentation, which states the same two rules this module did: a grid position is absolute, and `is_last` is the only thing that closes a series. The module keeps its own doc explaining the rules native nodes rely on and re-exports the type, so `native::batch::EvaluatedChunk` still names it. Signed-off-by: discord9 Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): reject a node that declares a different field count than it consumes A window and a map both report one value per input field, so they keep the layout of their input. Validation checked that both sides were samples and that the grids matched, but not that the field counts did, so a tree could declare two output fields over a one-field input and still pass `NativeTree::try_new`. The mismatch then surfaced at run time, against a schema the tree was supposed to have validated. Signed-off-by: discord9 Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): validate the field layout of every node that declares one Field counts were checked for a window and a map and nothing else, so an instant selection, either aggregate stage and a binary operation could each declare a layout their kernel does not produce and still pass validation, surfacing at run time against a schema the tree was supposed to have validated. Each node is now checked against the count its own kernel produces, which for a binary over two vectors is the shorter side. Grids are compared by their specification, as the kernels compare them, instead of by allocation. A compiler that allocates one grid per window -- `rate(a[5m]) + rate(b[5m])` -- built a tree the runtime would run and validation rejected. Binding identity is still compared by allocation: an input has to report the one binding it was given, which is a different question from what two equal grids mean. A rejected tree is a compiler defect, so it is reported as an internal error rather than as a mistake in the client's statement, which `Plan` would have implied. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): check a window's input ordering and range, and drop the tests that only repeat another module `validate` accepted an interleaved input for a window, although the range evaluator reads one series to its last chunk before it starts the next, so the tree this layer builds was guaranteed to fail at run time. A window now rejects both an interleaved input and a range that is not positive. An instant selection reads one series at a time, so it keeps accepting an interleaved input and now says why. The grid comment claimed the input grid is checked to cover the window's read range. It is not: only a subquery feeds an evaluated input to a window, and this compiler does not accept subqueries yet. The binary operation's catch-all arm is unreachable at run time -- state has no grid and the grid checks return before the match -- but the match still has to cover the state pairs, so they are spelled out and say so instead of claiming a value arrived. Rejection branches had no tests, so deleting any of them kept the suite green. Each now has one. The four tests in `native/batch.rs` repeated the ones in `evaluated.rs` through the re-export, so they could not fail on their own; they are gone, and the coverage stays in `evaluated.rs`. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): bind a grouping as `by` or `without`, not as a label list A label list can only spell `by`: `sum without (instance) (rate(x[5m]))` has no list of group labels, because its key is every label the series has except the ones named. Spelling that as a list would group the query by labels it meant to remove, and the kernel takes both shapes already (`aggregate::Grouping::{By, Without}`), so the tree had no way to say what the engine can already do. State carries its grouping in the same terms now, so a `Final` also refuses state built for the other shape, not only state built from a different label list. The `groups` gauge counts per partition: a `Partial` aggregate sees one group in every partition it reaches, so the summed gauge is the total of per-partition entries rather than distinct groups. Both the metric's doc and the comment that claimed otherwise now say so. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: discord9 --- src/promql/src/lib.rs | 1 + src/promql/src/native/batch.rs | 34 + src/promql/src/native/metrics.rs | 609 ++++++++++++++ src/promql/src/native/mod.rs | 32 + src/promql/src/native/plan.rs | 635 ++++++++++++++ src/promql/src/native/plan/tests.rs | 1187 +++++++++++++++++++++++++++ 6 files changed, 2498 insertions(+) create mode 100644 src/promql/src/native/batch.rs create mode 100644 src/promql/src/native/metrics.rs create mode 100644 src/promql/src/native/mod.rs create mode 100644 src/promql/src/native/plan.rs create mode 100644 src/promql/src/native/plan/tests.rs diff --git a/src/promql/src/lib.rs b/src/promql/src/lib.rs index 4ba3a66b96a..c9ee89ff62c 100644 --- a/src/promql/src/lib.rs +++ b/src/promql/src/lib.rs @@ -27,6 +27,7 @@ pub mod map; pub mod map_label; pub mod map_time; mod metrics; +pub mod native; pub mod range_array; pub mod range_eval; pub mod select; diff --git a/src/promql/src/native/batch.rs b/src/promql/src/native/batch.rs new file mode 100644 index 00000000000..f3982a0115e --- /dev/null +++ b/src/promql/src/native/batch.rs @@ -0,0 +1,34 @@ +// 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. + +//! The batches native nodes hand to each other. +//! +//! These are ordinary Rust values passed by call, not DataFusion streams and not encoded messages: +//! the Arrow arrays, buffers and shared schemas inside them are reused as they are. +//! +//! Two rules make a chunk self-describing: +//! +//! - A grid position is absolute on the chunk's grid. Batch-local row numbers never stand in for +//! one, and an inner grid is never addressed with an outer grid's index. +//! - A series closes exactly once, on the chunk that carries [`EvaluatedChunk::is_last`]. A closed +//! prefix still has to be followed by a final chunk even when that chunk holds no values, so a +//! consumer can tell "nothing more here" from "no more input". + +//! +//! The chunk type itself is [`crate::evaluated::EvaluatedChunk`]: one chunk protocol serves both the +//! evaluator that produces chunks and the native nodes that consume them. Defining a second one +//! here would let the two drift apart, and a chunk produced by the evaluator could not be handed to +//! a node without translation. + +pub use crate::evaluated::EvaluatedChunk; diff --git a/src/promql/src/native/metrics.rs b/src/promql/src/native/metrics.rs new file mode 100644 index 00000000000..bcae9ddcdd0 --- /dev/null +++ b/src/promql/src/native/metrics.rs @@ -0,0 +1,609 @@ +// 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. + +//! Where a node sits in the final execution tree, and the metrics named after that position. +//! +//! A node is identified by its structural position — `root`, `.input`, `.left`, `.right` — rather +//! than by an id it carries. The position is not stored on the node: it is derived while walking +//! the tree. [`walk`] is the one traversal rule, and [`register_tree`] is its caller. +//! +//! The position stays in the metric name because the collector aggregates by name first: two +//! `rate` nodes sharing a bare name would be merged into one number. Positions are only an +//! observation detail of one plan — they are not stable across queries, plan rewrites or versions, +//! so nothing may depend on them beyond correlating a metric with a tree it was taken from. + +use std::sync::Arc; + +use datafusion::physical_plan::metrics::MetricValue; +use datafusion::physical_plan::metrics::{ + Count, ExecutionPlanMetricsSet, Gauge, MetricBuilder, Time, +}; + +use crate::native::plan::{NativeExpr, NativeNode}; + +/// The reserved first segment of every native metric name. +const NATIVE_PREFIX: &str = "native"; + +/// A node's position in the final tree. +/// +/// Rendering allocates a short string once, when the run-time state is created; the hot path holds +/// the metric handles instead of looking anything up by name. +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct NodePath(String); + +impl NodePath { + /// The position of the root node. + pub fn root() -> Self { + Self("root".to_string()) + } + + /// The position of a unary node's input. + pub fn input(&self) -> Self { + Self(format!("{}.input", self.0)) + } + + /// The position of a binary node's left input. + pub fn left(&self) -> Self { + Self(format!("{}.left", self.0)) + } + + /// The position of a binary node's right input. + pub fn right(&self) -> Self { + Self(format!("{}.right", self.0)) + } + + /// The path without the metric namespace. + pub fn as_str(&self) -> &str { + &self.0 + } + + /// The full metric name for `metric` at this position. + pub fn metric_name(&self, metric: &str) -> String { + format!("{NATIVE_PREFIX}.{}.{}", self.0, metric) + } + + /// The full metric name for `metric` at this position in `partition`. + /// + /// A figure that must not be merged across partitions needs the partition in its name: the + /// collector aggregates by name first, and a sum of peaks is not a peak. + pub fn partitioned_metric_name(&self, metric: &str, partition: usize) -> String { + format!("{NATIVE_PREFIX}.p{partition}.{}.{metric}", self.0) + } +} + +impl std::fmt::Display for NodePath { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +/// Visits every node of `root` with its position, outermost first. +/// +/// This is the one traversal rule: [`register_tree`] is its caller, so a node's metric is always +/// named after the position this walk gives it. +pub fn walk(root: &NativeNode, visit: &mut dyn FnMut(&NativeNode, &NodePath)) { + fn go(node: &NativeNode, path: &NodePath, visit: &mut dyn FnMut(&NativeNode, &NodePath)) { + visit(node, path); + match &node.expr { + NativeExpr::Input(_) | NativeExpr::Scalar(_) => {} + NativeExpr::Instant { input, .. } + | NativeExpr::Window { input, .. } + | NativeExpr::Map { input, .. } + | NativeExpr::Aggregate { input, .. } => go(input, &path.input(), visit), + NativeExpr::Binary { left, right, .. } => { + go(left, &path.left(), visit); + go(right, &path.right(), visit); + } + } + } + go(root, &NodePath::root(), visit); +} + +/// The metric names one node reports. +/// +/// [`Self::ALL`] is the list a reader should treat as complete: [`NodeMetrics::register`] names each +/// handle itself, and `registered_metrics_are_exactly_the_declared_set` fails when the two sides +/// drift apart. +pub struct NodeMetric; + +impl NodeMetric { + /// How many batches this node consumed. + pub const INPUT_BATCHES: &'static str = "input_batches"; + /// How many batches this node produced. + pub const OUTPUT_BATCHES: &'static str = "output_batches"; + /// How many sample positions this node ingested. + pub const INPUT_SAMPLES: &'static str = "input_samples"; + /// How many grid positions this node evaluated, including those with no valid result. + pub const EVALUATED_GRID_POINTS: &'static str = "evaluated_grid_points"; + /// How many valid values this node produced. + pub const OUTPUT_VALUES: &'static str = "output_values"; + /// How many series this node closed. + pub const COMPLETED_SERIES: &'static str = "completed_series"; + /// Time this node spent in its own work, excluding its inputs. + pub const SELF_COMPUTE_TIME_NS: &'static str = "self_compute_time_ns"; + /// Bytes this node currently reserves. + pub const CURRENT_RESERVED_BYTES: &'static str = "current_reserved_bytes"; + /// The highest byte count this node reserved during the execution. + pub const PEAK_RESERVED_BYTES: &'static str = "peak_reserved_bytes"; + /// How many groups this node tracks, counted per partition: a group reached by several + /// partitions is one entry in each. + pub const GROUPS: &'static str = "groups"; + /// How many state entries this node tracks. + pub const STATE_ENTRIES: &'static str = "state_entries"; + + /// Every metric a node may report. + pub const ALL: &'static [&'static str] = &[ + Self::INPUT_BATCHES, + Self::OUTPUT_BATCHES, + Self::INPUT_SAMPLES, + Self::EVALUATED_GRID_POINTS, + Self::OUTPUT_VALUES, + Self::COMPLETED_SERIES, + Self::SELF_COMPUTE_TIME_NS, + Self::CURRENT_RESERVED_BYTES, + Self::PEAK_RESERVED_BYTES, + Self::GROUPS, + Self::STATE_ENTRIES, + ]; +} + +/// The metric handles one node updates as it runs. +/// +/// The node writes these directly; it never looks a metric up by name, and it never formats a path +/// on the hot path. +#[derive(Clone, Debug)] +pub struct NodeMetrics { + input_batches: Count, + output_batches: Count, + input_samples: Count, + evaluated_grid_points: Count, + output_values: Count, + completed_series: Count, + self_compute_time: Time, + current_reserved_bytes: Gauge, + peak_reserved_bytes: Gauge, + groups: Gauge, + state_entries: Gauge, +} + +impl Default for NodeMetrics { + fn default() -> Self { + Self { + input_batches: Count::new(), + output_batches: Count::new(), + input_samples: Count::new(), + evaluated_grid_points: Count::new(), + output_values: Count::new(), + completed_series: Count::new(), + self_compute_time: Time::new(), + current_reserved_bytes: Gauge::new(), + peak_reserved_bytes: Gauge::new(), + groups: Gauge::new(), + state_entries: Gauge::new(), + } + } +} + +impl NodeMetrics { + /// Registers this node's metrics under `path` in `set`. + /// + /// `partition` is part of the registration, so two partitions of one plan report separately. + /// It is also part of the peak's name, because a peak is the one figure a collector must not + /// add up across partitions. + pub fn register(&self, path: &NodePath, set: &ExecutionPlanMetricsSet, partition: usize) { + fn count(set: &ExecutionPlanMetricsSet, partition: usize, name: String, count: &Count) { + MetricBuilder::new(set) + .with_partition(partition) + .build(MetricValue::Count { + name: name.into(), + count: count.clone(), + }); + } + fn gauge(set: &ExecutionPlanMetricsSet, partition: usize, name: String, gauge: &Gauge) { + MetricBuilder::new(set) + .with_partition(partition) + .build(MetricValue::Gauge { + name: name.into(), + gauge: gauge.clone(), + }); + } + + count( + set, + partition, + path.metric_name(NodeMetric::INPUT_BATCHES), + &self.input_batches, + ); + count( + set, + partition, + path.metric_name(NodeMetric::OUTPUT_BATCHES), + &self.output_batches, + ); + count( + set, + partition, + path.metric_name(NodeMetric::INPUT_SAMPLES), + &self.input_samples, + ); + count( + set, + partition, + path.metric_name(NodeMetric::EVALUATED_GRID_POINTS), + &self.evaluated_grid_points, + ); + count( + set, + partition, + path.metric_name(NodeMetric::OUTPUT_VALUES), + &self.output_values, + ); + count( + set, + partition, + path.metric_name(NodeMetric::COMPLETED_SERIES), + &self.completed_series, + ); + gauge( + set, + partition, + path.metric_name(NodeMetric::CURRENT_RESERVED_BYTES), + &self.current_reserved_bytes, + ); + // The peak is the one figure named per partition. The collector groups by name and sums + // gauges, and a sum of two peaks is neither peak: partitions peak at different moments. + // Summing the other gauges and counts stays meaningful — the memory held now and the + // samples seen add up across partitions. `groups` does not: a `Partial` aggregate sees the + // same group in every partition it reaches, so the sum is per-partition entries, not + // distinct groups. + gauge( + set, + partition, + path.partitioned_metric_name(NodeMetric::PEAK_RESERVED_BYTES, partition), + &self.peak_reserved_bytes, + ); + gauge( + set, + partition, + path.metric_name(NodeMetric::GROUPS), + &self.groups, + ); + gauge( + set, + partition, + path.metric_name(NodeMetric::STATE_ENTRIES), + &self.state_entries, + ); + MetricBuilder::new(set) + .with_partition(partition) + .build(MetricValue::Time { + name: path.metric_name(NodeMetric::SELF_COMPUTE_TIME_NS).into(), + time: self.self_compute_time.clone(), + }); + } + + /// Records that this node ingested `batches` batches carrying `samples` sample positions. + pub fn record_input(&self, batches: usize, samples: usize) { + self.input_batches.add(batches); + self.input_samples.add(samples); + } + + /// Records that this node produced `batches` batches carrying `values` valid values. + pub fn record_output(&self, batches: usize, values: usize) { + self.output_batches.add(batches); + self.output_values.add(values); + } + + /// Records that this node evaluated `points` grid positions. + pub fn record_grid_points(&self, points: usize) { + self.evaluated_grid_points.add(points); + } + + /// Records that this node closed `series` series. + pub fn record_completed_series(&self, series: usize) { + self.completed_series.add(series); + } + + /// Records that this node now reserves `bytes`. + /// + /// The peak is a high-water mark of this node in this partition, never a sum of the peaks of + /// different nodes: those peaks can happen at different times. + pub fn record_reserved_bytes(&self, bytes: usize) { + self.current_reserved_bytes.set(bytes); + // A high-water mark, not a sum: two nodes peak at different moments. + self.peak_reserved_bytes.set_max(bytes); + } + + /// Records the size of the grouping and state this node holds. + pub fn record_state(&self, groups: usize, entries: usize) { + self.groups.set(groups); + self.state_entries.set(entries); + } + + /// Starts timing this node's own work. + /// + /// The returned timer must be dropped before the node polls its input: a node's self time + /// excludes the work its inputs do. + pub fn timer(&self) -> datafusion::physical_plan::metrics::ScopedTimerGuard<'_> { + self.self_compute_time.timer() + } +} + +/// Registers every node of `root` in `set` and returns the handles in traversal order. +pub fn register_tree( + root: &NativeNode, + set: &ExecutionPlanMetricsSet, + partition: usize, +) -> Vec<(NodePath, Arc)> { + let mut out = Vec::new(); + walk(root, &mut |_, path| { + let metrics = Arc::new(NodeMetrics::default()); + metrics.register(path, set, partition); + out.push((path.clone(), metrics)); + }); + out +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricValue}; + use datatypes::series::samples::SeriesSchema; + use datatypes::series::time::{EvalGrid, GridSpec}; + + use super::{NodeMetric, NodePath, register_tree, walk}; + use crate::map::MathOp; + use crate::native::plan::{ + BoundInput, BoundMap, BoundWindow, NativeExpr, NativeNode, NativeType, + }; + use crate::range_eval::{RangeFunction, RangeWindow}; + + fn grid() -> Arc { + Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 60_000, 4).unwrap()).unwrap()) + } + + fn schema() -> Arc { + Arc::new(SeriesSchema { + fields: Arc::new([]), + eval_grid: None, + }) + } + + fn input() -> NativeNode { + let output = NativeType::Samples(schema()); + NativeNode { + expr: NativeExpr::Input(BoundInput { + output: output.clone(), + ordering: datatypes::series::SeriesOrdering::KeyThenTime, + }), + output, + } + } + + fn window(inner: NativeNode) -> NativeNode { + let grid = grid(); + NativeNode { + expr: NativeExpr::Window { + input: Box::new(inner), + op: BoundWindow { + function: RangeFunction::Rate, + window: RangeWindow { + range_ms: 300_000, + offset_ms: 0, + anchor_ms: None, + }, + input_grid: None, + output_grid: Arc::clone(&grid), + }, + }, + output: NativeType::Samples(Arc::new(SeriesSchema { + fields: Arc::new([]), + eval_grid: Some(grid), + })), + } + } + + fn map(inner: NativeNode) -> NativeNode { + let grid = inner.output.grid().expect("a map needs a grid").clone(); + NativeNode { + expr: NativeExpr::Map { + input: Box::new(inner), + op: BoundMap { + ops: Arc::new([MathOp::Abs]), + }, + }, + output: NativeType::Samples(Arc::new(SeriesSchema { + fields: Arc::new([]), + eval_grid: Some(grid), + })), + } + } + + /// A unary chain is addressed by `.input`, so a metric can be attributed to one node. + #[test] + fn a_unary_chain_is_addressed_by_input_steps() { + let root = map(window(input())); + let mut paths = Vec::new(); + walk(&root, &mut |_, path| paths.push(path.to_string())); + assert_eq!(paths, vec!["root", "root.input", "root.input.input"]); + } + + /// A binary node addresses its two sides by name, so the two `rate` subtrees stay distinct. + #[test] + fn a_binary_node_addresses_both_sides() { + let binary = crate::native::plan::NativeNode { + expr: NativeExpr::Binary { + left: Box::new(window(input())), + right: Box::new(map(window(input()))), + matching: crate::vector_match::BinaryMatching { + op: crate::vector_match::BinaryOp::Add, + return_bool: false, + labels: crate::vector_match::MatchingLabels::Ignoring(Vec::new()), + cardinality: crate::vector_match::Cardinality::OneToOne, + }, + }, + output: { + let grid = grid(); + NativeType::Samples(Arc::new(SeriesSchema { + fields: Arc::new([]), + eval_grid: Some(grid), + })) + }, + }; + let mut paths = Vec::new(); + walk(&binary, &mut |_, path| paths.push(path.to_string())); + assert_eq!( + paths, + vec![ + "root", + "root.left", + "root.left.input", + "root.right", + "root.right.input", + "root.right.input.input", + ] + ); + } + + /// The name carries the position because the collector aggregates by name; a bare name would + /// merge two nodes of the same operation into one number. + #[test] + fn a_metric_name_carries_its_position() { + let path = NodePath::root().left().input(); + assert_eq!(path.as_str(), "root.left.input"); + assert_eq!( + path.metric_name(NodeMetric::SELF_COMPUTE_TIME_NS), + "native.root.left.input.self_compute_time_ns" + ); + assert_eq!( + path.partitioned_metric_name(NodeMetric::PEAK_RESERVED_BYTES, 1), + "native.p1.root.left.input.peak_reserved_bytes" + ); + } + + /// The peak keeps the partition in its name, so the collector's aggregation by name cannot add + /// two partitions' peaks: 1024 and 2048 peak at different moments, and 3072 is neither peak. + #[test] + fn a_peak_is_not_merged_across_partitions() { + let root = map(window(input())); + let set = ExecutionPlanMetricsSet::new(); + let peaks = [(0, 1024), (1, 2048)]; + for (partition, peak) in peaks { + let registered = register_tree(&root, &set, partition); + // The root comes first in traversal order. + registered[0].1.record_reserved_bytes(peak); + } + + // The real collector path: `aggregate_by_name` groups by metric name and sums the + // values in each group. + let aggregated = set.clone_inner().aggregate_by_name(); + let peak = |partition: usize| -> Option { + let expected = NodePath::root() + .partitioned_metric_name(NodeMetric::PEAK_RESERVED_BYTES, partition); + aggregated.iter().find_map(|metric| match metric.value() { + MetricValue::Gauge { name, gauge } if name.as_ref() == expected.as_str() => { + Some(gauge.value()) + } + _ => None, + }) + }; + // Two separate names, each holding its own partition's peak: had the name not carried the + // partition, aggregation would have merged them into one gauge summing to 3072. + assert_eq!(peak(0), Some(1024)); + assert_eq!(peak(1), Some(2048)); + } + + /// Registration covers every node, so no node is executed without a way to observe it. + /// + /// The set is read back rather than named again from the tree, so this checks what + /// registration actually did. + #[test] + fn registration_covers_every_node() { + let root = map(window(input())); + let set = datafusion::physical_plan::metrics::ExecutionPlanMetricsSet::new(); + let registered = register_tree(&root, &set, 2); + assert_eq!(registered.len(), 3); + + // Read the names back from the set registration wrote to, so the check is against what + // registration did rather than against the names built a second time here. + let set = set.clone_inner(); + let names: Vec<&str> = set.iter().map(|metric| metric.value().name()).collect(); + // One entry per node per declared metric, and every node's position among them: no node + // is executed without a way to observe it. + assert_eq!(names.len(), registered.len() * NodeMetric::ALL.len()); + for (path, _) in ®istered { + // The peak is named per partition, so a plain name stands for the node here. + let name = path.metric_name(NodeMetric::OUTPUT_VALUES); + assert!( + names.contains(&name.as_str()), + "{path} registered nothing under its own position" + ); + } + } + + /// Every metric a node reports is listed in one place, so registration cannot drift from the + /// documented set. + #[test] + fn every_metric_is_declared() { + assert_eq!(NodeMetric::ALL.len(), 11); + assert!(NodeMetric::ALL.contains(&NodeMetric::SELF_COMPUTE_TIME_NS)); + assert!(NodeMetric::ALL.contains(&NodeMetric::PEAK_RESERVED_BYTES)); + } + + /// Registration reports exactly the metrics `ALL` declares, in both directions. + /// + /// The names are read back from the set a node registered in, so the check is against what + /// registration actually did, not against `ALL` itself. Adding a metric to `ALL` without + /// registering it in `register`, or registering a metric that `ALL` does not list, fails this + /// test: it compares the two sides instead of trusting either. + #[test] + fn registered_metrics_are_exactly_the_declared_set() { + let root = map(window(input())); + let set = ExecutionPlanMetricsSet::new(); + register_tree(&root, &set, 0); + + // One node reports each metric once; the node paths and the partition prefix separate + // nodes, so the bare name is the last dot-separated segment. + let set = set.clone_inner(); + let mut registered: Vec<&str> = set + .iter() + .map(|metric| { + metric + .value() + .name() + .rsplit('.') + .next() + .expect("a metric name has a last segment") + }) + .collect(); + registered.sort_unstable(); + registered.dedup(); + + let mut declared: Vec<&str> = NodeMetric::ALL.to_vec(); + declared.sort_unstable(); + + assert_eq!(registered, declared); + } + + /// A peak is a high-water mark of one node, so it survives a later drop in reservations. + #[test] + fn a_peak_is_a_high_water_mark() { + let metrics = super::NodeMetrics::default(); + metrics.record_reserved_bytes(1024); + metrics.record_reserved_bytes(64); + assert_eq!(metrics.peak_reserved_bytes.value(), 1024); + assert_eq!(metrics.current_reserved_bytes.value(), 64); + } +} diff --git a/src/promql/src/native/mod.rs b/src/promql/src/native/mod.rs new file mode 100644 index 00000000000..89926c77d4b --- /dev/null +++ b/src/promql/src/native/mod.rs @@ -0,0 +1,32 @@ +// 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. + +//! The native PromQL evaluation model. +//! +//! A query compiles to a [`plan::NativeTree`]: nodes whose operations, fields and grids are already +//! bound. Nodes hand each other [`batch::EvaluatedChunk`]s as ordinary Rust values, so no row batch +//! is materialised between them. [`metrics`] names each node after its position in that tree. +//! +//! Three properties keep this apart from recognising a query shape in a finished physical plan: +//! +//! - Legality is a type and data-dependency question ([`plan::NativeNode::validate`]), so a new +//! combination of supported operations is an ordinary tree, not another matcher. +//! - A window binds its input and output grids separately, so a subquery's inner grid is consumed +//! as it is instead of being resampled onto the outer grid. +//! - A series closes exactly once, on the chunk that says so, so a consumer can tell "nothing more +//! here" from "no more input". + +pub mod batch; +pub mod metrics; +pub mod plan; diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs new file mode 100644 index 00000000000..6cca42134f3 --- /dev/null +++ b/src/promql/src/native/plan.rs @@ -0,0 +1,635 @@ +// 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. + +//! The bound tree a native PromQL evaluation compiles to. +//! +//! A query is compiled into [`NativeTree`]: [`NativeNode`]s whose operations are already bound to +//! kernels, fields and grids. Nothing in this module interprets a function name, resolves a label +//! or picks a column at run time; those decisions are made before the tree exists, so the hot +//! path only feeds batches to kernels it was handed. +//! +//! The tree is also the unit of legality. [`NativeNode::validate`] checks the contract every +//! 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: [`NativeExpr::Binary`] holds an +//! arithmetic or comparison operation, so `and`, `or` and `unless` have no node, and +//! [`NativeExpr::Scalar`] holds a constant, so `time()` and `scalar(v)` have no node either. The +//! legality below is therefore about the operations that exist, not about PromQL's whole surface. + +use std::sync::Arc; + +use datafusion::error::{DataFusionError, Result}; +use datatypes::series::SeriesOrdering; +use datatypes::series::samples::{FieldSpec, SeriesSchema}; +use datatypes::series::time::EvalGrid; + +use crate::map::MathOp; +use crate::range_eval::{RangeFunction, RangeWindow}; +use crate::vector_match::BinaryMatching; + +/// What one node produces. +/// +/// A sample type with an evaluation grid is an evaluated series: one value per grid point. Without +/// a grid it is raw samples carrying their own timestamps. A scalar is one value per grid point +/// and no labels. Intermediate aggregate state is a different type altogether, so an operation +/// that cannot consume it is rejected by type rather than by convention. +#[derive(Clone, Debug)] +pub enum NativeType { + /// Samples, evaluated when the schema carries a grid and raw otherwise. + Samples(Arc), + /// One value per grid point and no labels, which binary operations broadcast. + /// + /// PromQL scalars vary with the evaluation time (`time()`, `scalar(v)`), so this carries the + /// grid it is evaluated on. Its value carrier is a series without labels, but its type is + /// what keeps a broadcast from becoming label matching. + Scalar(Arc), + /// Intermediate aggregate state, not a query value. + AggregateState(Arc), +} + +impl NativeType { + /// The schema of a sample type, or `None` for a scalar or aggregate state. + pub fn samples(&self) -> Option<&Arc> { + match self { + Self::Samples(schema) => Some(schema), + Self::Scalar(_) | Self::AggregateState(_) => None, + } + } + + /// The grid this type is evaluated on, or `None` when it holds raw samples or state. + /// + /// A scalar is evaluated on its grid like a vector, which is what lets a binary operation + /// line it up with a vector point by point; only raw samples and state have no positions. + pub fn grid(&self) -> Option<&Arc> { + match self { + Self::Samples(schema) => schema.eval_grid.as_ref(), + Self::Scalar(grid) => Some(grid), + Self::AggregateState(_) => None, + } + } + + /// What this type is, for a rejection that has to say what a consumer was handed. + fn describe(&self) -> &'static str { + match self { + Self::Samples(schema) if schema.eval_grid.is_some() => "an evaluated vector", + Self::Samples(_) => "raw samples", + Self::Scalar(_) => "a scalar", + Self::AggregateState(_) => "aggregate state", + } + } + + /// Whether `self` binds the same thing as `other`: the same samples or the same state. + /// + /// Bindings are compared by allocation identity, because a compiler that meant one schema or + /// one state has no reason to allocate it twice. A scalar is produced by a node and never + /// handed to one by a source, so it never matches. This is about the binding itself: grids are + /// aligned by [`EvalGrid::spec`] everywhere else, here and in the kernels. + fn same_binding(&self, other: &NativeType) -> bool { + match (self, other) { + (Self::Samples(mine), Self::Samples(theirs)) => Arc::ptr_eq(mine, theirs), + (Self::AggregateState(mine), Self::AggregateState(theirs)) => Arc::ptr_eq(mine, theirs), + _ => false, + } + } +} + +/// The layout of intermediate aggregate state. +/// +/// State is indexed by `(group labels, absolute grid index)`, so it carries the grid it was +/// accumulated on. It is not a query value and does not pretend to be one. +#[derive(Clone, Debug)] +pub struct AggregateStateSchema { + /// How the groups its rows are indexed by are identified. + pub group: Grouping, + /// The accumulator columns, one per aggregated field. + pub accumulators: Arc<[FieldSpec]>, + /// The grid the state is indexed on. + pub grid: Arc, +} + +/// A scan or exchange boundary bound as a node input. +/// +/// The binding records what the input was created for: the output a consumer type-checks against, +/// and the ordering it guarantees. A window reads the ordering, rejecting an interleaved input; no +/// other consumer reads it yet. +#[derive(Clone, Debug)] +pub struct BoundInput { + /// What the input produces: raw samples from a scan, or state from an exchange boundary. + pub output: NativeType, + /// The ordering the input guarantees. + pub ordering: SeriesOrdering, +} + +/// A constant scalar bound for broadcast. +/// +/// 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. This binding +/// only holds the constant case. +#[derive(Clone, Debug)] +pub struct BoundScalar { + /// The constant value. + pub value: f64, +} + +/// An instant selection: the last sample at or before each grid point, within a lookback. +#[derive(Clone, Debug)] +pub struct BoundInstant { + /// How far back a sample may lie before it counts as stale. + pub lookback_ms: i64, + /// Selector offset; positive reads earlier samples. + pub offset_ms: i64, + /// The grid the selection produces values on. + pub output_grid: Arc, +} + +/// A window function over raw samples or over an evaluated input. +/// +/// The input and output grids are bound separately: a subquery evaluates its inner expression on +/// its own grid and consumes the result directly, without resampling it onto the outer grid. +#[derive(Clone, Debug)] +pub struct BoundWindow { + /// The function this node runs. + pub function: RangeFunction, + /// The window length, offset and anchor. + pub window: RangeWindow, + /// The grid the input is evaluated on, or `None` when the input is raw samples. + pub input_grid: Option>, + /// The grid this node produces values on. + pub output_grid: Arc, +} + +/// A chain of point-wise operations applied to a whole evaluated value. +#[derive(Clone, Debug)] +pub struct BoundMap { + /// The operations, applied in order. + pub ops: Arc<[MathOp]>, +} + +/// Which aggregate a node computes. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum AggregateKind { + /// `sum` over the values of each group at each grid point. + Sum, +} + +/// Where an aggregate sits in the execution, which decides whether its output is a query value. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum AggregateStage { + /// Every contribution to each group lies within this execution, so the output is complete. + Complete, + /// Only this execution's contributions were seen, so the output is state for a later merge. + Partial, + /// The state was collected from every contributing execution, so the output is complete. + Final, +} + +/// How an aggregate's groups are identified, in the binding's own terms. +/// +/// The kernel already takes both shapes ([`aggregate::Grouping`]), and the two are not +/// interchangeable: `by(job)` keeps `job` and drops everything else, while `without(job)` keeps +/// every other label and drops `job` too. A list of group labels can only spell the first, so a +/// `without` written that way would group a query by labels it meant to remove. +/// +/// [`aggregate::Grouping`]: crate::aggregate::Grouping +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Grouping { + /// The labels a group is identified by; every other label is dropped. + By(Arc<[Arc]>), + /// The labels a group is not identified by; these are dropped, and so is `__name__`. + Without(Arc<[Arc]>), +} + +/// An aggregate bound to its kind, grouping and stage. +#[derive(Clone, Debug)] +pub struct BoundAggregate { + /// Which aggregate to compute. + pub kind: AggregateKind, + /// How its groups are identified. + pub group: Grouping, + /// Where this aggregate sits in the execution. + pub stage: AggregateStage, +} + +/// What a node computes. Every variant holds a bound operation, never a name to resolve. +#[derive(Clone, Debug)] +pub enum NativeExpr { + /// A scan or an exchange boundary. + Input(BoundInput), + /// A constant scalar, evaluated on the statement's grid. + Scalar(BoundScalar), + /// An instant selection over its input. + Instant { + input: Box, + op: BoundInstant, + }, + /// A window function over its input. + Window { + input: Box, + op: BoundWindow, + }, + /// Point-wise operations over its input. + Map { + input: Box, + op: BoundMap, + }, + /// An aggregate over its input. + Aggregate { + input: Box, + op: BoundAggregate, + }, + /// A binary operation between two inputs, with its matching rules. + /// + /// Scalar operands are broadcast, so `matching` only applies to vector–vector operations. + Binary { + left: Box, + right: Box, + matching: BinaryMatching, + }, +} + +/// One node of a bound tree. +#[derive(Clone, Debug)] +pub struct NativeNode { + /// What this node computes, including its bound operands. + pub expr: NativeExpr, + /// What this node produces. + pub output: NativeType, +} + +/// A bound evaluation tree. +/// +/// The tree itself carries no run-time state: every execution builds its own, so one compiled plan +/// can serve every partition. +#[derive(Clone, Debug)] +pub struct NativeTree { + root: NativeNode, +} + +impl NativeTree { + /// Creates a tree over `root`, checking that every node's contract holds. + /// + /// A rejected tree is a compiler defect rather than a query error: the compiler only builds a + /// node after deciding that this combination is supported, so the error is + /// [`DataFusionError::Internal`] and not a statement the client wrote wrong. + pub fn try_new(root: NativeNode) -> Result { + root.validate()?; + Ok(Self { root }) + } + + /// The root node. + pub fn root(&self) -> &NativeNode { + &self.root + } +} + +impl NativeNode { + /// Checks this node and its inputs against the contract of its operation. + pub fn validate(&self) -> Result<()> { + match &self.expr { + NativeExpr::Input(input) => { + // An input node's output is exactly the binding it was bound to: a scan hands over + // the samples it was created for, and an exchange boundary the state. + if !self.output.same_binding(&input.output) { + return Err(invalid( + "an input node's output is not the binding it was bound to", + )); + } + } + NativeExpr::Scalar(_) => { + // 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::Instant { input, op } => { + input.validate()?; + // An instant selection reads raw samples and stamps them onto its own grid: an + // evaluated vector is already stamped, and a scalar has no timestamps at all. + match &input.output { + NativeType::Samples(schema) if schema.eval_grid.is_none() => {} + NativeType::Samples(_) => { + return Err(invalid("an instant selection reads evaluated values")); + } + other => { + return Err(invalid(&format!( + "an instant selection consumes raw samples, not {}", + other.describe() + ))); + } + } + ensure_samples(&self.output, "an instant selection")?; + // A selection is one source row per grid point: `select_instants` picks row + // indices and every field is taken at them, so it carries one value per input + // field. + ensure_same_fields(&input.output, &self.output, "an instant selection")?; + ensure_grid(&self.output, &op.output_grid, "an instant selection")?; + // `select_instants` reads one series at a time and holds no cross-series state, so + // an interleaved input is enough for it: no ordering check here. + } + NativeExpr::Window { input, op } => { + input.validate()?; + // A window reads a range of samples per grid point, so it consumes samples: state + // is not a sample stream and a scalar has no timestamps. + let NativeType::Samples(schema) = &input.output else { + return Err(invalid(&format!( + "a window consumes samples, not {}", + input.output.describe() + ))); + }; + // A range that is not positive reads nothing at every grid point. + if op.window.range_ms <= 0 { + return Err(invalid( + "a window reads an empty range: its range must be positive", + )); + } + // A range evaluator consumes one series to its last chunk before it starts the + // next, so interleaved chunks of different series cannot be fed to it. Only an + // input node carries an ordering, so only there is there one to check. + if let NativeExpr::Input(binding) = &input.expr + && binding.ordering == SeriesOrdering::Interleaved + { + return Err(invalid( + "a window reads one series to its last chunk before the next, so \ + interleaved chunks cannot be fed to it", + )); + } + // What is checked here and just below is that the declared input grid equals the + // input's grid, and that the output grid matches the declared one. The grid is + // not checked to cover the window's read range + // `(out.start - offset - range, out.end - offset]`: only a subquery feeds an + // evaluated input to a window, and this compiler does not accept subqueries yet, + // so that coverage check waits for it. + match (op.input_grid.as_ref(), schema.eval_grid.as_ref()) { + (None, None) => {} + (Some(bound), Some(actual)) if bound.spec() == actual.spec() => {} + _ => { + return Err(invalid( + "a window's declared input grid is not the input's grid", + )); + } + } + ensure_samples(&self.output, "a window")?; + ensure_same_fields(&input.output, &self.output, "a window")?; + ensure_grid(&self.output, &op.output_grid, "a window")?; + } + NativeExpr::Map { input, op } => { + input.validate()?; + // A map is point-wise over an evaluated vector: raw samples have no shared + // positions to be point-wise over, and a scalar is not a vector. + let Some((_, input_grid)) = evaluated_vector(&input.output) else { + return Err(invalid(&format!( + "a map consumes an evaluated vector, not {}", + input.output.describe() + ))); + }; + if op.ops.is_empty() { + return Err(invalid("a map holds no operation")); + } + ensure_samples(&self.output, "a map")?; + ensure_same_fields(&input.output, &self.output, "a map")?; + ensure_grid(&self.output, input_grid, "a map")?; + } + NativeExpr::Aggregate { input, op } => { + input.validate()?; + // A stage's contract is a pair of types, not only what it produces: `Partial` + // turns a vector into state and `Final` turns state back into a vector, so state + // never sits where a value is expected and a value never sits where state is. + match op.stage { + AggregateStage::Complete => { + // Every contribution lies within this execution: vector in, vector out, + // both on the input's grid, one value per input field. + let Some((_, input_grid)) = evaluated_vector(&input.output) else { + return Err(invalid(&format!( + "a complete aggregate consumes an evaluated vector, not {}", + input.output.describe() + ))); + }; + ensure_samples(&self.output, "a complete aggregate")?; + ensure_same_fields(&input.output, &self.output, "a complete aggregate")?; + ensure_grid(&self.output, input_grid, "a complete aggregate")?; + } + AggregateStage::Partial => { + // Only this execution's contributions were seen, so the output is state + // for a later merge: on the input's grid, grouped the same way, one + // accumulator per input field. + let Some((input_schema, input_grid)) = evaluated_vector(&input.output) + else { + return Err(invalid(&format!( + "a partial aggregate consumes an evaluated vector, not {}", + input.output.describe() + ))); + }; + let NativeType::AggregateState(state) = &self.output else { + return Err(invalid(&format!( + "a partial aggregate produces aggregate state, not {}", + self.output.describe() + ))); + }; + ensure_field_count( + state.accumulators.len(), + input_schema.fields.len(), + "a partial aggregate", + )?; + if state.grid.spec() != input_grid.spec() { + return Err(invalid( + "aggregate state is indexed on a grid the input does not produce", + )); + } + if state.group != op.group { + return Err(invalid( + "aggregate state groups differ from the binding: state built for \ + one `by` or `without` cannot be merged into another", + )); + } + } + AggregateStage::Final => { + // The state was collected from every contributing execution, so the + // output is the complete value on the grid that state is indexed on. + let NativeType::AggregateState(state) = &input.output else { + return Err(invalid(&format!( + "a final aggregate consumes aggregate state, not {}", + input.output.describe() + ))); + }; + if state.group != op.group { + return Err(invalid( + "aggregate state groups differ from the binding: state built for \ + one `by` or `without` cannot be merged into another", + )); + } + let output = ensure_samples(&self.output, "a final aggregate")?; + // The merged value carries one value per accumulator the state holds. + ensure_field_count( + output.fields.len(), + state.accumulators.len(), + "a final aggregate", + )?; + ensure_grid(&self.output, &state.grid, "a final aggregate")?; + } + } + } + NativeExpr::Binary { left, right, .. } => { + left.validate()?; + right.validate()?; + // A binary operation is point-wise, so both sides are evaluated on the statement's + // grid: state has no positions at all, and raw samples carry their own timestamps + // instead of positions on a shared grid. + let Some(left_grid) = left.output.grid() else { + return Err(invalid(&format!( + "a binary operation consumes evaluated values, not {} on its left", + left.output.describe() + ))); + }; + let Some(right_grid) = right.output.grid() else { + return Err(invalid(&format!( + "a binary operation consumes evaluated values, not {} on its right", + right.output.describe() + ))); + }; + if left_grid.spec() != right_grid.spec() { + return Err(invalid("a binary operation aligns two different grids")); + } + match (&left.output, &right.output) { + // Two scalars make a scalar: there is nothing to match and nothing to keep. + (NativeType::Scalar(_), NativeType::Scalar(_)) => { + if !matches!(&self.output, NativeType::Scalar(grid) if grid.spec() == left_grid.spec()) + { + return Err(invalid( + "two scalars do not produce a scalar on the statement's grid", + )); + } + } + // A scalar side carries no labels and no fields, so the operation is a + // broadcast on the statement's grid and `matching` is not consulted. A scalar + // has no fields of its own, so the broadcast keeps the vector side's fields. + (NativeType::Scalar(_), NativeType::Samples(vector)) + | (NativeType::Samples(vector), NativeType::Scalar(_)) => { + let Some(output) = self.output.samples() else { + return Err(invalid(&format!( + "a scalar broadcast produces the vector side's samples, not {}", + self.output.describe() + ))); + }; + ensure_grid(&self.output, left_grid, "a scalar broadcast")?; + if output.fields.len() != vector.fields.len() { + return Err(invalid( + "a scalar broadcast does not produce the vector side's fields", + )); + } + } + // Two vectors are matched by `matching` on the grid they share, and the match + // pairs fields by position, so it keeps the shorter side's fields. + (NativeType::Samples(left_vector), NativeType::Samples(right_vector)) => { + let Some(output) = self.output.samples() else { + return Err(invalid(&format!( + "a binary operation over two vectors produces samples, not {}", + self.output.describe() + ))); + }; + ensure_field_count( + output.fields.len(), + left_vector.fields.len().min(right_vector.fields.len()), + "a binary operation over two vectors", + )?; + ensure_grid(&self.output, left_grid, "a binary operation")?; + } + // State has no grid, so the grid checks above return for either side before + // this match and these two arms cannot be reached at run time. They are + // spelled out because `NativeType` names `AggregateState` and the match has to + // cover the pair; the message says so rather than pretending a value arrived. + (NativeType::AggregateState(_), _) | (_, NativeType::AggregateState(_)) => { + return Err(invalid( + "a binary operation reached aggregate state, which has no grid: the \ + grid check above should have rejected it", + )); + } + } + } + } + Ok(()) + } +} + +/// The schema and grid of an evaluated vector, or `None` when `ty` is raw samples, a scalar or +/// state. +/// +/// An operation over values needs both halves: that its input is samples at all, which rules out +/// a scalar, and that those samples are stamped onto a grid, which rules out raw samples. The +/// schema comes back with the grid because a consumer that needs the grid also declares a field +/// layout for the fields the schema carries. +fn evaluated_vector(ty: &NativeType) -> Option<(&Arc, &Arc)> { + let schema = ty.samples().filter(|schema| schema.eval_grid.is_some())?; + Some((schema, schema.eval_grid.as_ref()?)) +} + +/// Checks that a node declares the field layout its input produces. +/// +/// A window, a map, an instant selection and a complete aggregate all report one value per input +/// field, so they keep the layout of their input. A node that declares a different number of +/// fields than it consumes describes a computation this model has no meaning for, and letting it +/// through would only surface at run time as a field count mismatch against a schema the tree was +/// supposed to have validated. +fn ensure_same_fields(input: &NativeType, output: &NativeType, what: &str) -> Result<()> { + let (NativeType::Samples(input), NativeType::Samples(output)) = (input, output) else { + return Ok(()); + }; + ensure_field_count(output.fields.len(), input.fields.len(), what) +} + +/// Checks that a node declares `declared` fields where its input provides `expected`. +/// +/// Some nodes compute their layout instead of keeping their input's: a final aggregate produces one +/// value per accumulator the state holds, and a vector-vector binary operation produces one field +/// per matched pair, which is the shorter side's count. Both are checked against the count the +/// kernel produces, so the message names the node and both numbers. +fn ensure_field_count(declared: usize, expected: usize, what: &str) -> Result<()> { + if declared != expected { + return Err(invalid(&format!( + "{what} declares {declared} field(s) but its input provides {expected}" + ))); + } + Ok(()) +} + +/// Checks that `ty` produces samples and returns them. +fn ensure_samples<'a>(ty: &'a NativeType, what: &str) -> Result<&'a Arc> { + ty.samples() + .ok_or_else(|| invalid(&format!("{what} does not produce samples"))) +} + +/// Checks that `ty` is evaluated on the grid `expected` describes. +/// +/// The comparison is by [`EvalGrid::spec`], not by allocation: that is the rule the kernels apply +/// (`vector_match::same_grid`), so a compiler that allocates one grid per window but binds equal +/// specs produces a tree the kernel runs correctly. +fn ensure_grid(ty: &NativeType, expected: &Arc, what: &str) -> Result<()> { + match ty.grid() { + Some(actual) if actual.spec() == expected.spec() => Ok(()), + Some(_) => Err(invalid(&format!( + "{what} declares a grid its output does not carry" + ))), + None => Err(invalid(&format!("{what} produces values without a grid"))), + } +} + +fn invalid(reason: &str) -> DataFusionError { + // A rejected tree is a compiler defect, not a query error: the compiler only builds a node + // after deciding that this combination is supported. `Internal` keeps that meaning on the wire + // instead of reporting a mistake in the client's statement, which `Plan` would. + DataFusionError::Internal(reason.to_string()) +} + +#[cfg(test)] +mod tests; diff --git a/src/promql/src/native/plan/tests.rs b/src/promql/src/native/plan/tests.rs new file mode 100644 index 00000000000..85be7426a58 --- /dev/null +++ b/src/promql/src/native/plan/tests.rs @@ -0,0 +1,1187 @@ +// 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. + +use std::sync::Arc; + +use datatypes::series::SeriesOrdering; +use datatypes::series::samples::{FieldSpec, SeriesSchema}; +use datatypes::series::time::{EvalGrid, GridSpec}; + +use super::{ + AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar, + BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, +}; +use crate::map::MathOp; +use crate::range_eval::{RangeFunction, RangeWindow}; +use crate::vector_match::{BinaryMatching, BinaryOp, Cardinality, MatchingLabels}; + +fn grid(start: i64, step: i64, len: u64) -> Arc { + Arc::new(EvalGrid::try_new(GridSpec::try_new(start, step, len).unwrap()).unwrap()) +} + +fn evaluated( + schema_fields: Arc<[datatypes::series::samples::FieldSpec]>, + grid: Arc, +) -> NativeType { + NativeType::Samples(Arc::new(SeriesSchema { + fields: schema_fields, + eval_grid: Some(grid), + })) +} + +/// `count` fields named `f0`, `f1`, ..., which is the layout a multi-field source has. +fn fields(count: usize) -> Arc<[FieldSpec]> { + (0..count) + .map(|index| FieldSpec { + name: Arc::from(format!("f{index}")), + }) + .collect::>() + .into() +} + +fn raw() -> NativeType { + NativeType::Samples(Arc::new(SeriesSchema { + fields: Arc::new([]), + eval_grid: None, + })) +} + +fn input() -> NativeNode { + let output = raw(); + NativeNode { + expr: NativeExpr::Input(BoundInput { + output: output.clone(), + ordering: SeriesOrdering::KeyThenTime, + }), + output, + } +} + +fn input_on(grid: Arc) -> NativeNode { + // One schema allocation shared by the binding and the output: `validate` checks that the output + // is the binding it was bound to, which is about the binding itself, not the grid's spec. + let output = evaluated(Arc::new([]), grid); + NativeNode { + expr: NativeExpr::Input(BoundInput { + output: output.clone(), + ordering: SeriesOrdering::Contiguous, + }), + output, + } +} + +/// A scan of `count` fields, so a consumer's field check has something to disagree with. +fn input_with_fields(count: usize) -> NativeNode { + let output = NativeType::Samples(Arc::new(SeriesSchema { + fields: fields(count), + eval_grid: None, + })); + NativeNode { + expr: NativeExpr::Input(BoundInput { + output: output.clone(), + ordering: SeriesOrdering::KeyThenTime, + }), + output, + } +} + +/// A constant scalar, which is a scalar type evaluated on the statement's grid. +fn scalar_on(value: f64, grid: Arc) -> NativeNode { + NativeNode { + expr: NativeExpr::Scalar(BoundScalar { value }), + output: NativeType::Scalar(grid), + } +} + +/// The grid a state node is indexed on. +fn state_grid(node: &NativeNode) -> Arc { + let NativeType::AggregateState(schema) = &node.output else { + unreachable!("expected aggregate state") + }; + Arc::clone(&schema.grid) +} + +fn partial_aggregate(inner: NativeNode) -> NativeNode { + let grid = inner + .output + .grid() + .expect("a partial aggregate consumes an evaluated vector") + .clone(); + NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(inner), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Partial, + }, + }, + output: NativeType::AggregateState(Arc::new(super::AggregateStateSchema { + group: Grouping::By(Arc::new(["job".into()])), + accumulators: Arc::new([]), + grid, + })), + } +} + +fn final_aggregate(state: NativeNode) -> NativeNode { + let grid = state_grid(&state); + NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(state), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Final, + }, + }, + output: evaluated(Arc::new([]), grid), + } +} + +fn window_over(inner: NativeNode, input_grid: Option>) -> NativeNode { + window_on(inner, input_grid, grid(0, 60_000, 4)) +} + +/// A window over `inner` that produces `count` fields on `out`, so a consumer of it can be given a +/// layout the window itself holds legally. +fn window_fields(inner: NativeNode, out: Arc, count: usize) -> NativeNode { + NativeNode { + expr: NativeExpr::Window { + input: Box::new(inner), + op: BoundWindow { + function: RangeFunction::Rate, + window: RangeWindow { + range_ms: 300_000, + offset_ms: 0, + anchor_ms: None, + }, + input_grid: None, + output_grid: Arc::clone(&out), + }, + }, + output: evaluated(fields(count), out), + } +} + +/// A window over `inner` that produces `out`, so two windows can be given the same grid. +fn window_on( + inner: NativeNode, + input_grid: Option>, + out: Arc, +) -> NativeNode { + NativeNode { + expr: NativeExpr::Window { + input: Box::new(inner), + op: BoundWindow { + function: RangeFunction::Rate, + window: RangeWindow { + range_ms: 300_000, + offset_ms: 0, + anchor_ms: None, + }, + input_grid, + output_grid: Arc::clone(&out), + }, + }, + output: evaluated(Arc::new([]), out), + } +} + +fn map_over(inner: NativeNode) -> NativeNode { + let out = inner.output.grid().expect("a map needs a grid").clone(); + NativeNode { + expr: NativeExpr::Map { + input: Box::new(inner), + op: BoundMap { + ops: Arc::new([MathOp::Abs]), + }, + }, + output: evaluated(Arc::new([]), out), + } +} + +fn add() -> BinaryMatching { + BinaryMatching { + op: BinaryOp::Add, + return_bool: false, + labels: MatchingLabels::Ignoring(Vec::new()), + cardinality: Cardinality::OneToOne, + } +} + +/// A compiled tree is accepted when every node's contract holds. +#[test] +fn a_window_over_raw_samples_is_valid() { + let root = window_over(input(), None); + assert!(NativeTree::try_new(root).is_ok()); +} + +/// Adding a map to a window changes the tree and nothing else: no new shape is recognised. +#[test] +fn adding_a_map_only_changes_the_tree() { + let bare = window_over(input(), None); + let mapped = map_over(window_over(input(), None)); + // Both trees are legal, so the comparison below is about the shape and not about both failing. + assert!(bare.validate().is_ok(), "{}", bare.validate().unwrap_err()); + assert!( + mapped.validate().is_ok(), + "{}", + mapped.validate().unwrap_err() + ); + // The mapped tree has one more node, and it is the same window below it. + let mut depth = 0; + super::super::metrics::walk(&mapped, &mut |_, _| depth += 1); + assert_eq!(depth, 3); + let NativeExpr::Map { input, .. } = &mapped.expr else { + unreachable!("the mapped tree's root is a map") + }; + assert!(matches!(input.expr, NativeExpr::Window { .. })); +} + +/// A map is point-wise over values, so raw samples are not something it can consume. +#[test] +fn a_map_cannot_consume_raw_samples() { + let node = NativeNode { + expr: NativeExpr::Map { + input: Box::new(input()), + op: BoundMap { + ops: Arc::new([MathOp::Abs]), + }, + }, + output: evaluated(Arc::new([]), grid(0, 60_000, 4)), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("an evaluated vector"), "{error}"); +} + +/// A scalar is one value per grid point, not a vector, so there are no fields to map. +#[test] +fn a_map_cannot_consume_a_scalar() { + let out = grid(0, 60_000, 4); + let node = NativeNode { + expr: NativeExpr::Map { + input: Box::new(scalar_on(1.0, Arc::clone(&out))), + op: BoundMap { + ops: Arc::new([MathOp::Abs]), + }, + }, + output: evaluated(Arc::new([]), out), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("not a scalar"), "{error}"); +} + +/// A map with no operation would be a node that does nothing, which is a compiler mistake. +#[test] +fn a_map_needs_an_operation() { + let inner = window_over(input(), None); + let out = inner.output.grid().unwrap().clone(); + let node = NativeNode { + expr: NativeExpr::Map { + input: Box::new(inner), + op: BoundMap { ops: Arc::new([]) }, + }, + output: evaluated(Arc::new([]), out), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("no operation"), "{error}"); +} + +/// A map reports one value per input field, so it cannot declare a layout its input does not +/// have. Without this check the tree would validate and only fail when the runtime compared the +/// node's schema against the one its input actually produces. +#[test] +fn a_map_cannot_change_the_field_count() { + let inner = window_over(input(), None); + let out = inner.output.grid().unwrap().clone(); + let node = NativeNode { + expr: NativeExpr::Map { + input: Box::new(inner), + op: BoundMap { + ops: Arc::new([MathOp::Abs]), + }, + }, + output: evaluated( + Arc::new([datatypes::series::samples::FieldSpec { + name: Arc::from("value"), + }]), + out, + ), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("a map declares 1 field(s) but its input provides 0"), + "{error}" + ); +} + +/// A window reports one value per input field, so it carries the same rule as a map. +#[test] +fn a_window_cannot_change_the_field_count() { + let inner = window_over(input(), None); + // The output grid is the input's, so the grid check passes and the field check is what the + // test actually exercises. + let out = inner.output.grid().unwrap().clone(); + let node = NativeNode { + expr: NativeExpr::Window { + input: Box::new(inner), + op: BoundWindow { + function: RangeFunction::Rate, + window: RangeWindow { + range_ms: 30_000, + offset_ms: 0, + anchor_ms: None, + }, + input_grid: Some(Arc::clone(&out)), + output_grid: Arc::clone(&out), + }, + }, + output: evaluated( + Arc::new([datatypes::series::samples::FieldSpec { + name: Arc::from("value"), + }]), + out, + ), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("a window declares 1 field(s) but its input provides 0"), + "{error}" + ); +} + +/// An instant selection picks one row of the source per grid point, so it carries one value per +/// input field. A tree that declares anything else would only fail at run time, against a schema +/// the tree was supposed to have validated. +#[test] +fn an_instant_selection_cannot_change_the_field_count() { + let out = grid(0, 60_000, 4); + // The source has two fields; the selection declares one value. + let node = NativeNode { + expr: NativeExpr::Instant { + input: Box::new(input_with_fields(2)), + op: BoundInstant { + lookback_ms: 300_000, + offset_ms: 0, + output_grid: Arc::clone(&out), + }, + }, + output: evaluated(fields(1), out), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("an instant selection declares 1 field(s) but its input provides 2"), + "{error}" + ); +} + +/// An instant selection reads raw samples; feeding it evaluated values would re-anchor a grid +/// that is already fixed. +#[test] +fn an_instant_selection_reads_raw_samples() { + let out = grid(0, 60_000, 4); + let node = NativeNode { + expr: NativeExpr::Instant { + input: Box::new(window_over(input(), None)), + op: BoundInstant { + lookback_ms: 300_000, + offset_ms: 0, + output_grid: Arc::clone(&out), + }, + }, + output: evaluated(Arc::new([]), out), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("reads evaluated"), "{error}"); +} + +/// Aggregate state is not a sample stream, so an instant selection cannot read it. +#[test] +fn an_instant_selection_cannot_consume_state() { + let out = grid(0, 60_000, 4); + let node = NativeNode { + expr: NativeExpr::Instant { + input: Box::new(partial_aggregate(window_on( + input(), + None, + Arc::clone(&out), + ))), + op: BoundInstant { + lookback_ms: 300_000, + offset_ms: 0, + output_grid: Arc::clone(&out), + }, + }, + output: evaluated(Arc::new([]), out), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("aggregate state"), "{error}"); +} + +/// A window's declared input grid has to be the grid its input actually produces. +#[test] +fn a_window_checks_its_declared_input_grid() { + let inner_grid = grid(0, 60_000, 4); + let other_grid = grid(0, 30_000, 8); + // The input produces `inner_grid`, but the window claims a different one. + let node = window_over(input_on(inner_grid), Some(other_grid)); + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("input grid is not the input's grid"), + "{error}" + ); +} + +/// A range evaluator reads one raw series to its last chunk before it starts the next, so raw +/// chunks of different series may not interleave. +#[test] +fn a_window_cannot_read_interleaved_input() { + let output = raw(); + let interleaved = NativeNode { + expr: NativeExpr::Input(BoundInput { + output: output.clone(), + ordering: SeriesOrdering::Interleaved, + }), + output, + }; + let error = window_over(interleaved, None).validate().unwrap_err(); + assert!( + error + .to_string() + .contains("interleaved chunks cannot be fed to it"), + "{error}" + ); +} + +/// A window whose range is empty or negative reads nothing at every grid point. +#[test] +fn a_window_reads_a_non_empty_range() { + let mut node = window_over(input(), None); + let NativeExpr::Window { op, .. } = &mut node.expr else { + unreachable!("a window tree has a window root") + }; + op.window.range_ms = 0; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("empty range"), "{error}"); +} + +/// A node that declares an output grid its output does not carry would be addressed on one grid +/// and hand out positions on another. +#[test] +fn a_window_checks_its_output_grid() { + let mut node = window_over(input(), None); + let NativeExpr::Window { op, .. } = &mut node.expr else { + unreachable!("a window tree has a window root") + }; + op.output_grid = grid(0, 30_000, 8); + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("declares a grid its output does not carry"), + "{error}" + ); +} + +/// A window consumes samples: state holds accumulators, not a sample stream to range over. +#[test] +fn a_window_cannot_consume_state() { + let out = grid(0, 60_000, 4); + let node = window_on( + partial_aggregate(window_on(input(), None, Arc::clone(&out))), + None, + grid(0, 60_000, 4), + ); + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("aggregate state"), "{error}"); +} + +/// A window ranges over timestamps, and a scalar has none of its own. +#[test] +fn a_window_cannot_consume_a_scalar() { + let out = grid(0, 60_000, 4); + let node = window_on( + scalar_on(1.0, Arc::clone(&out)), + Some(out), + grid(0, 60_000, 4), + ); + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("not a scalar"), "{error}"); +} + +/// A binary node aligns two inputs, so two different grids cannot be aligned implicitly. +#[test] +fn a_binary_operation_rejects_two_grids() { + let left = window_over(input(), None); + let left_grid = left.output.grid().unwrap().clone(); + let right = window_on(input(), None, grid(0, 30_000, 8)); + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(left), + right: Box::new(right), + matching: add(), + }, + output: evaluated(Arc::new([]), left_grid), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("different grids"), "{error}"); +} + +/// Two inputs on the same grid are a legal binary operation, which is how +/// `rate(a[5m]) + rate(b[5m])` becomes an ordinary tree. +#[test] +fn a_binary_operation_over_one_grid_is_valid() { + // Both sides evaluate on the statement's grid, so the compiler binds one grid to both. + let shared = grid(0, 60_000, 4); + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(window_on(input(), None, Arc::clone(&shared))), + right: Box::new(window_on(input(), None, Arc::clone(&shared))), + matching: add(), + }, + output: evaluated(Arc::new([]), shared), + }; + assert!(node.validate().is_ok()); +} + +/// A compiler that allocates one grid per window still produces one grid: the kernel aligns +/// operands by `EvalGrid::spec` (`vector_match::same_grid`), so `validate` compares specs too and +/// does not reject a tree the runtime would run correctly. +#[test] +fn a_binary_operation_aligns_separately_allocated_equal_specs() { + // Two windows, one fresh `EvalGrid` each, and the statement's grid a third: equal specs, no + // shared allocation anywhere in the tree. + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(window_on(input(), None, grid(0, 60_000, 4))), + right: Box::new(window_on(input(), None, grid(0, 60_000, 4))), + matching: add(), + }, + output: evaluated(Arc::new([]), grid(0, 60_000, 4)), + }; + assert!(node.validate().is_ok(), "{}", node.validate().unwrap_err()); + + // The same tree with a different spec is still rejected: the kernel rejects that pair too. + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(window_on(input(), None, grid(0, 60_000, 4))), + right: Box::new(window_on(input(), None, grid(30_000, 60_000, 4))), + matching: add(), + }, + output: evaluated(Arc::new([]), grid(0, 60_000, 4)), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("different grids"), "{error}"); +} + +/// A scalar side is broadcast, not matched: `sum by(job)(rate(x[5m])) + time()` has no labels to +/// line up, so `matching` is not consulted and the output keeps the vector side's fields. +#[test] +fn a_binary_operation_broadcasts_a_scalar() { + let shared = grid(0, 60_000, 4); + let vector = window_on(input(), None, Arc::clone(&shared)); + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(vector), + right: Box::new(scalar_on(1.0, Arc::clone(&shared))), + matching: add(), + }, + output: evaluated(Arc::new([]), Arc::clone(&shared)), + }; + assert!(node.validate().is_ok()); + + // The mirror case: the scalar on the left keeps the right side's fields. + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(scalar_on(1.0, Arc::clone(&shared))), + right: Box::new(window_on(input(), None, Arc::clone(&shared))), + matching: add(), + }, + output: evaluated(Arc::new([]), Arc::clone(&shared)), + }; + assert!(node.validate().is_ok()); +} + +/// A scalar has no fields of its own, so a broadcast that declares a different count than the +/// vector side would report a layout the kernel does not produce. +#[test] +fn a_scalar_broadcast_cannot_change_the_field_count() { + let shared = grid(0, 60_000, 4); + // The vector side has two fields; the broadcast declares one. + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(window_fields(input_with_fields(2), Arc::clone(&shared), 2)), + right: Box::new(scalar_on(1.0, Arc::clone(&shared))), + matching: add(), + }, + output: evaluated(fields(1), shared), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("does not produce the vector side's fields"), + "{error}" + ); +} + +/// Two vectors pair their fields by position, so the match reports the shorter side's count +/// (`vector_match::binary_output_schema` takes `min(lhs, rhs)`); a tree declaring anything else is +/// rejected instead of failing at run time. +#[test] +fn a_binary_operation_over_two_vectors_matches_the_shorter_side() { + let shared = grid(0, 60_000, 4); + // Two fields matched against one: the match produces one field, not two. + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(window_fields(input_with_fields(2), Arc::clone(&shared), 2)), + right: Box::new(window_fields(input_with_fields(1), Arc::clone(&shared), 1)), + matching: add(), + }, + output: evaluated(fields(2), Arc::clone(&shared)), + }; + let error = node.validate().unwrap_err(); + assert!( + error.to_string().contains( + "a binary operation over two vectors declares 2 field(s) but its input provides 1" + ), + "{error}" + ); + + // Declaring the shorter side's count is the legal tree, so the check is not vacuous. + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(window_fields(input_with_fields(2), Arc::clone(&shared), 2)), + right: Box::new(window_fields(input_with_fields(1), Arc::clone(&shared), 1)), + matching: add(), + }, + output: evaluated(fields(1), shared), + }; + assert!(node.validate().is_ok(), "{}", node.validate().unwrap_err()); +} + +/// Two scalars make a scalar, because there is nothing to attach labels or fields to. +#[test] +fn two_scalars_make_a_scalar() { + let shared = grid(0, 60_000, 4); + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(scalar_on(1.0, Arc::clone(&shared))), + right: Box::new(scalar_on(2.0, Arc::clone(&shared))), + matching: add(), + }, + output: NativeType::Scalar(Arc::clone(&shared)), + }; + assert!(node.validate().is_ok()); +} + +/// Two scalars have nothing to attach labels or fields to, so they cannot produce a vector. +#[test] +fn two_scalars_cannot_produce_a_vector() { + let shared = grid(0, 60_000, 4); + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(scalar_on(1.0, Arc::clone(&shared))), + right: Box::new(scalar_on(2.0, Arc::clone(&shared))), + matching: add(), + }, + output: evaluated(Arc::new([]), shared), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("two scalars do not produce a scalar"), + "{error}" + ); +} + +/// A binary operation is point-wise on a grid, so a side of raw samples has no positions to line +/// up with the other side. +#[test] +fn a_binary_operation_cannot_consume_raw_samples() { + let shared = grid(0, 60_000, 4); + let vector = || window_on(input(), None, Arc::clone(&shared)); + + let left_raw = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(input()), + right: Box::new(vector()), + matching: add(), + }, + output: evaluated(Arc::new([]), Arc::clone(&shared)), + }; + let error = left_raw.validate().unwrap_err(); + assert!( + error.to_string().contains("not raw samples on its left"), + "{error}" + ); + + let right_raw = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(vector()), + right: Box::new(input()), + matching: add(), + }, + output: evaluated(Arc::new([]), shared), + }; + let error = right_raw.validate().unwrap_err(); + assert!( + error.to_string().contains("not raw samples on its right"), + "{error}" + ); +} + +/// State is not a value, so neither side of a binary operation may be state. +#[test] +fn a_binary_operation_cannot_consume_state() { + let shared = grid(0, 60_000, 4); + let node = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(window_on(input(), None, Arc::clone(&shared))), + right: Box::new(partial_aggregate(window_on( + input(), + None, + Arc::clone(&shared), + ))), + matching: add(), + }, + output: evaluated(Arc::new([]), Arc::clone(&shared)), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("aggregate state"), "{error}"); +} + +/// A partial aggregate produces state, not a value; a complete one produces a value on the same +/// grid as its input. +#[test] +fn an_aggregate_output_matches_its_stage() { + let inner = window_over(input(), None); + let value_grid = inner.output.grid().unwrap().clone(); + + let complete = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(inner.clone()), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Complete, + }, + }, + output: evaluated(Arc::new([]), Arc::clone(&value_grid)), + }; + assert!(complete.validate().is_ok()); + + let partial = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(inner), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Partial, + }, + }, + output: NativeType::AggregateState(Arc::new(super::AggregateStateSchema { + group: Grouping::By(Arc::new(["job".into()])), + accumulators: Arc::new([]), + grid: value_grid, + })), + }; + assert!(partial.validate().is_ok()); +} + +/// A complete aggregate folds each field on its own, so it reports one value per input field +/// (`aggregate::aggregate_float` builds its rows from the input schema's fields). +#[test] +fn a_complete_aggregate_cannot_change_the_field_count() { + let out = grid(0, 60_000, 4); + // The input is a legal two-field window; the aggregate declares one value. + let node = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(window_fields(input_with_fields(2), Arc::clone(&out), 2)), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Complete, + }, + }, + output: evaluated(fields(1), out), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("a complete aggregate declares 1 field(s) but its input provides 2"), + "{error}" + ); +} + +/// Declaring a value where the stage produces state is a compiler mistake, and is rejected. +#[test] +fn an_aggregate_stage_must_agree_with_its_type() { + let inner = window_over(input(), None); + let value_grid = inner.output.grid().unwrap().clone(); + let node = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(inner), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Partial, + }, + }, + output: evaluated(Arc::new([]), value_grid), + }; + let error = node.validate().unwrap_err(); + assert!( + error.to_string().contains("produces aggregate state"), + "{error}" + ); +} + +/// A `Partial` aggregate keeps one accumulator per input field (`aggregate_float` lays its state +/// out in `schema.fields.len()`-sized points), so state declaring another count is rejected. +#[test] +fn a_partial_aggregate_cannot_change_the_accumulator_count() { + let out = grid(0, 60_000, 4); + // The input is a legal two-field window; the state declares one accumulator. + let node = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(window_fields(input_with_fields(2), Arc::clone(&out), 2)), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Partial, + }, + }, + output: NativeType::AggregateState(Arc::new(super::AggregateStateSchema { + group: Grouping::By(Arc::new(["job".into()])), + accumulators: fields(1), + grid: out, + })), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("a partial aggregate declares 1 field(s) but its input provides 2"), + "{error}" + ); +} + +/// A `Partial` aggregate's state is grouped by the binding's group labels, so state declaring +/// other labels is not the state this aggregate produces. +#[test] +fn a_partial_aggregate_groups_its_state_like_its_binding() { + let mut node = partial_aggregate(window_over(input(), None)); + let NativeExpr::Aggregate { op, .. } = &mut node.expr else { + unreachable!("a partial aggregate tree has an aggregate root") + }; + op.group = Grouping::By(Arc::new(["instance".into()])); + let error = node.validate().unwrap_err(); + assert!( + error.to_string().contains("groups differ from the binding"), + "{error}" + ); +} + +/// A `Partial` aggregate indexes its state on the grid its input produces, so state on another +/// grid is addressed with positions the input never wrote. +#[test] +fn a_partial_aggregate_indexes_its_state_on_the_inputs_grid() { + let node = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(window_over(input(), None)), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Partial, + }, + }, + output: NativeType::AggregateState(Arc::new(super::AggregateStateSchema { + group: Grouping::By(Arc::new(["job".into()])), + accumulators: Arc::new([]), + grid: grid(0, 30_000, 8), + })), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("indexed on a grid the input does not produce"), + "{error}" + ); +} + +/// A `Final` aggregate merges into the binding's group labels, so the state it reads has to be +/// grouped the same way. +#[test] +fn a_final_aggregate_groups_its_state_like_its_binding() { + let mut node = final_aggregate(partial_aggregate(window_over(input(), None))); + let NativeExpr::Aggregate { op, .. } = &mut node.expr else { + unreachable!("a final aggregate tree has an aggregate root") + }; + op.group = Grouping::By(Arc::new(["instance".into()])); + let error = node.validate().unwrap_err(); + assert!( + error.to_string().contains("groups differ from the binding"), + "{error}" + ); +} + +/// `by` and `without` are not interchangeable, even over the same label. +/// +/// `by(job)` keeps `job` and drops every other label; `without(job)` drops `job` and keeps the +/// rest. A binding that spelled `without` as a label list could only produce the first, so state +/// built for one has to be refused for the other rather than merged under the wrong labels. +#[test] +fn a_grouping_cannot_swap_by_for_without() { + let mut node = final_aggregate(partial_aggregate(window_over(input(), None))); + let NativeExpr::Aggregate { op, .. } = &mut node.expr else { + unreachable!("a final aggregate tree has an aggregate root") + }; + op.group = Grouping::Without(Arc::new(["job".into()])); + let error = node.validate().unwrap_err(); + assert!( + error.to_string().contains("groups differ from the binding"), + "{error}" + ); + + // The same labels under the same shape validate, so the refusal is about the shape. + let mut node = final_aggregate(partial_aggregate(window_over(input(), None))); + let NativeExpr::Aggregate { op, .. } = &mut node.expr else { + unreachable!("a final aggregate tree has an aggregate root") + }; + op.group = Grouping::By(Arc::new(["job".into()])); + assert!(node.validate().is_ok()); +} + +/// A `Final` aggregate consumes the state a `Partial` produced, which is what an exchange +/// boundary hands over between the two stages. +#[test] +fn a_final_aggregate_merges_partial_state() { + let shared = grid(0, 60_000, 4); + let state = partial_aggregate(window_on(input(), None, Arc::clone(&shared))); + let merged = final_aggregate(state); + assert!(merged.validate().is_ok()); + assert!( + matches!(&merged.output, NativeType::Samples(schema) if schema.eval_grid.as_ref().is_some_and(|grid| Arc::ptr_eq(grid, &shared))) + ); +} + +/// A `Final` aggregate merges the state into one value per accumulator, so it reports the state's +/// count and nothing else. +#[test] +fn a_final_aggregate_cannot_change_the_field_count() { + let out = grid(0, 60_000, 4); + // A legal partial aggregate: two input fields, two accumulators. + let state = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(window_fields(input_with_fields(2), Arc::clone(&out), 2)), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Partial, + }, + }, + output: NativeType::AggregateState(Arc::new(super::AggregateStateSchema { + group: Grouping::By(Arc::new(["job".into()])), + accumulators: fields(2), + grid: Arc::clone(&out), + })), + }; + assert!( + state.validate().is_ok(), + "{}", + state.validate().unwrap_err() + ); + + // The merge declares one value for the state's two accumulators. + let node = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(state), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Final, + }, + }, + output: evaluated(fields(1), out), + }; + let error = node.validate().unwrap_err(); + assert!( + error + .to_string() + .contains("a final aggregate declares 1 field(s) but its input provides 2"), + "{error}" + ); +} + +/// A `Final` aggregate consumes state, so a vector is not something it can merge. +#[test] +fn a_final_aggregate_cannot_consume_a_window() { + let inner = window_over(input(), None); + let value_grid = inner.output.grid().unwrap().clone(); + let node = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(inner), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Final, + }, + }, + output: evaluated(Arc::new([]), value_grid), + }; + let error = node.validate().unwrap_err(); + assert!( + error.to_string().contains("consumes aggregate state"), + "{error}" + ); +} + +/// A `Complete` aggregate aggregates vectors, so state is not something it can consume. +#[test] +fn a_complete_aggregate_cannot_consume_state() { + let state = partial_aggregate(window_over(input(), None)); + let state_grid = state_grid(&state); + let node = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(state), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Complete, + }, + }, + output: evaluated(Arc::new([]), state_grid), + }; + let error = node.validate().unwrap_err(); + assert!( + error.to_string().contains("consumes an evaluated vector"), + "{error}" + ); +} + +/// The first-phase walkthrough shapes compile to trees, which is the acceptance the RFC states +/// for composition. +#[test] +fn the_walkthrough_shapes_are_trees() { + // Every node in one query evaluates on the statement's grid; `validate` compares the specs of + // the grids, not their allocations, which is the rule the kernels apply. + let statement_grid = grid(0, 60_000, 4); + + // `abs(rate(x[5m]))` + let abs_rate = map_over(window_over(input(), None)); + assert!(abs_rate.validate().is_ok()); + + // `sum by(job)(rate(x[5m]))`, complete because one execution sees every contribution. + let sum = NativeNode { + expr: NativeExpr::Aggregate { + input: Box::new(window_on(input(), None, Arc::clone(&statement_grid))), + op: BoundAggregate { + kind: AggregateKind::Sum, + group: Grouping::By(Arc::new(["job".into()])), + stage: AggregateStage::Complete, + }, + }, + output: evaluated(Arc::new([]), Arc::clone(&statement_grid)), + }; + assert!(sum.validate().is_ok()); + + // `sum by(job)(rate(x[5m])) + 1`: the aggregate is complete, and the scalar side is + // broadcast, so `matching` is not consulted for it. + let plus_one = NativeNode { + expr: NativeExpr::Binary { + left: Box::new(sum.clone()), + right: Box::new(scalar_on(1.0, Arc::clone(&statement_grid))), + matching: add(), + }, + output: evaluated(Arc::new([]), statement_grid), + }; + assert!(plus_one.validate().is_ok()); +} + +/// A scalar node's output is the scalar type, which is what keeps a broadcast from turning into +/// label matching. +#[test] +fn a_scalar_node_is_valid() { + let node = scalar_on(1.0, grid(0, 60_000, 4)); + assert!(node.validate().is_ok()); +} + +/// A scalar node declared as a vector would let a broadcast be planned as a match. +#[test] +fn a_scalar_node_must_produce_a_scalar() { + let node = NativeNode { + expr: NativeExpr::Scalar(BoundScalar { value: 1.0 }), + output: evaluated(Arc::new([]), grid(0, 60_000, 4)), + }; + let error = node.validate().unwrap_err(); + assert!( + error.to_string().contains("does not produce a scalar"), + "{error}" + ); +} + +/// An input whose declared output is not the binding it was bound to cannot be trusted. +#[test] +fn an_input_must_report_the_binding_it_was_bound_to() { + let node = NativeNode { + expr: NativeExpr::Input(BoundInput { + output: raw(), + ordering: SeriesOrdering::KeyThenTime, + }), + output: raw(), + }; + let error = node.validate().unwrap_err(); + assert!(error.to_string().contains("not the binding"), "{error}"); +} + +/// An exchange boundary binds state, and the node has to report that same state. +#[test] +fn an_input_can_bind_aggregate_state() { + let state = partial_aggregate(window_over(input(), None)); + let bound = NativeNode { + expr: NativeExpr::Input(BoundInput { + output: state.output.clone(), + ordering: SeriesOrdering::Contiguous, + }), + output: state.output.clone(), + }; + assert!(bound.validate().is_ok()); + + // A `Final` aggregate can then consume it, which is the shape of a distributed aggregate. + assert!(final_aggregate(bound).validate().is_ok()); + + // A fresh state allocation is a different binding, so it cannot be reported as the bound one. + let other = NativeNode { + expr: NativeExpr::Input(BoundInput { + output: state.output.clone(), + ordering: SeriesOrdering::Contiguous, + }), + output: partial_aggregate(window_over(input(), None)).output, + }; + let error = other.validate().unwrap_err(); + assert!(error.to_string().contains("not the binding"), "{error}"); +}