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}"); +}