diff --git a/src/promql/src/histogram_eval/aggregate.rs b/src/promql/src/histogram_eval/aggregate.rs index 9e1721c8b59..0ffcca5e8e8 100644 --- a/src/promql/src/histogram_eval/aggregate.rs +++ b/src/promql/src/histogram_eval/aggregate.rs @@ -156,7 +156,8 @@ fn aggregate_point<'a>( /// The Kahan sum or the mean of `histograms` in order, without compaction; `None` when there is /// none or they mix exponential and custom buckets. Prometheus `sum_over_time` and /// `avg_over_time` run the same steps as the aggregations. Histograms are consumed one at a -/// time, so the first error ends the fold. +/// time, so the first error ends the fold. The result is derived, so it carries no start +/// timestamp. pub(crate) fn sum_or_avg( op: HistogramAggregate, histograms: impl IntoIterator>, @@ -236,5 +237,11 @@ pub(crate) fn sum_or_avg( } }, } + // The result was derived from its inputs, so it carries no start timestamp of its own, like + // every other operation that builds one. Clearing it here covers the fold that added nothing: + // the first histogram of a window becomes the result as it arrived, and the adds above are + // what drop the field otherwise, which would make the payload depend on the sample count. + // The bucket layout is left as the fold built it -- compaction belongs to the caller. + value.start_timestamp = None; Ok(Some(value)) } diff --git a/src/promql/src/histogram_eval/float_histogram.rs b/src/promql/src/histogram_eval/float_histogram.rs index acce48c1776..22d016418ad 100644 --- a/src/promql/src/histogram_eval/float_histogram.rs +++ b/src/promql/src/histogram_eval/float_histogram.rs @@ -35,6 +35,22 @@ type Buckets = Vec<(i32, f64)>; #[derive(Debug, Clone)] pub(crate) struct FloatHistogram { pub(crate) reset_hint: CounterResetHint, + /// The start timestamp of the sample this histogram was read from, in milliseconds. + /// + /// It is metadata of GreptimeDB's histogram row, not part of the histogram value that + /// Prometheus's `FloatHistogram` models. This owned form preserves it -- [`Self::try_from_view`] + /// reads it off the sample and [`Self::push_to`] writes it back -- while every operation that + /// derives a histogram from its input ([`Self::add`], [`Self::kahan_add`], [`Self::mul`], + /// [`Self::div`], [`Self::compact`], [`Self::copy_to_schema`], with the constructors beside + /// them) clears it. [`Self::equals`] does not compare it. + /// + /// Both halves are this module's own policy and say nothing about how Prometheus propagates + /// these timestamps: its `FloatHistogram` does not carry the field, while its rate and reset + /// evaluation receives start timestamps through a separate channel. This crate's native range + /// kernel does not read the field; the legacy extrapolation path does, to synthesize a zero + /// point. So a selection that keeps the field preserves what was stored, and does not by itself + /// make a native rate agree with the legacy one. + pub(crate) start_timestamp: Option, pub(crate) schema: i32, pub(crate) zero_threshold: f64, pub(crate) zero_count: f64, @@ -62,6 +78,7 @@ impl FloatHistogram { pub(crate) fn try_from_view(view: &HistogramView<'_>) -> Result { Ok(Self { reset_hint: view.reset_hint, + start_timestamp: view.start_timestamp, schema: view.schema, zero_threshold: view.zero_threshold, zero_count: view.zero_count, @@ -130,7 +147,7 @@ impl FloatHistogram { count: self.count, zero_count: self.zero_count, reset_hint: self.reset_hint, - start_timestamp: None, + start_timestamp: self.start_timestamp, custom_values: &self.custom_values, positive: BucketSideView { span_offsets: &positive_offsets, @@ -151,6 +168,7 @@ impl FloatHistogram { pub(crate) fn empty_like(&self) -> Self { Self { reset_hint: CounterResetHint::Unknown, + start_timestamp: None, schema: self.schema, zero_threshold: 0.0, zero_count: 0.0, @@ -169,6 +187,8 @@ impl FloatHistogram { /// Go `Mul`. pub(crate) fn mul(&mut self, factor: f64) { + // A result this operation derived, so it carries no start timestamp of its own. + self.start_timestamp = None; self.zero_count *= factor; self.count *= factor; self.sum *= factor; @@ -182,6 +202,8 @@ impl FloatHistogram { /// Go `Div`; dividing by zero removes every bucket. pub(crate) fn div(&mut self, scalar: f64) { + // A result this operation derived, so it carries no start timestamp of its own. + self.start_timestamp = None; self.zero_count /= scalar; self.count /= scalar; self.sum /= scalar; @@ -215,6 +237,8 @@ impl FloatHistogram { /// Go `Compact(0)`: drops empty buckets. pub(crate) fn compact(&mut self) { + // A result this operation derived, so it carries no start timestamp of its own. + self.start_timestamp = None; self.positive.retain(|&(_, count)| count != 0.0); self.negative.retain(|&(_, count)| count != 0.0); self.layout = None; @@ -223,13 +247,19 @@ impl FloatHistogram { /// Go `CopyToSchema`; `target` must not exceed the schema. pub(crate) fn copy_to_schema(&self, target: i32) -> Result { if target == self.schema { - return Ok(self.clone()); + // The copy is a histogram this call produced, like the reduced one below: a start + // timestamp belongs to the sample that carried it, not to a value derived from it. + return Ok(Self { + start_timestamp: None, + ..self.clone() + }); } if self.uses_custom_buckets() || target == CUSTOM_BUCKETS_SCHEMA || target > self.schema { return Err(incompatible_schema()); } Ok(Self { reset_hint: CounterResetHint::Unknown, + start_timestamp: None, schema: target, zero_threshold: self.zero_threshold, zero_count: self.zero_count, @@ -255,6 +285,8 @@ impl FloatHistogram { fn add_or_sub(&mut self, other: &Self, subtract: bool) -> Result { check_schemas(self, other)?; self.layout = None; + // A result this operation derived, so it carries no start timestamp of its own. + self.start_timestamp = None; let mut outcome = AddOutcome { counter_reset_collision: self.adjust_counter_reset(other), custom_bounds_reconciled: false, @@ -323,6 +355,8 @@ impl FloatHistogram { ) -> Result<(Self, AddOutcome)> { check_schemas(self, other)?; self.layout = None; + // A result this operation derived, so it carries no start timestamp of its own. + self.start_timestamp = None; let mut outcome = AddOutcome { counter_reset_collision: self.adjust_counter_reset(other), custom_bounds_reconciled: false, @@ -410,6 +444,7 @@ impl FloatHistogram { fn new_compensation(&self) -> Self { Self { reset_hint: self.reset_hint, + start_timestamp: None, schema: self.schema, zero_threshold: self.zero_threshold, zero_count: 0.0, diff --git a/src/promql/src/histogram_eval/tests.rs b/src/promql/src/histogram_eval/tests.rs index 8b815b0a49d..258e747a4cb 100644 --- a/src/promql/src/histogram_eval/tests.rs +++ b/src/promql/src/histogram_eval/tests.rs @@ -16,6 +16,9 @@ use std::collections::BTreeMap; use std::sync::Arc; use datafusion::error::DataFusionError; +use datatypes::series::histogram::{ + BucketSideView, CounterResetHint, HistogramBatchBuilder, HistogramView, +}; use datatypes::series::samples::{FieldSpec, SampleRef, Series, SeriesSchema}; use datatypes::series::time::{EvalGrid, GridRange, GridSpec, TimeAxis}; @@ -383,6 +386,75 @@ fn histogram_points(series: &Series) -> BTreeMap { .collect() } +/// The start timestamp is payload of a histogram that was read, and not of one that was computed. +/// +/// This pins GreptimeDB's own payload policy, not a Prometheus expectation: the timestamp is read +/// off a sample ([`FloatHistogram::try_from_view`]) and written back +/// ([`FloatHistogram::push_to`]), which is what a selection needs to hand a sample on unchanged, +/// and every operation that derives a value from its input clears it. Nothing else depends on the +/// field: the type derives `Debug` and `Clone` only, so it has no equality, hash or serialized form +/// that could compare it, and [`FloatHistogram::equals`] is Go `Equals` on counts and bucket layout. +#[test] +fn a_start_timestamp_belongs_to_a_read_histogram_and_not_to_a_computed_one() { + let offsets = [0]; + let lengths = [2]; + let buckets = [1.0, 2.0]; + let view = HistogramView { + schema: 0, + zero_threshold: 0.0, + zero_count: 0.0, + count: 3.0, + sum: 3.0, + reset_hint: CounterResetHint::Unknown, + start_timestamp: Some(1500), + custom_values: &[], + positive: BucketSideView { + span_offsets: &offsets, + span_lengths: &lengths, + buckets: &buckets, + }, + negative: BucketSideView { + span_offsets: &[], + span_lengths: &[], + buckets: &[], + }, + }; + let histogram = FloatHistogram::try_from_view(&view).unwrap(); + assert_eq!(histogram.start_timestamp, Some(1500)); + + // Writing it back keeps it, which is what makes a selected sample a sample. + let mut builder = HistogramBatchBuilder::new(); + histogram.push_to(&mut builder).unwrap(); + let table = builder.finish().unwrap(); + assert_eq!( + table.view(0).unwrap().start_timestamp, + Some(1500), + "the histogram was written back without the start timestamp it was read with" + ); + + // A value derived from it carries none, whichever operation derived it. + let mut added = histogram.clone(); + added.add(&histogram).unwrap(); + assert_eq!(added.start_timestamp, None); + let mut summed = histogram.clone(); + let (compensation, _) = summed.kahan_add(&histogram, None).unwrap(); + assert_eq!(summed.start_timestamp, None); + assert_eq!(compensation.start_timestamp, None); + let mut multiplied = histogram.clone(); + multiplied.mul(2.0); + assert_eq!(multiplied.start_timestamp, None); + let mut divided = histogram.clone(); + divided.div(2.0); + assert_eq!(divided.start_timestamp, None); + let mut compacted = histogram.clone(); + compacted.compact(); + assert_eq!(compacted.start_timestamp, None); + assert_eq!(histogram.copy_to_schema(0).unwrap().start_timestamp, None); + assert_eq!(histogram.empty_like().start_timestamp, None); + // The one it was read into is still the sample it was read from. + assert_eq!(histogram.start_timestamp, Some(1500)); +} + /// `histogram_quantile` drops all metadata labels; the single-histogram functions drop only /// `__name__`, as Prometheus does. #[test] diff --git a/src/promql/src/native/execute.rs b/src/promql/src/native/execute.rs index 25e2c4a3c94..84bc07aa752 100644 --- a/src/promql/src/native/execute.rs +++ b/src/promql/src/native/execute.rs @@ -30,6 +30,12 @@ //! outer window reads the inner result as the samples it is, not as something to be resampled onto //! the outer grid first. //! +//! An instant selection owns no evaluator, because [`select_instants`] takes a complete source +//! axis: the node buffers the chunks of the series it is selecting until the chunk that carries +//! `is_last`, selects the last sample at or before each grid point of it, and hands the series over +//! as one chunk addressed on its own grid. Nothing is decided for a series before it ends, which is +//! the price of the kernel taking one axis rather than a chunk of one. +//! //! A node kind this runtime does not execute yet is rejected when the execution is built, naming //! the kind and its position. An unsupported tree therefore never starts, rather than answering //! half a query or panicking halfway through one. @@ -68,20 +74,23 @@ use datafusion::error::{DataFusionError, Result}; use datafusion::execution::memory_pool::{MemoryConsumer, MemoryPool, MemoryReservation}; use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet; use datatypes::arrow::array::{Array, Float64Array}; -use datatypes::arrow::buffer::Buffer; -use datatypes::series::histogram::HistogramBatch; -use datatypes::series::raw::RawSeriesChunk; +use datatypes::arrow::buffer::{Buffer, ScalarBuffer}; +use datatypes::series::histogram::{HistogramBatch, HistogramView}; +use datatypes::series::raw::{RawSeriesChunk, RawTimestamps}; use datatypes::series::samples::{ - DenseSamplesRef, SampleRef, Samples, Series, SeriesFields, SeriesSchema, + DenseSamplesRef, RowCursor, SampleRef, Samples, Series, SeriesFields, SeriesSchema, }; -use datatypes::series::time::{GridRange, TimeAxis}; +use datatypes::series::time::{EvalGrid, GridRange, GridSpec, TimeAxis}; use datatypes::series::{MetricName, SeriesOrdering}; -use crate::grid_series::{LabelSet, label_set, merge_same_labelset}; +use crate::grid_series::{ + GridSample, GridSeriesBuilder, LabelSet, label_set, merge_same_labelset, owned, +}; +use crate::instant_eval::select_instants; use crate::map::{MathOp, drop_metric_name, map_math}; use crate::native::batch::EvaluatedChunk; use crate::native::metrics::{NodeMetrics, NodePath}; -use crate::native::plan::{NativeExpr, NativeTree}; +use crate::native::plan::{BoundInstant, NativeExpr, NativeTree}; use crate::range_eval::RangeEvaluator; #[cfg(test)] @@ -261,6 +270,8 @@ fn buffer_bytes(buffer: &Buffer) -> usize { enum NodeTag { /// Forwards the chunks it is given. Input, + /// Buffers one series and selects the last sample at or before each of its grid points. + Instant, /// Feeds its input to its own [`RangeEvaluator`]. Window, /// Applies point-wise operations to its input chunks. @@ -272,6 +283,7 @@ impl NodeTag { fn describe(self) -> &'static str { match self { Self::Input => "input", + Self::Instant => "instant selection", Self::Window => "window", Self::Map => "map", } @@ -836,6 +848,438 @@ struct MapRuntime { schema: Arc, } +/// What an instant node holds: the chunks of the one series it is selecting, until that series ends. +/// +/// [`select_instants`] takes a complete source axis, so this node cannot decide a grid point from +/// the samples one chunk carries the way a window decides a range from what it has retained: it +/// buffers a whole series and hands one chunk over for it. What it holds is therefore the series' +/// own bytes, not a fixed window of them, and the reservation follows the buffer. +struct InstantRuntime { + /// The bound selection: how far back it reads, the offset it reads at, and the grid its values + /// are addressed on. + selection: BoundInstant, + /// The layout the chunk it hands out is addressed on: its input's fields on that grid. + schema: Arc, + /// Where the rows the selection picked are gathered, one series at a time. + output: GridSeriesBuilder, + /// The chunks of the series being selected, in arrival order; empty while no series is open. + chunks: Vec, + /// What `chunks` hold, added as each chunk arrives and given back when the series is gathered. + /// + /// Each chunk is measured once, with the estimate a queued chunk is charged with: an allocation + /// two chunks of the series share is counted once per chunk, which is the over-count a figure + /// that keeps no state across chunks accepts. + held: usize, +} + +impl InstantRuntime { + /// The run-time state of the instant selection `op`, reading the samples of `input`. + fn try_new(input: &SeriesSchema, op: &BoundInstant) -> Result { + // The chunk this node hands out is addressed on the selection's own grid, so the grid the + // rows are gathered on is the binding's rather than a copy of it. + let schema = Arc::new(SeriesSchema { + fields: Arc::clone(&input.fields), + eval_grid: Some(Arc::clone(&op.output_grid)), + }); + Ok(Self { + selection: op.clone(), + output: GridSeriesBuilder::try_new(Arc::clone(&schema))?, + schema, + chunks: Vec::new(), + held: 0, + }) + } + + /// The bytes the buffered chunks of the open series hold. + fn buffered_bytes(&self) -> usize { + self.held + } + + /// The bytes gathering the series `closing` ends is about to allocate beside its chunks. + /// + /// An upper bound derived from what the selection has in hand rather than a measurement: the + /// rows the series holds once `closing` has joined it, the fields it carries, the points the + /// selection can decide for it, and, per field, the largest histogram a row of it can carry. + /// Each term is one thing the gather allocates: + /// + /// - The concatenated axis ([`buffered_time`]): one `i64` per row, reserved by the row count. + /// - The descriptors that read it ([`buffered_fields`]): one [`Samples`] per (chunk, field) for + /// the per-field vectors, three per segment for the vector [`Samples::try_from_chunks`] + /// flattens them into -- the segment it came from, the vector it is flattened into and the + /// `Arc<[Samples]>` the result is copied into -- and one `Vec` header per field. + /// - The two arrays [`select_instants`] returns: a `u64` source row index and an `i64` + /// evaluation timestamp per point, reserved by the count they can reach at most. + /// - The rows [`GridSeriesBuilder`] keeps: one `usize` grid index, one `f64` per field and a + /// validity byte per (row, field). A field of native histograms adds one `Option` index + /// per row and the payload of the histogram that index names, charged at the largest one the + /// field holds and per *point* rather than per row: one source row can be selected by every + /// point the selection decides, so the point count is what that payload is multiplied by. + /// - The row the selection fills, one [`GridSample`] slot per field, and one [`RowCursor`] per + /// field to read the axis with. + /// - Doubling: every column the builder keeps is a `Vec` that grows by doubling, so each is + /// charged twice what it ends up holding, which bounds its capacity rather than its length. + /// + /// Left out, deliberately: the headers and Arrow buffers behind the buffered chunks (the + /// node's own reservation already holds them, allocation by allocation), the `Vec` headers the + /// gather drops before it returns, `Series`' own header and the runs [`GridSeriesBuilder`] + /// hands the output over with, which the output chunk is charged for by whoever reads it, and + /// the temporaries [`FloatHistogram::push_to`] builds for one row. None of those is + /// proportional to the points the selection decides, which is the amplification the pool has to + /// be asked for, and all of them are constant per gather or already counted row by row. + /// + /// The estimate applies to every gather: nothing refuses one before it allocates, so the points + /// the series can be decided at are reserved for whatever their number is. + /// + /// [`FloatHistogram::push_to`]: crate::histogram_eval::float_histogram::FloatHistogram::push_to + fn gather_bytes(&self, closing: &RawSeriesChunk) -> usize { + // The points the selection decides, not the points the grid plans: the gather clips its + // scan to the series it holds, so a figure over the whole grid would charge the pool for + // rows this series can never produce. + let points = self.selected_points(closing); + let chunks = self.chunks.len().saturating_add(1); + let rows = self + .chunks + .iter() + .chain(std::iter::once(closing)) + .map(|chunk| chunk.time.len()) + .fold(0usize, usize::saturating_add); + let fields = self.schema.fields.len(); + let descriptor = std::mem::size_of::(); + let segments = self + .chunks + .iter() + .chain(std::iter::once(closing)) + .flat_map(|chunk| chunk.fields.as_slice()) + .map(|field| field.segments().len()) + .fold(0usize, usize::saturating_add); + let axis = rows.saturating_mul(std::mem::size_of::()); + let layout = chunks + .saturating_mul(fields) + .saturating_mul(descriptor) + .saturating_add(segments.saturating_mul(3).saturating_mul(descriptor)) + .saturating_add(fields.saturating_mul(std::mem::size_of::>())); + let selection = + points.saturating_mul(std::mem::size_of::() + std::mem::size_of::()); + let mut rows_of_points = points.saturating_mul( + std::mem::size_of::() + fields.saturating_mul(std::mem::size_of::() + 1), + ); + for largest in self.largest_histograms(closing).iter().flatten() { + rows_of_points = rows_of_points.saturating_add( + points.saturating_mul(std::mem::size_of::>().saturating_add(*largest)), + ); + } + let row = fields.saturating_mul( + std::mem::size_of::>() + std::mem::size_of::>(), + ); + axis.saturating_add(layout) + .saturating_add(selection) + .saturating_add(row) + .saturating_add(rows_of_points.saturating_mul(2)) + } + + /// The number of grid points [`select_instants`] clips the buffered series to. + /// + /// The gather clips its scan to the series it holds instead of reading the whole planned grid: + /// the candidate points run from the first sample's instant to the last sample's plus the + /// lookback, aligned to the grid and held inside the query range. A workspace is charged per + /// point, so the count has to be the clipped one -- charging the planned grid charges the pool + /// for rows a short series can never produce, and a pool sized for what the series costs then + /// refuses the gather. The clipping below is [`select_instants`]' own, repeated rather than + /// approximated: an estimate that disagrees with the kernel is a bug here rather than a + /// conservative bound. It is the kernel's own candidate count -- the one it sizes its arrays + /// with -- so it bounds the rows it can report even though a candidate at the lookback edge + /// itself selects nothing. Both ends come from chunks the node already holds, so the count + /// reads two timestamps rather than the rows. + fn selected_points(&self, closing: &RawSeriesChunk) -> usize { + let spec = self.selection.output_grid.spec(); + // The selection refuses these itself, before it can decide anything. + if spec.step_ms <= 0 || self.selection.lookback_ms < 0 { + return 0; + } + let query_start = i128::from(spec.start_ms); + let query_end = i128::from(grid_end_ms(spec)); + if query_end < query_start { + return 0; + } + let series = || self.chunks.iter().chain(std::iter::once(closing)); + let (Some(first), Some(last)) = ( + series().find_map(|chunk| chunk.time.as_buffer().first().copied()), + series() + .rev() + .find_map(|chunk| chunk.time.as_buffer().last().copied()), + ) else { + // A series of no sample decides no point. + return 0; + }; + let offset = i128::from(self.selection.offset_ms); + let lookback = i128::from(self.selection.lookback_ms); + let first = i128::from(first) + offset; + let last = i128::from(last) + offset; + let useful_last = if lookback == 0 { + last + } else { + last + lookback - 1 + }; + let max_start = first.max(query_start); + let min_end = useful_last.min(query_end); + if max_start > min_end { + return 0; + } + let step = i128::from(spec.step_ms); + let aligned_start = query_start + (max_start - query_start) / step * step; + let aligned_end = query_end - (query_end - min_end) / step * step; + let points = (aligned_end - aligned_start) / step + 1; + usize::try_from(points).unwrap_or(usize::MAX) + } + + /// The largest histogram a row of each field can carry, `None` for a field of floats. + /// + /// A row of native histograms is copied into the builder once per grid point that selects it, + /// so what such a field costs is one histogram per point: the largest row is the one to charge, + /// because any row can be the one every decided point picks. + fn largest_histograms(&self, closing: &RawSeriesChunk) -> Vec> { + let mut largest = vec![None; self.schema.fields.len()]; + for chunk in self.chunks.iter().chain(std::iter::once(closing)) { + for (slot, field) in largest.iter_mut().zip(chunk.fields.as_slice()) { + if let Some(bytes) = histogram_row_bytes(field) { + *slot = Some(slot.unwrap_or(0).max(bytes)); + } + } + } + largest + } + + /// Buffers one chunk of the series being selected, and gathers that series when it ends. + /// + /// `None` means the series is still open: the selection reads one whole axis, so nothing is + /// decided for a series until the chunk carrying `is_last` arrives. + fn push(&mut self, path: &NodePath, chunk: RawSeriesChunk) -> Result> { + if let Some(open) = self.chunks.first() + && (open.primary_key.as_slice() != chunk.primary_key.as_slice() + || open.metric != chunk.metric) + { + // One axis is gathered before the kernel reads any of it, so a chunk of another series + // has nowhere to go: it would be selected as though its samples were the open series'. + return Err(series_changed(path)); + } + let closing = chunk.is_last; + self.held = self.held.saturating_add(raw_bytes(&chunk)); + self.chunks.push(chunk); + if !closing { + return Ok(None); + } + self.gather(path).map(Some) + } + + /// Selects the instants of the buffered series, hands it over closed, and drops the buffer. + fn gather(&mut self, path: &NodePath) -> Result { + let chunk = self.select(path)?; + // The selection copied what it needed out of the buffered chunks, so a series that has been + // answered must not keep charging the pool for them. + self.chunks.clear(); + self.chunks.shrink_to_fit(); + self.held = 0; + Ok(chunk) + } + + /// The last sample at or before each grid point of the buffered series, as one chunk. + fn select(&mut self, path: &NodePath) -> Result { + let Some(first) = self.chunks.first() else { + return Err(DataFusionError::Execution(format!( + "the instant selection at {path} holds no series to select" + ))); + }; + let metric = first.metric.clone().ok_or_else(|| no_labels(path))?; + // The buffered chunks are one series, so they are read as the one axis the selection takes. + let time = buffered_time(&self.chunks)?; + let fields = buffered_fields(&self.chunks, self.schema.fields.len())?; + let grid = Arc::clone(&self.selection.output_grid); + let spec = grid.spec(); + // The grid's own points are what the selection decides, so the range and the step it is + // asked for are the grid's: `start..=start + (len - 1) * step`, step by step. + let (indices, timestamps) = select_instants( + &time, + &fields, + spec.start_ms..=grid_end_ms(spec), + spec.step_ms, + self.selection.lookback_ms, + self.selection.offset_ms, + )?; + self.output.begin_series(); + // One cursor per field, so a row is read as the field holds it rather than by walking the + // chunks of a field from its start once per row. + let mut cursors: Vec> = fields.iter().map(Samples::cursor).collect(); + // `GridSample` is not `Clone`, so the row is filled slot by slot rather than from one value. + let mut row: Vec> = (0..fields.len()).map(|_| None).collect(); + let mut read: Option = None; + for (index, timestamp) in indices.iter().zip(×tamps) { + let index = *index as usize; + // The selection steps the grid's range by its step, so every timestamp it reports is a + // point of the grid and the position is the point's index in it. + let position = grid_position(&grid, *timestamp)?; + // Two grid points can select one source row, and the row is read once for both. + if read != Some(index) { + for (slot, cursor) in row.iter_mut().zip(&mut cursors) { + let sample = cursor.at(index).ok_or_else(|| missing_row(path, index))?; + *slot = owned(sample)?; + } + read = Some(index); + } + // A position no field holds a value at is not a row: the builder drops it, which is + // what keeps a hole in the grid out of the result instead of padding it. + self.output.push_samples(position, &row)?; + } + // The rows the selection picked are the rows this series holds, and the whole series is + // decided at once, so this one chunk closes it. + match self.output.finish(metric.clone())? { + Some(series) => EvaluatedChunk::try_new(series, true), + None => no_row_chunk(&self.schema, metric), + } + } +} + +/// The buffered chunks as the one source axis they make up: their timestamps in arrival order. +/// +/// The axis type validates the result as strictly increasing, which is also what refuses a series +/// whose chunks do not continue each other. +fn buffered_time(chunks: &[RawSeriesChunk]) -> Result { + let len = chunks.iter().map(|chunk| chunk.time.len()).sum(); + let mut timestamps: Vec = Vec::with_capacity(len); + for chunk in chunks { + timestamps.extend_from_slice(chunk.time.as_buffer()); + } + RawTimestamps::try_new(ScalarBuffer::from(timestamps)).map_err(series_error) +} + +/// The buffered chunks as one axis: per field, the chunks' samples in arrival order. +/// +/// A field retains the chunks it arrived in instead of copying them out, so the axis costs one +/// descriptor per chunk and the selection reads the values where they already are. +fn buffered_fields(chunks: &[RawSeriesChunk], fields: usize) -> Result { + let mut per_field: Vec> = vec![Vec::with_capacity(chunks.len()); fields]; + for chunk in chunks { + for (field, slot) in chunk.fields.iter().zip(&mut per_field) { + slot.push(field.clone()); + } + } + let fields = per_field + .into_iter() + .map(|chunks| Samples::try_from_chunks(chunks).map_err(series_error)) + .collect::>>()?; + SeriesFields::try_from_vec(fields).map_err(series_error) +} + +/// The bytes the builder's table holds for one row of the field, `None` for a field of floats. +/// +/// The largest row of the field rather than its total: every grid point that selects one source row +/// copies that row, so the widest histogram the field holds is what a point can cost. +fn histogram_row_bytes(field: &Samples) -> Option { + let holds_histograms = field.segments().iter().any(|segment| { + matches!( + segment.dense(), + Some(DenseSamplesRef::Histogram(_) | DenseSamplesRef::Mixed(_)) + ) + }); + if !holds_histograms { + return None; + } + Some( + field + .iter() + .filter_map(|sample| match sample { + SampleRef::Histogram(view) => Some(histogram_bytes(&view)), + _ => None, + }) + .fold(0, usize::max), + ) +} + +/// The bytes the histogram table holds for one row of `view`. +/// +/// The row `HistogramBatchBuilder::push` writes: its scalar columns, the offset each of the seven +/// bucket lists pays for it, and the values those lists hold -- the custom bounds, and each side's +/// bucket counts with the span offsets and lengths that place them. The spans are the view's own, +/// because a histogram handed on unchanged keeps its layout through +/// [`crate::histogram_eval::float_histogram::FloatHistogram::push_to`]. +fn histogram_bytes(view: &HistogramView<'_>) -> usize { + // Schema, zero threshold, sum, count, zero count, reset hint and start timestamp, the narrow + // columns counted at the width of the widest. + let scalars = 7 * std::mem::size_of::(); + // One offset per list column: custom bounds, two span columns and one bucket column per side. + let list_offsets = 7 * std::mem::size_of::(); + let custom = view + .custom_values + .len() + .saturating_mul(std::mem::size_of::()); + let sides = [view.positive, view.negative] + .iter() + .map(|side| { + side.span_offsets + .len() + .saturating_mul(2) + .saturating_mul(std::mem::size_of::()) + .saturating_add( + side.buckets + .len() + .saturating_mul(std::mem::size_of::()), + ) + }) + .fold(0usize, usize::saturating_add); + scalars + .saturating_add(list_offsets) + .saturating_add(custom) + .saturating_add(sides) +} + +/// The last timestamp of `spec`, which is inclusive: its `len` points step from its start. +/// +/// [`GridSpec::try_new`] checked that the point fits `i64`, so widening it here cannot overflow. +fn grid_end_ms(spec: GridSpec) -> i64 { + (i128::from(spec.start_ms) + i128::from(spec.step_ms) * (i128::from(spec.len) - 1)) as i64 +} + +/// The position of the grid point at `timestamp`, which the selection reported as one of its own. +/// +/// The selection steps the range it is given by the step it is given, so every timestamp it reports +/// is `start_ms + position * step_ms` for one position of the grid: subtracting the start and +/// dividing by the step gives that position back. A timestamp off that lattice, or one past the end +/// of the grid, is a disagreement between the selection and its binding rather than a sample to +/// skip, so it is reported instead of dropped. +fn grid_position(grid: &EvalGrid, timestamp: i64) -> Result { + let spec = grid.spec(); + let offset = i128::from(timestamp) - i128::from(spec.start_ms); + let step = i128::from(spec.step_ms); + let position = offset.div_euclid(step); + if offset.rem_euclid(step) != 0 || position < 0 || position >= i128::from(spec.len) { + return Err(DataFusionError::Execution(format!( + "the instant selection reported the timestamp {timestamp}ms, which is not a point of \ + its grid {spec:?}" + ))); + } + Ok(position as usize) +} + +/// A closing chunk carrying `metric` and no row at all. +/// +/// A series whose selection picked no source row still closes exactly once, on a chunk that holds +/// nothing: the labels say which series closed, and the address says it closed before its first +/// point, as the empty chunk a window hands out for such a series does. +fn no_row_chunk(schema: &SeriesSchema, metric: MetricName) -> Result { + let grid = schema.eval_grid.clone().ok_or_else(|| { + DataFusionError::Execution( + "an instant selection produces an evaluated chunk, which is addressed on a grid" + .to_string(), + ) + })?; + let empty = Samples::from_float(Float64Array::from(Vec::::new())); + let fields = + SeriesFields::try_from_vec(vec![empty; schema.fields.len()]).map_err(series_error)?; + let time = GridRange::try_new(grid, 0, 0).map_err(series_error)?; + let series = + Series::try_new(schema, metric, TimeAxis::Grid(time), fields).map_err(series_error)?; + EvaluatedChunk::try_new(series, true) +} + /// The per-node run-time state of a node. struct NodeRuntime { buffer: NodeBuffer, @@ -846,6 +1290,9 @@ struct NodeRuntime { /// enters, and the figure it was queued with is given back as it leaves, so neither the queue /// nor the handover walks it. queued: VecDeque<(NativeChunk, usize)>, + /// What an instant node holds: the chunks of the series it is selecting, and, while it gathers + /// that series, the workspace the gather reserved for. `None` for every other kind of node. + instant: Option, /// A window node's evaluator, with its own input and output grid; `None` before it is built and /// once it is closed. window: Option, @@ -1029,6 +1476,95 @@ impl NodeRuntime { Ok(output) } + /// Feeds one chunk to this node's instant selection. + /// + /// `None` means the series this chunk belongs to has not ended: the selection reads one whole + /// axis, so it decides nothing until the chunk carrying `is_last` arrives. + fn instant(&mut self, path: &NodePath, chunk: NativeChunk) -> Result> { + let NativeChunk::Raw(chunk) = chunk else { + return Err(DataFusionError::Execution(format!( + "the instant selection at {path} consumes raw samples, not an evaluated chunk" + ))); + }; + let (positions, closing) = (chunk.time.len(), chunk.is_last); + // The chunk is held from here on: a selection keeps every chunk of a series until the one + // that closes it, so what this node reserves for is the series it has buffered. + self.buffer.grow(raw_bytes(&chunk))?; + // The chunk that closes the series gathers it in one go, and the gather allocates a + // workspace of its own beside the buffered chunks: the axis it concatenates, the descriptors + // that read it, the two arrays the selection returns and the rows the builder keeps. That + // workspace is reserved here, before it exists, so a pool too small for it refuses the + // gather instead of learning about it afterwards. Without this the reservation would fall + // back to the buffered chunks alone -- nothing at all, on the chunk that ends the series -- + // and a selection over a million grid points would take its tens of megabytes without + // asking. [`InstantRuntime::gather_bytes`] derives the figure, because the chunks it counts + // are that node's. + let Some(instant) = self.instant.as_ref() else { + return Err(DataFusionError::Execution(format!( + "the node at {path} holds no instant selection" + ))); + }; + let scratch = if closing { + instant.gather_bytes(&chunk) + } else { + 0 + }; + self.buffer.grow(scratch)?; + let metrics = Arc::clone(&self.buffer.metrics); + let outcome = { + // The timer covers this node's own work only: it is dropped before the driver polls + // the input again. + let _timer = metrics.timer(); + match self.instant.as_mut() { + Some(instant) => instant.push(path, chunk), + // Rejected when the state was built, since a selection reads raw samples; kept as + // an error rather than a panic so an impossible tree cannot take the process down. + None => Err(DataFusionError::Execution(format!( + "the node at {path} holds no instant selection" + ))), + } + }; + metrics.record_input(1, positions); + if closing { + // A series closes exactly once, on the chunk carrying `is_last`: the selection gathers + // the whole series on it, so it is counted here rather than on the chunk handed out. + metrics.record_completed_series(1); + } + // A failed gather gives its workspace back here rather than leaving it to the driver: the + // driver releases every node when the execution stops, but the workspace is this call's and + // not something the node keeps, so holding the pool's budget for it until the stopped + // executor is dropped would charge the pool for memory nothing reads any more. + let output = match outcome { + Ok(output) => output, + Err(error) => { + self.settle_buffered()?; + return Err(error); + } + }; + if let Some(output) = &output { + metrics.record_grid_points(output.series.time.len()); + metrics.record_output(1, valid_values(output)); + } + // What this node still holds is the chunks of the series it is selecting: a whole series is + // held until it ends, so the reservation is what those chunks hold rather than a figure + // that grows with the chunk count of anything already answered. The chunk just consumed and + // the workspace of the gather it closed are both gone, so this is what gives them back. + self.settle_buffered()?; + Ok(output) + } + + /// Sets this node's reservation to the chunks of the series it is still gathering. + /// + /// The chunk the node just consumed and the workspace of a gather it closed are not part of it: + /// the first has left the buffer and the second has been handed to the output. + fn settle_buffered(&mut self) -> Result<()> { + let buffered = self + .instant + .as_ref() + .map_or(0, InstantRuntime::buffered_bytes); + self.buffer.resize(buffered) + } + /// Keeps one fragment of this node's collected input, charged for as it arrives. /// /// The fragment is measured once, here, and its estimate stands until the flush that merges it @@ -1124,12 +1660,13 @@ impl NodeRuntime { /// Returns everything this node holds, including queued chunks. fn release(&mut self) { - // The window goes first: dropping the evaluator is what gives back the memory its - // reservation was taken for. Returning the reservation while the evaluator still held the - // samples would let the pool admit memory the process has not released, for as long as the - // caller keeps the stopped executor alive. The label sets this node handed out and the - // fragments it collected go with it. + // A node's own state goes first: dropping the evaluator and the buffered series is what + // gives back the memory their reservation was taken for. Returning the reservation while + // they still held the samples would let the pool admit memory the process has not released, + // for as long as the caller keeps the stopped executor alive. The label sets this node + // handed out and the fragments it collected go with it. self.window = None; + self.instant = None; self.emitted = None; self.merge = None; // A node that holds no chunk holds no series either, so nothing it handed over is still @@ -1213,6 +1750,7 @@ impl NativeExecutionState { source = Some((index, samples)); (NodeTag::Input, None) } + NativeExpr::Instant { input, .. } => (NodeTag::Instant, Some(input.as_ref())), NativeExpr::Window { input, .. } => (NodeTag::Window, Some(input.as_ref())), NativeExpr::Map { input, .. } => (NodeTag::Map, Some(input.as_ref())), // The kinds below are legal in a bound tree, and `validate` stays the authority on @@ -1220,7 +1758,6 @@ impl NativeExecutionState { // chunk is read: a tree the runtime cannot execute must not start, because a // partial answer would be one the caller never asked for. NativeExpr::Scalar(_) => return Err(not_implemented("Scalar", &path)), - NativeExpr::Instant { .. } => return Err(not_implemented("Instant", &path)), NativeExpr::Aggregate { .. } => return Err(not_implemented("Aggregate", &path)), NativeExpr::Binary { .. } => return Err(not_implemented("Binary", &path)), }; @@ -1228,6 +1765,7 @@ impl NativeExecutionState { let mut runtime = NodeRuntime { buffer, queued: VecDeque::new(), + instant: None, window: None, map: None, emitted: None, @@ -1236,6 +1774,19 @@ impl NativeExecutionState { }; match &node.expr { NativeExpr::Input(_) => {} + NativeExpr::Instant { input, op } => { + let input_schema = input.output.samples().ok_or_else(|| { + DataFusionError::Execution(format!( + "the instant selection at {path} consumes samples, but its input \ + produces none" + )) + })?; + runtime.instant = Some(InstantRuntime::try_new(input_schema, op)?); + // A selection keeps the labels of the series it reads, empty-valued ones + // included, so it is not one of the nodes that can turn two series into one + // label set and it keeps no table of what it handed out: the node that drops a + // label is the one that can collide, and it records its own. + } NativeExpr::Map { op, .. } => { let schema = node.output.samples().cloned().ok_or_else(|| { DataFusionError::Execution(format!( @@ -1271,7 +1822,6 @@ impl NativeExecutionState { runtime.emitted = Some(Emitted::default()); } NativeExpr::Scalar(_) - | NativeExpr::Instant { .. } | NativeExpr::Aggregate { .. } | NativeExpr::Binary { .. } => { // Rejected above, where the pass has the position to name. @@ -1321,6 +1871,17 @@ impl NativeExecutionState { nodes[index].input.is_some_and(|child| holds_shared[child]); match nodes[index].tag { NodeTag::Input => {} + NodeTag::Instant => { + // A selection keeps the labels it read, so it is not where two series + // become one label set and its own output carries its input's flag. Its + // input is the source -- `validate` binds a selection to raw samples, and + // no node above a source produces them -- and a source hands each series' + // chunks over together, so the flag is false there in every tree this + // runtime runs. Should the pass ever compute it true, the selection still + // needs no collected input: it reads one series at a time and refuses a + // chunk of another before it is buffered. + holds_shared[index] = input_holds_shared; + } NodeTag::Map => holds_shared[index] = true, NodeTag::Window => { let (drops_metric_name, input_schema) = match expr { @@ -1414,6 +1975,19 @@ impl NativeExecutionState { .count() } + /// How many nodes still hold the chunks of a series an instant selection was gathering. + /// + /// Test-only, and for the same reason as [`Self::nodes_holding_windows`]: the buffered chunks + /// have to be dropped, not only the bytes they were reserved for, and that is not observable + /// from the pool. + #[cfg(test)] + pub(crate) fn nodes_holding_instants(&self) -> usize { + self.runtimes + .iter() + .filter(|runtime| runtime.instant.is_some()) + .count() + } + /// The input that ended before its last chunk, if there is one. /// /// It is read before anything is merged: a window that collects its input closes every series @@ -1729,6 +2303,19 @@ impl NativeExecutor { } } } + NodeTag::Instant => { + let child = input.ok_or_else(|| missing_input(tag))?; + // A selection decides a whole series at once, so every chunk before the one that + // closes it yields nothing and the selection asks its input again. + loop { + let Some(chunk) = self.pull(child)? else { + return Ok(None); + }; + if let Some(output) = self.instant_node(index, chunk)? { + return Ok(Some(NativeChunk::Evaluated(output))); + } + } + } } } @@ -1752,6 +2339,13 @@ impl NativeExecutor { runtime.window(path, chunk) } + /// Feeds one chunk to the instant selection at `index`. + fn instant_node(&mut self, index: usize, chunk: NativeChunk) -> Result> { + let path = &self.state.paths[index]; + let runtime = &mut self.state.runtimes[index]; + runtime.instant(path, chunk) + } + /// Queues a source chunk at the tree's input node. fn push_source(&mut self, chunk: NativeChunk) -> Result<()> { self.check_running()?; @@ -1881,6 +2475,39 @@ fn closed(path: &NodePath) -> DataFusionError { DataFusionError::Execution(format!("the window at {path} is already closed")) } +/// The error for a chunk of another series arriving while one is being selected. +/// +/// [`select_instants`] reads one whole axis, so the buffered chunks have to be one series: a chunk +/// of another series has nowhere to go, and appending it would select its samples as though they +/// were the open series'. +fn series_changed(path: &NodePath) -> DataFusionError { + DataFusionError::Execution(format!( + "the instant selection at {path} combines one series into one axis, and a chunk of another \ + series arrived before the first one ended" + )) +} + +/// The error for a series an instant selection has no labels to report it under. +/// +/// A source bound in label mode carries its identity encoded and no decoded labels, and a chunk +/// that is handed on has to name the series it holds. +fn no_labels(path: &NodePath) -> DataFusionError { + DataFusionError::Execution(format!( + "the instant selection at {path} requires materialized series labels" + )) +} + +/// The error for a source row the selection named but the series does not hold. +/// +/// The selection returns indices into the axis it was given, so a row outside it means the two +/// disagree rather than that a sample is missing. +fn missing_row(path: &NodePath, index: usize) -> DataFusionError { + DataFusionError::Execution(format!( + "the instant selection at {path} selected the source row {index}, which the series it read \ + does not hold" + )) +} + /// The rejection of an input ordering this runtime cannot execute. /// /// It names the input that declares the ordering and the ordering found, so a reader can tell a diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index 829e79b8802..67e9f7466be 100644 --- a/src/promql/src/native/execute/tests.rs +++ b/src/promql/src/native/execute/tests.rs @@ -35,10 +35,11 @@ use datatypes::series::{Label, MetricName, SeriesOrdering}; use super::{ Emitted, NativeChunk, NativeExecutionState, NativeExecutor, NodeBuffer, NodeRuntime, - absorb_rebuilds, evaluated_bytes, map_chunk, raw_bytes, + absorb_rebuilds, evaluated_bytes, grid_end_ms, map_chunk, raw_bytes, }; use crate::evaluated::EvaluatedChunk; use crate::grid_series::merge_same_labelset; +use crate::instant_eval::select_instants; use crate::map::MathOp; use crate::native::metrics::{NodeMetric, NodeMetrics, NodePath}; use crate::native::plan::{ @@ -93,11 +94,16 @@ fn metric() -> MetricName { /// The raw source every supported tree starts from. fn input() -> NativeNode { + raw_input_with_ordering(SeriesOrdering::KeyThenTime) +} + +/// The same raw source, declaring `ordering`. +fn raw_input_with_ordering(ordering: SeriesOrdering) -> NativeNode { let output = NativeType::Samples(raw_schema()); NativeNode { expr: NativeExpr::Input(BoundInput { output: output.clone(), - ordering: SeriesOrdering::KeyThenTime, + ordering, }), output, } @@ -193,6 +199,39 @@ fn window_over_fields(count: usize, output_grid: Arc) -> NativeNode { } } +/// A bare selector over `inner`: the last sample at or before each grid point, read within +/// `lookback_ms` and at `offset_ms`. +/// +/// The selection reports one value per input field, so its layout is the input's, just as the +/// compiler builds it. +fn instant( + inner: NativeNode, + lookback_ms: i64, + offset_ms: i64, + output_grid: Arc, +) -> NativeNode { + let fields = inner + .output + .samples() + .expect("a selection reads samples") + .fields + .clone(); + NativeNode { + expr: NativeExpr::Instant { + input: Box::new(inner), + op: BoundInstant { + lookback_ms, + offset_ms, + output_grid: Arc::clone(&output_grid), + }, + }, + output: NativeType::Samples(Arc::new(SeriesSchema { + fields, + eval_grid: Some(output_grid), + })), + } +} + fn map(inner: NativeNode, ops: Vec) -> NativeNode { let output = inner.output.grid().expect("a map needs a grid").clone(); NativeNode { @@ -265,6 +304,34 @@ fn raw_chunk_repeated( .unwrap() } +/// One raw chunk of `series-a` whose fields carry `columns`, one column per field. +/// +/// A `None` is a position that field holds no sample at, which is what a partially filled field +/// looks like at the source. +fn raw_chunk_columns( + schema: &SeriesSchema, + timestamps: &[i64], + columns: Vec>>, + is_last: bool, +) -> RawSeriesChunk { + let fields = SeriesFields::try_from_vec( + columns + .into_iter() + .map(|values| Samples::from_float(Float64Array::from(values))) + .collect(), + ) + .unwrap(); + RawSeriesChunk::try_new( + schema, + Buffer::from_slice_ref(b"series-a"), + Some(metric()), + RawTimestamps::try_new(ScalarBuffer::from(timestamps.to_vec())).unwrap(), + fields, + is_last, + ) + .unwrap() +} + /// Pushes one native histogram whose positive side is one span of `buckets` buckets of one count /// into `builder`, and returns the index it was filed under. fn push_histogram(builder: &mut HistogramBatchBuilder, buckets: i32) -> u32 { @@ -320,6 +387,59 @@ fn raw_histogram_chunk( .unwrap() } +/// One raw chunk of `series-a` whose one field holds a four-bucket native histogram at each of +/// `timestamps`, each carrying `start_timestamp` as the payload it was read from reports it. +fn raw_histogram_chunk_with_start( + schema: &SeriesSchema, + timestamps: &[i64], + start_timestamp: Option, + is_last: bool, +) -> RawSeriesChunk { + let mut builder = HistogramBatchBuilder::new(); + let offsets = [0]; + let lengths = [4]; + let counts = vec![1.0; 4]; + let indices: UInt32Array = timestamps + .iter() + .map(|_| { + Some( + builder + .push(HistogramView { + schema: 0, + zero_threshold: 0.0, + zero_count: 0.0, + count: 4.0, + sum: 4.0, + reset_hint: CounterResetHint::Unknown, + start_timestamp, + custom_values: &[], + positive: BucketSideView { + span_offsets: &offsets, + span_lengths: &lengths, + buckets: &counts, + }, + negative: BucketSideView { + span_offsets: &[], + span_lengths: &[], + buckets: &[], + }, + }) + .unwrap(), + ) + }) + .collect(); + let samples = Samples::try_from_histogram(indices, builder.finish().unwrap()).unwrap(); + RawSeriesChunk::try_new( + schema, + Buffer::from_slice_ref(b"series-a"), + Some(metric()), + RawTimestamps::try_new(ScalarBuffer::from(timestamps.to_vec())).unwrap(), + SeriesFields::one(samples), + is_last, + ) + .unwrap() +} + /// One raw chunk of `series-a` whose one field holds the histogram at `index` of `table`. /// /// Two chunks built this way share one `Arc`, which is what makes the allocation @@ -354,6 +474,7 @@ fn source_runtime(pool: &Arc) -> NodeRuntime { pool, ), queued: VecDeque::new(), + instant: None, window: None, map: None, emitted: None, @@ -458,11 +579,16 @@ fn run(executor: &mut NativeExecutor) -> Vec { } fn sample_values(chunk: &EvaluatedChunk) -> Vec> { + field_values(chunk, 0) +} + +/// The values field `index` carries, one per row the chunk holds. +fn field_values(chunk: &EvaluatedChunk, index: usize) -> Vec> { let field = chunk .series .fields - .get(0) - .expect("these series have one field"); + .get(index) + .unwrap_or_else(|| panic!("the chunk carries no field {index}")); field .iter() .map(|sample| match sample { @@ -472,6 +598,25 @@ fn sample_values(chunk: &EvaluatedChunk) -> Vec> { .collect() } +/// The start timestamp of the histogram at every position of a chunk, in grid order. +/// +/// It is read off the output's own histogram views rather than off a converted value, because the +/// figure under test is the one a later reader sees. +fn histogram_starts(chunk: &EvaluatedChunk) -> Vec> { + let mut cursor = chunk.series.fields.as_slice()[0].cursor(); + chunk + .series + .time + .positions() + .filter_map( + |(_, row)| match cursor.at(row).expect("the axis and the fields are aligned") { + SampleRef::Histogram(view) => Some(view.start_timestamp), + _ => None, + }, + ) + .collect() +} + fn series_start(chunk: &EvaluatedChunk) -> usize { chunk .series @@ -828,23 +973,6 @@ fn an_unimplemented_node_is_reported_not_panicked() { let statement = || grid(0, 120_000, 2); let scalar_tree = scalar(1.0, statement()); - let instant_tree = { - let out = statement(); - map( - NativeNode { - expr: NativeExpr::Instant { - input: Box::new(input()), - op: BoundInstant { - lookback_ms: 300_000, - offset_ms: 0, - output_grid: Arc::clone(&out), - }, - }, - output: NativeType::Samples(evaluated_schema(out)), - }, - vec![MathOp::Abs], - ) - }; let aggregate_tree = { let out = statement(); NativeNode { @@ -879,7 +1007,6 @@ fn an_unimplemented_node_is_reported_not_panicked() { let cases = [ ("Scalar", "root", scalar_tree), - ("Instant", "root.input", instant_tree), ("Aggregate", "root", aggregate_tree), ("Binary", "root", binary_tree), ]; @@ -2464,6 +2591,68 @@ fn an_interleaved_input_is_refused_without_a_window() { assert!(message.contains("root.input"), "{message}"); } +/// A selection over an interleaved input is refused when the tree is bound, before it can run. +/// +/// The binding is where the ordering is known, and the plan's `Instant` arm refuses an input that +/// declares `Interleaved` for the reason the window arm does: a selection's consumer collects one +/// series' chunks into one buffer, so it reads a series to its last chunk before the next. Nothing +/// about an interleaved input is therefore left to the runtime along a selection, and the tree +/// never reaches [`NativeExecutionState::try_new`]. The ordering check there stays for the shape +/// the binding cannot see through: it only looks at a window's own input, so a window over a map +/// over the source still reads an interleaved stream, which is +/// [`a_window_over_an_interleaved_input_is_rejected`]. The shape with a map above the selection is +/// [`an_interleaved_input_is_refused_under_a_map_over_a_selection`]. +#[test] +fn an_interleaved_input_is_refused_under_a_selection() { + let out_grid = grid(0, 60_000, 2); + let error = NativeTree::try_new(instant( + raw_input_with_ordering(SeriesOrdering::Interleaved), + 60_000, + 0, + Arc::clone(&out_grid), + )) + .expect_err("a selection over an interleaved input must not bind"); + // A rejected tree is a compiler defect rather than a query error, so the binding reports it as + // `Internal`; what this pins is that the refusal names the ordering it came from, since the + // same check refuses a window over the same input. + assert!(matches!(error, DataFusionError::Internal(_)), "{error}"); + let message = error.to_string(); + assert!( + message.contains("interleaved chunks cannot be fed to it"), + "{message}" + ); +} + +/// The same refusal with a map above the selection. +/// +/// The tree here is `Input -> Instant -> Map`: the map sits above the selection, and the selection +/// reads the interleaved input directly. A map between the input and the selection is not a tree +/// this compiler can build at all -- a `Map` produces an evaluated vector while an instant +/// selection requires raw samples, and the plan's `Instant` arm rejects evaluated values -- so what +/// this test pins is the recursion: `validate` descends through the outer map to the selection +/// below it, whose own arm then refuses the input it reads, and the tree is refused at the same +/// point as [`an_interleaved_input_is_refused_under_a_selection`]. +#[test] +fn an_interleaved_input_is_refused_under_a_map_over_a_selection() { + let out_grid = grid(0, 60_000, 2); + let error = NativeTree::try_new(map( + instant( + raw_input_with_ordering(SeriesOrdering::Interleaved), + 60_000, + 0, + Arc::clone(&out_grid), + ), + vec![MathOp::Abs], + )) + .expect_err("a map over a selection over an interleaved input must not bind"); + assert!(matches!(error, DataFusionError::Internal(_)), "{error}"); + let message = error.to_string(); + assert!( + message.contains("interleaved chunks cannot be fed to it"), + "{message}" + ); +} + /// A series that produced values under a label set, followed by one that produced none under the /// same label set, is not a collision: the second contributes no sample, so the answer is just the /// first one's values. @@ -3186,3 +3375,651 @@ fn a_sparse_series_that_moves_forward_is_appended_to_its_history() { "the merged history is not sorted, disjoint and joined: {runs:?}" ); } + +/// A bare selector stamps the last sample at or before each grid point onto its own grid. +/// +/// The chunk is addressed on the node's output grid, not on the source's timestamps; a grid point +/// may select a row an earlier point already did (the point after the last sample repeats it); and +/// the same samples delivered as one chunk and as three answer the same, because the selection reads +/// a whole axis rather than a chunk. +#[test] +fn a_bare_selector_selects_instants_onto_its_own_grid() { + let out_grid = grid(0, 60_000, 4); + let tree = NativeTree::try_new(instant(input(), 120_000, 0, Arc::clone(&out_grid))).unwrap(); + let schema = raw_schema(); + let whole = raw_chunk( + &schema, + &[0, 30_000, 90_000, 120_000, 200_000], + &[1.0, 2.0, 3.0, 4.0, 5.0], + true, + ); + // The same series as three chunks, so nothing about it is decided by a chunk boundary. + let split = [ + raw_chunk(&schema, &[0, 30_000], &[1.0, 2.0], false), + raw_chunk(&schema, &[90_000], &[3.0], false), + raw_chunk(&schema, &[120_000, 200_000], &[4.0, 5.0], true), + ]; + + let mut once = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + once.push_raw(whole).unwrap(); + let out = run(&mut once); + + let mut piecewise = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + let mut in_pieces = Vec::new(); + for chunk in split { + let closing = chunk.is_last; + piecewise.push_raw(chunk).unwrap(); + // Until the series ends nothing is decided for it, so only the closing chunk reports. + let pulled = piecewise.next_chunk().unwrap(); + assert_eq!( + pulled.is_some(), + closing, + "a series reported a chunk before its last one arrived" + ); + in_pieces.extend(pulled); + } + in_pieces.extend(piecewise.finish().unwrap()); + + assert_eq!(out.len(), 1, "one series is one chunk, however it arrived"); + let chunk = &out[0]; + assert!(chunk.is_last, "the only chunk of a series closes it"); + assert!( + Arc::ptr_eq(chunk.series.time.grid().expect("a grid chunk"), &out_grid), + "the chunk is addressed on a grid that is not the selection's output grid" + ); + // 0 -> 0, 60s -> 30s, 120s -> 120s, 180s -> 120s: the row at 90s is never the last one before a + // grid point, and the point past the last sample repeats the row before it. + assert_eq!( + decided(&out), + vec![ + (0, Some(1.0)), + (1, Some(2.0)), + (2, Some(4.0)), + (3, Some(4.0)), + ] + ); + // A selection keeps the labels it read, the metric name included: unlike a range function it + // does not rewrite what the selector named. + assert_eq!(chunk.series.metric, metric()); + assert_eq!( + decided(&in_pieces), + decided(&out), + "the same series split into chunks answered differently" + ); +} + +/// The lookback edge is exclusive: a sample exactly `lookback_ms` before a grid point is not +/// selected, one millisecond later is. +/// +/// This pins what the selection kernel does rather than what Prometheus' documentation suggests: a +/// sample at the point itself is always selected, and the window is `(point - lookback, point]`. +#[test] +fn a_sample_at_the_lookback_edge_is_not_selected() { + // The statement's grid is the single point 1000ms, and the lookback is 200ms. + let out_grid = grid(1000, 100, 1); + let schema = raw_schema(); + let selected = |timestamp: i64| { + let tree = NativeTree::try_new(instant(input(), 200, 0, Arc::clone(&out_grid))).unwrap(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + execution + .push_raw(raw_chunk(&schema, &[timestamp], &[1.0], true)) + .unwrap(); + run(&mut execution) + }; + + // 800 is exactly `1000 - 200`, so the edge itself selects nothing. + let edge = selected(800); + assert_eq!(edge.len(), 1, "the series still closes on one chunk"); + assert!(edge[0].is_last); + assert!( + decided(&edge).is_empty(), + "a sample exactly at the lookback edge was selected: {:?}", + decided(&edge) + ); + // One millisecond inside the window is selected; a sample after the grid point never is. + assert_eq!(decided(&selected(801)), vec![(0, Some(1.0))]); + assert!(decided(&selected(1001)).is_empty()); +} + +/// A positive offset reads earlier samples, and a negative one later samples. +#[test] +fn an_offset_reads_a_different_sample_at_each_grid_point() { + let out_grid = grid(0, 60_000, 2); + let schema = raw_schema(); + let values = |offset_ms: i64| { + let tree = NativeTree::try_new(instant(input(), 600_000, offset_ms, Arc::clone(&out_grid))) + .unwrap(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + execution + .push_raw(raw_chunk( + &schema, + &[-60_000, 0, 60_000], + &[1.0, 2.0, 3.0], + true, + )) + .unwrap(); + decided(&run(&mut execution)) + }; + + assert_eq!( + values(0), + vec![(0, Some(2.0)), (1, Some(3.0))], + "without an offset a point reads the sample at or before it" + ); + assert_eq!( + values(60_000), + vec![(0, Some(1.0)), (1, Some(2.0))], + "a minute of offset did not read the samples a minute earlier" + ); + // A negative offset reads samples the statement's own range does not reach: both points read + // the sample one minute after them. + assert_eq!( + values(-60_000), + vec![(0, Some(3.0)), (1, Some(3.0))], + "a negative offset did not read the samples a minute later" + ); +} + +/// A series that selects nothing still closes, exactly once, on a chunk that holds no row. +#[test] +fn a_series_with_no_selectable_sample_closes_once() { + // Every sample carries the stale marker, which suppresses the selection at every point. + let stale = f64::from_bits(common_query::prometheus::PROMETHEUS_STALE_NAN_BITS); + let out_grid = grid(0, 60_000, 3); + let tree = NativeTree::try_new(instant(input(), 300_000, 0, Arc::clone(&out_grid))).unwrap(); + let set = ExecutionPlanMetricsSet::new(); + let mut execution = executor(&tree, pool(), &set); + let schema = raw_schema(); + execution + .push_raw(raw_chunk(&schema, &[0, 60_000], &[stale, stale], true)) + .unwrap(); + let out = run(&mut execution); + + assert_eq!(out.len(), 1, "the series has to close exactly once"); + let chunk = &out[0]; + assert!(chunk.is_last); + assert!( + chunk.series.time.is_empty(), + "the closing chunk of a series that selected nothing holds a row" + ); + assert_eq!( + series_start(chunk), + 0, + "the empty chunk is not addressed on the grid" + ); + assert!( + Arc::ptr_eq(chunk.series.time.grid().expect("a grid chunk"), &out_grid), + "the empty chunk is addressed on a grid that is not the selection's output grid" + ); + assert_eq!( + chunk.series.metric, + metric(), + "the series lost its identity" + ); + assert_eq!( + count(&set, &NodePath::root(), NodeMetric::COMPLETED_SERIES), + 1, + "the series did not close exactly once" + ); +} + +/// A row one field fills and another misses stays one row, with a null where the field misses it. +#[test] +fn a_selected_row_keeps_the_fields_that_miss_it_null() { + let out_grid = grid(0, 60_000, 2); + let tree = + NativeTree::try_new(instant(raw_input_of(2), 60_000, 0, Arc::clone(&out_grid))).unwrap(); + let schema = raw_schema_of(field_count(2)); + // Each field holds a value at one of the two rows and none at the other. + let chunk = raw_chunk_columns( + &schema, + &[0, 60_000], + vec![vec![Some(1.0), None], vec![None, Some(4.0)]], + true, + ); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + execution.push_raw(chunk).unwrap(); + let out = run(&mut execution); + + assert_eq!(out.len(), 1); + let chunk = &out[0]; + assert_eq!( + chunk.series.time.len(), + 2, + "a row no field of which holds a value was written down" + ); + assert_eq!(field_values(chunk, 0), vec![Some(1.0), None]); + assert_eq!(field_values(chunk, 1), vec![None, Some(4.0)]); + assert_eq!( + chunk + .series + .time + .positions() + .map(|(position, _)| position) + .collect::>(), + vec![0, 1] + ); +} + +/// A grid position nobody fills is not a row, and a hole does not close the series early. +/// +/// The samples lie at the first and the last point of the grid with nothing in between, and the +/// lookback is too short to carry either one across: two rows, two runs, and no row per position +/// between them. +#[test] +fn a_position_no_field_fills_is_not_a_row() { + let out_grid = grid(0, 60_000, 5); + let tree = NativeTree::try_new(instant(input(), 1_000, 0, Arc::clone(&out_grid))).unwrap(); + let schema = raw_schema(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + execution + .push_raw(raw_chunk(&schema, &[0], &[1.0], false)) + .unwrap(); + // The hole between the two samples decides nothing, and it does not end the series. + assert!( + execution.next_chunk().unwrap().is_none(), + "a hole in the grid closed the series" + ); + assert!( + execution.state().reserved_bytes() > 0, + "the open series is not reserved for" + ); + execution + .push_raw(raw_chunk(&schema, &[240_000], &[5.0], true)) + .unwrap(); + let out = run(&mut execution); + + assert_eq!(out.len(), 1); + let chunk = &out[0]; + assert!(chunk.is_last); + assert_eq!( + chunk + .series + .time + .positions() + .map(|(position, _)| position) + .collect::>(), + vec![0, 4], + "the positions nobody fills were written down as rows" + ); + assert_eq!(decided(&out), vec![(0, Some(1.0)), (4, Some(5.0))]); + assert!( + chunk.range().is_none(), + "a chunk of two runs of positions was addressed as one range" + ); +} + +/// An instant selection selects: the histogram it picks keeps its start timestamp. +/// +/// The sample is read, converted into the histogram type the builder writes back, and written out +/// again, and that type used to have no start timestamp to write: a sample carrying +/// `start_timestamp = Some(1500)` at 2000ms came out `None`. It matters downstream -- a reader +/// detects a counter reset from a start timestamp that falls inside the range it reads -- and it is +/// the difference between selecting a sample and computing a value. +/// +/// One source sample is selected by three grid points here, which is the case a per-row conversion +/// can get wrong twice: applied once and forgotten on the second use, or applied to a value that no +/// longer carries the timestamp. +#[test] +fn a_selected_histogram_keeps_its_start_timestamp() { + let out_grid = grid(2000, 1, 3); + let tree = NativeTree::try_new(instant(input(), 60_000, 0, Arc::clone(&out_grid))).unwrap(); + let schema = raw_schema(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + execution + .push_raw(raw_histogram_chunk_with_start( + &schema, + &[2000], + Some(1500), + true, + )) + .unwrap(); + let out = run(&mut execution); + + assert_eq!(out.len(), 1); + assert_eq!( + histogram_starts(&out[0]), + vec![Some(1500), Some(1500), Some(1500)], + "the selected sample lost the start timestamp it was read with" + ); +} + +/// Two series that differ only in an empty-valued label are two series, and both are answered. +/// +/// The label set the runtime's occupancy table compares drops empty values, so a node that recorded +/// what it handed out under one would refuse the second series. A selection keeps the labels it +/// read, empty values included, so it keeps them apart and records nothing. +#[test] +fn two_series_differing_only_in_an_empty_label_are_both_answered() { + let out_grid = grid(0, 60_000, 1); + let tree = NativeTree::try_new(instant(input(), 300_000, 0, Arc::clone(&out_grid))).unwrap(); + let schema = raw_schema(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + // `zone=""` against an absent `zone`: one label set to everything that drops empty values. + execution + .push_raw(raw_chunk_of( + &schema, + b"a", + host_metric(Some("")), + &[0], + &[1.0], + true, + )) + .unwrap(); + execution + .push_raw(raw_chunk_of( + &schema, + b"b", + host_metric(None), + &[0], + &[2.0], + true, + )) + .unwrap(); + let out = run(&mut execution); + + assert_eq!(out.len(), 2, "one of the two series was not answered"); + assert_eq!(out[0].series.metric, host_metric(Some(""))); + assert_eq!(out[1].series.metric, host_metric(None)); + assert_eq!(decided(&out[..1]), vec![(0, Some(1.0))]); + assert_eq!(decided(&out[1..]), vec![(0, Some(2.0))]); +} + +/// What a selection reserves is the series it has buffered, measured chunk by chunk. +/// +/// The node holds a whole series, so its reservation grows with the chunks that series arrived in +/// and goes back when the series is answered. Each chunk is measured once, with the same estimate a +/// queued chunk is charged with: an allocation two chunks share is counted once per chunk, which is +/// the over-count a figure that keeps no state across chunks accepts. +/// +/// What this checks is the buffer alone. The workspace the gather of that buffer allocates -- the +/// axis, the selection's two arrays and the rows the builder keeps -- is reserved for while it is +/// live and is not part of this figure, so this test cannot see it, however many bytes it is: +/// [`an_instant_selection_reserves_the_workspace_it_gathers`] is the one that does. +#[test] +fn an_instant_selection_reserves_the_series_it_buffers() { + let out_grid = grid(0, 60_000, 4); + let tree = NativeTree::try_new(instant(input(), 300_000, 0, Arc::clone(&out_grid))).unwrap(); + let schema = raw_schema(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + let mut previous = 0; + for timestamps in [vec![0, 60_000], vec![120_000], vec![180_000]] { + let values = vec![1.0; timestamps.len()]; + let chunk = raw_chunk(&schema, ×tamps, &values, false); + let bytes = raw_bytes(&chunk); + execution.push_raw(chunk).unwrap(); + assert!( + execution.next_chunk().unwrap().is_none(), + "a series was reported before its last chunk arrived" + ); + let held = execution.state().reserved_bytes(); + assert_eq!( + held, + previous + bytes, + "the buffered series is not what is reserved for: {held} bytes held after {previous} \ + and a chunk of {bytes}" + ); + previous = held; + } + execution + .push_raw(raw_chunk(&schema, &[240_000], &[2.0], true)) + .unwrap(); + let out = run(&mut execution); + assert_eq!(out.len(), 1); + assert_eq!( + execution.state().reserved_bytes(), + 0, + "the series that was answered is still reserved for" + ); +} + +/// The workspace the gather allocates is reserved for before it exists. +/// +/// A selection decides the points of the grid it can be selected at, so its workspace is +/// proportional to those points rather than to the samples of the series: the axis it concatenates, +/// the source row and evaluation timestamp it reports per point, and the rows the builder keeps. +/// The lookback here reaches from the sample to the last point, so this series is decididable at +/// every one of the million points -- the clipped count equals the grid's -- and the two selection +/// arrays alone are 16 MB while the built rows are more, against a series the node holds of one +/// sample of a few hundred bytes. [`a_selection_reserves_the_points_it_decides`] is the other side +/// of the same figure: a series whose samples cover only the start of a long grid reserves for the +/// few points it can decide and is admitted by a pool this one is refused by. +/// +/// The pool here holds 8 KiB, which is enough for the buffered chunk and nothing near the +/// workspace, so the gather is refused instead of being performed. Nothing else refuses a million +/// points, so this is the reservation and not a point limit answering: without it the gather simply +/// runs. +#[test] +fn an_instant_selection_reserves_the_workspace_it_gathers() { + let out_grid = grid(0, 1, 1_000_000); + let tree = NativeTree::try_new(instant(input(), 1_000_000, 0, out_grid)).unwrap(); + let pool = bounded_pool(8 * 1024); + let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new()); + let schema = raw_schema(); + execution + .push_raw(raw_chunk(&schema, &[0], &[1.0], true)) + .unwrap(); + + let error = execution.next_chunk().unwrap_err(); + assert!( + matches!(error, DataFusionError::ResourcesExhausted(_)), + "the gather's workspace was taken without asking the pool: {error}" + ); + assert!( + !error.to_string().contains("exceed limit"), + "a point limit answered in the pool's place: {error}" + ); + assert_eq!( + pool.reserved(), + 0, + "the stopped execution still holds bytes" + ); + assert_eq!( + execution.state().reserved_bytes(), + 0, + "the workspace of the refused gather is still reserved for" + ); + assert_eq!( + execution.state().nodes_holding_instants(), + 0, + "the stopped execution still holds the buffered series" + ); + // A refused gather stops the execution rather than leaving it to be resumed. + let again = execution.next_chunk().unwrap_err(); + assert!( + again.to_string().contains("stopped after an earlier error"), + "{again}" + ); +} + +/// A selection reserves the points it can decide, not the points the grid plans. +/// +/// The review's shape: a seven-day grid at a one-minute step is 10,081 points, and one sample at +/// the start of the series read with a five-minute lookback is decided at five of them. The +/// grid-scaled terms of `InstantRuntime::gather_bytes` at the planned grid -- the selection's two +/// arrays and the doubled builder columns -- are about 504 KB, which the 64 KiB pool here is far +/// below; the handful of points really decided need a few hundred bytes, so this gather has to be +/// admitted. What it hands out is the rows `select_instants` reports for the same series, which is +/// the count the reservation has to agree with. +/// +/// The second half is the other direction: the same grid, the same pool, and a series that spans +/// the grid with a lookback long enough to cover it, so all 10,081 points can be decided. The +/// figure then exceeds the pool and the gather is refused, which is what keeps this from being a +/// reservation of nothing. +#[test] +fn a_selection_reserves_the_points_it_decides() { + let out_grid = grid(0, 60_000, 10_081); + let schema = raw_schema(); + let budget = 64 << 10; + // The two grid-scaled terms of `gather_bytes` at the planned grid: 10,081 points x 16 bytes for + // the selection's arrays, and 10,081 points x (8 + 8 + 1) x 2 for the doubled builder columns. + let full_grid = 10_081 * (8 + 8) + 10_081 * (8 + 8 + 1) * 2; + assert!( + full_grid > budget, + "the pool is not below what the full grid would be reserved: {full_grid} bytes" + ); + + // One sample at the start, read within five minutes: the kernel reports 0s..240s, and the + // candidate bound it sizes its arrays with is 0s..300s. + let tree = NativeTree::try_new(instant(input(), 300_000, 0, Arc::clone(&out_grid))).unwrap(); + let chunk = raw_chunk(&schema, &[0], &[1.0], true); + let (_, timestamps) = select_instants( + &chunk.time, + &chunk.fields, + out_grid.spec().start_ms..=grid_end_ms(out_grid.spec()), + out_grid.spec().step_ms, + 300_000, + 0, + ) + .unwrap(); + assert_eq!( + timestamps.len(), + 5, + "the clipping kernel decides another count" + ); + let mut execution = executor(&tree, bounded_pool(budget), &ExecutionPlanMetricsSet::new()); + execution.push_raw(chunk).unwrap(); + let out = run(&mut execution); + assert_eq!(out.len(), 1); + let expected: Vec<(usize, Option)> = timestamps + .iter() + .map(|timestamp| { + ( + usize::try_from((timestamp - out_grid.spec().start_ms) / out_grid.spec().step_ms) + .unwrap(), + Some(1.0), + ) + }) + .collect(); + assert_eq!( + decided(&out), + expected, + "the gather handed out another set of points than the clipping decides" + ); + + // The same series spans the grid with a lookback that covers it, so every point can be decided + // and the workspace no longer fits the same pool. + let tree = + NativeTree::try_new(instant(input(), 604_800_001, 0, Arc::clone(&out_grid))).unwrap(); + let pool = bounded_pool(budget); + let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new()); + execution + .push_raw(raw_chunk(&schema, &[0, 604_800_000], &[1.0, 2.0], true)) + .unwrap(); + let error = execution.next_chunk().unwrap_err(); + assert!( + matches!(error, DataFusionError::ResourcesExhausted(_)), + "a series covering the grid was gathered without asking the pool: {error}" + ); + assert_eq!(pool.reserved(), 0); + assert_eq!(execution.state().reserved_bytes(), 0); +} + +/// A grid longer than any point limit is charged for, and the pool is what refuses it. +/// +/// `gather_bytes` used to return zero for a grid this long, so that a point limit inside the +/// kernel stayed the error reported and the workspace of a long gather was never reserved. The +/// estimate now applies to every gather, so the pool is asked before the kernel is and its refusal +/// is the one the caller sees — a bound in bytes, on the path that decides the points, rather than +/// a count that says nothing about what they cost. +#[test] +fn an_oversized_selection_is_refused_by_the_pool() { + // One million and two points over two samples: longer than the short circuit used to allow, + // so the estimate has to cover it rather than return zero. + let out_grid = grid(0, 1, 1_000_002); + let tree = NativeTree::try_new(instant(input(), 1_000_000, 0, out_grid)).unwrap(); + let pool = bounded_pool(8 * 1024); + let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new()); + let schema = raw_schema(); + execution + .push_raw(raw_chunk(&schema, &[0, 1_000_001], &[1.0, 2.0], true)) + .unwrap(); + + let error = execution.next_chunk().unwrap_err(); + assert!( + matches!(error, DataFusionError::ResourcesExhausted(_)), + "the gather's workspace was taken without asking the pool: {error}" + ); + assert!( + !error.to_string().contains("exceed limit"), + "a point limit answered in the pool's place: {error}" + ); + // Nothing is dropped silently: the execution stops and gives its reservation back. + assert_eq!(pool.reserved(), 0); + assert_eq!(execution.state().reserved_bytes(), 0); + assert_eq!(execution.state().nodes_holding_instants(), 0); + let again = execution.next_chunk().unwrap_err(); + assert!( + again.to_string().contains("stopped after an earlier error"), + "{again}" + ); +} + +/// An input that ends inside a series is an error, not an empty answer. +/// +/// The selection holds the series and gathers it only on the chunk that closes it, so nothing is +/// answered for it. The node that notices is the input: the chunk it handed over did not say +/// `is_last`, and the tree's close reports that before any collected input is merged. +#[test] +fn an_input_that_ends_inside_a_series_is_reported() { + let tree = NativeTree::try_new(instant(input(), 300_000, 0, grid(0, 60_000, 2))).unwrap(); + let pool = pool(); + let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new()); + let schema = raw_schema(); + execution + .push_raw(raw_chunk(&schema, &[0], &[1.0], false)) + .unwrap(); + assert!(execution.next_chunk().unwrap().is_none()); + + let error = execution.finish().unwrap_err(); + assert!( + error.to_string().contains("ended before the last chunk"), + "{error}" + ); + assert_eq!(pool.reserved(), 0); + assert_eq!(execution.state().reserved_bytes(), 0); + assert_eq!( + execution.state().nodes_holding_instants(), + 0, + "the stopped execution still holds the buffered series" + ); +} + +/// A selection combines one series into one axis, so a chunk of another series is refused. +/// +/// The buffered chunks are read as one axis, so a chunk of a second series has nowhere to go: it +/// would be selected as though its samples were the open series'. +#[test] +fn a_chunk_of_another_series_is_refused() { + let tree = NativeTree::try_new(instant(input(), 300_000, 0, grid(0, 60_000, 2))).unwrap(); + let schema = raw_schema(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + execution + .push_raw(raw_chunk(&schema, &[0], &[1.0], false)) + .unwrap(); + assert!(execution.next_chunk().unwrap().is_none()); + // The second chunk carries another series' key. + execution + .push_raw(raw_chunk_of( + &schema, + b"series-b", + metric(), + &[60_000], + &[2.0], + true, + )) + .unwrap(); + + let error = execution.next_chunk().unwrap_err(); + assert!( + error + .to_string() + .contains("a chunk of another series arrived"), + "{error}" + ); + assert_eq!( + execution.state().reserved_bytes(), + 0, + "the refused series is still reserved for" + ); +} diff --git a/src/promql/src/range_eval/tests.rs b/src/promql/src/range_eval/tests.rs index 1e5e6488a71..56c830841a0 100644 --- a/src/promql/src/range_eval/tests.rs +++ b/src/promql/src/range_eval/tests.rs @@ -1010,6 +1010,180 @@ fn histogram_chunk(schema: &SeriesSchema, buckets: i32, ts: i64) -> RawSeriesChu .unwrap() } +/// The single histogram sample of [`histogram_chunk`], read with `start_timestamp`. +/// +/// A sample carries the start timestamp a scan read it with, and it is payload rather than a +/// property of the value. +fn histogram_chunk_from( + schema: &SeriesSchema, + buckets: i32, + ts: i64, + start_timestamp: Option, + is_last: bool, +) -> RawSeriesChunk { + let offsets = [0]; + let lengths = [buckets]; + let counts = vec![1.0; usize::try_from(buckets).unwrap()]; + let mut builder = HistogramBatchBuilder::new(); + let index = builder + .push(HistogramView { + schema: 0, + zero_threshold: 0.0, + zero_count: 0.0, + count: f64::from(buckets), + sum: f64::from(buckets), + reset_hint: CounterResetHint::Unknown, + start_timestamp, + custom_values: &[], + positive: BucketSideView { + span_offsets: &offsets, + span_lengths: &lengths, + buckets: &counts, + }, + negative: BucketSideView { + span_offsets: &[], + span_lengths: &[], + buckets: &[], + }, + }) + .unwrap(); + let samples = Samples::try_from_histogram( + UInt32Array::from(vec![Some(index)]), + builder.finish().unwrap(), + ) + .unwrap(); + RawSeriesChunk::try_new( + schema, + Buffer::from(0u64.to_le_bytes().to_vec()), + Some(metric(0)), + RawTimestamps::try_new(ScalarBuffer::from(vec![ts])).unwrap(), + SeriesFields::try_from_vec(vec![samples]).unwrap(), + is_last, + ) + .unwrap() +} + +/// A function that hands back the sample it selected hands back that sample's payload. +/// +/// `last_over_time` and `first_over_time` select a sample instead of computing one, so the +/// histogram that leaves them is the histogram that arrived, start timestamp included. The window +/// here holds two samples that differ in bucket count and in start timestamp, so the count and the +/// timestamp together say which one was selected rather than only that some timestamp survived: +/// the last sample for `last_over_time`, the first one for `first_over_time`. What this pins is +/// that the selection passes the stored payload through; it does not pin a Prometheus result, whose +/// histogram values carry no such field, and it does not claim that a native rate reads the field +/// afterwards. A histogram computed from samples carries none, as +/// `histogram_eval::tests::a_start_timestamp_belongs_to_a_read_histogram_and_not_to_a_computed_one` +/// pins for the operations that build one. +#[test] +fn a_selected_histogram_keeps_its_start_timestamp() { + let schema = raw_schema(1); + let (first_start, last_start): (i64, i64) = (1_500, 2_500); + // One bucket at 0ms and two at 5000ms, each read with its own start timestamp. Both samples + // are inside the window of the second grid point; the first point is decided on the first + // chunk alone, before the second sample has arrived. + let chunks = [ + histogram_chunk_from(&schema, 1, 0, Some(first_start), false), + histogram_chunk_from(&schema, 2, 5_000, Some(last_start), true), + ]; + // The histogram of every row the function hands out, read off the output's own views: the + // bucket count tells the selected sample apart from any other, the start timestamp is the + // payload this test is about. + let selected = |function: RangeFunction| -> Vec<(usize, f64, Option)> { + let mut evaluator = RangeEvaluator::try_new( + &schema, + grid(0, 10_000, 2), + RangeWindow { + range_ms: 20_000, + offset_ms: 0, + anchor_ms: None, + }, + function, + ) + .unwrap(); + let out = collect_chunks(&mut evaluator, &chunks); + assert!(!out.is_empty(), "{function:?} answered no chunk"); + let mut picked = Vec::new(); + for chunk in &out { + let mut cursor = chunk.series.fields.as_slice()[0].cursor(); + for (position, row) in chunk.series.time.positions() { + match cursor.at(row).expect("the axis and the fields are aligned") { + SampleRef::Histogram(view) => { + picked.push((position, view.count, view.start_timestamp)); + } + SampleRef::Float(_) => panic!("{function:?} answered a float for a histogram"), + SampleRef::Missing => { + panic!("{function:?} handed out a row of its own axis without a sample") + } + } + } + } + picked + }; + + assert_eq!( + selected(RangeFunction::LastOverTime), + [(0, 1.0, Some(first_start)), (1, 2.0, Some(last_start))], + "last_over_time did not hand back the sample it selected" + ); + assert_eq!( + selected(RangeFunction::FirstOverTime), + [(0, 1.0, Some(first_start)), (1, 1.0, Some(first_start))], + "first_over_time did not hand back the sample it selected" + ); +} + +/// A histogram computed from a window carries no start timestamp, even when the window holds one. +/// +/// `sum_over_time` computes a histogram rather than selecting one, so the timestamp the sample was +/// read with is not part of the result -- the rule the histogram operations themselves follow, as +/// `histogram_eval::tests::a_start_timestamp_belongs_to_a_read_histogram_and_not_to_a_computed_one` +/// pins. The fold clears the field while adding, so a window holding two histograms already loses +/// it; a window holding one must not depend on that, or whether a result carries the field would +/// say how many samples the window held rather than what the function did. +#[test] +fn a_computed_histogram_drops_its_start_timestamp() { + let schema = raw_schema(1); + let start: i64 = 1_500; + // One histogram sample inside the window of both grid points, so the fold adds nothing to it + // and the single input is the whole result. + let chunks = [histogram_chunk_from(&schema, 1, 0, Some(start), true)]; + let mut evaluator = RangeEvaluator::try_new( + &schema, + grid(0, 10_000, 2), + RangeWindow { + range_ms: 20_000, + offset_ms: 0, + anchor_ms: None, + }, + RangeFunction::SumOverTime, + ) + .unwrap(); + let out = collect_chunks(&mut evaluator, &chunks); + let mut computed = Vec::new(); + for chunk in &out { + let mut cursor = chunk.series.fields.as_slice()[0].cursor(); + for (position, row) in chunk.series.time.positions() { + match cursor.at(row).expect("the axis and the fields are aligned") { + SampleRef::Histogram(view) => { + computed.push((position, view.count, view.start_timestamp)); + } + SampleRef::Float(_) => panic!("sum_over_time answered a float for a histogram"), + SampleRef::Missing => { + panic!("sum_over_time handed out a row of its own axis without a sample") + } + } + } + } + // The sum of one histogram is that histogram, so the count says the sample was summed rather + // than passed over; only the start timestamp has to be gone. + assert_eq!( + computed, + [(0, 1.0, None), (1, 1.0, None)], + "sum_over_time kept the start timestamp of the histogram it summed" + ); +} + /// A window retaining a native histogram is charged for the payload it owns, not only the slot. /// /// The histogram sample below stays ahead of the second grid point's window, so the evaluator keeps diff --git a/src/promql/src/select/tests.rs b/src/promql/src/select/tests.rs index 0cc2763bf64..3bac6739214 100644 --- a/src/promql/src/select/tests.rs +++ b/src/promql/src/select/tests.rs @@ -269,6 +269,67 @@ fn limit_operators_keep_histogram_samples() { assert_eq!(kinds(&kept[0], 0), [(1, "histogram")]); } +/// A selection hands back the sample it selected, start timestamp included. +/// +/// `limitk` and `limit_ratio` choose samples rather than compute them, so a histogram that arrives +/// with a start timestamp leaves with it. The count and the sum are compared too, so a computed +/// value that happened to carry a timestamp would not pass for the sample selected. What this pins +/// is that the selection passes the stored payload through, not a Prometheus result and not a +/// downstream rate that reads the field. A histogram computed from samples carries none, as +/// `histogram_eval::tests::a_start_timestamp_belongs_to_a_read_histogram_and_not_to_a_computed_one` +/// pins for the operations that build one. +#[test] +fn limit_operators_keep_the_start_timestamp_of_a_selected_sample() { + let histogram = + OwnedHistogram::parse("0/0/0/3ff0000000000000/3ff0000000000000/0/1_1/3ff0000000000000///"); + let start = 1_500; + // The sample a scan read, which is what a selection hands on. + let stamped = || { + let mut read = FloatHistogram::try_from_view(&histogram.view()).unwrap(); + read.start_timestamp = Some(start); + GridSample::Histogram(Box::new(read)) + }; + // What a limit operator keeps out of one series: the count and sum of the histogram it kept, + // so the sample selected is named, and the start timestamp it kept with it. + let kept = |series: &Series| -> Vec<(usize, f64, f64, Option)> { + let samples = &series.fields.as_slice()[0]; + series + .time + .positions() + .filter_map(|(point, row)| match samples.sample(row) { + SampleRef::Histogram(view) => { + Some((point, view.count, view.sum, view.start_timestamp)) + } + _ => None, + }) + .collect() + }; + let source = histogram.view(); + let expected = |point: usize| (point, source.count, source.sum, Some(start)); + + let schema = schema(3, 1); + let mut builder = GridSeriesBuilder::try_new(Arc::new(schema.clone())).unwrap(); + builder.push_samples(0, &[Some(stamped())]).unwrap(); + builder.push_samples(2, &[Some(stamped())]).unwrap(); + let series = builder.finish(metric("__name__=a")).unwrap().unwrap(); + + let by_ratio = limit_ratio(1.0, &schema, std::slice::from_ref(&series)).unwrap(); + assert_eq!(by_ratio.len(), 1); + assert_eq!( + kept(&by_ratio[0]), + [expected(0), expected(2)], + "limit_ratio did not hand back the samples it selected" + ); + + let by_count = limitk(1.0, &Grouping::By(Vec::new()), &schema, &[series]).unwrap(); + assert_eq!(by_count.len(), 1); + assert_eq!( + kept(&by_count[0]), + [expected(0), expected(2)], + "limitk did not hand back the samples it selected" + ); +} + fn grouping(kind: &str, labels: &str) -> Grouping { match kind { "without" => Grouping::Without(names(labels)), diff --git a/src/promql/src/vector_match/tests.rs b/src/promql/src/vector_match/tests.rs index 3fac9bfb07e..9b830462c8b 100644 --- a/src/promql/src/vector_match/tests.rs +++ b/src/promql/src/vector_match/tests.rs @@ -26,11 +26,11 @@ use super::{ BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels, SetOp, binary_vectors, set_vectors, }; -use crate::grid_series::GridSeriesBuilder; use crate::grid_series::reference::{ Dense, OwnedHistogram, Points, assert_histogram_close, assert_result_eq, golden_cases, mixed_series, piece_ranges, random_dense, run_ranges, sparse_series, }; +use crate::grid_series::{GridSample, GridSeriesBuilder}; use crate::histogram_eval::float_histogram::FloatHistogram; /// Labels with `__name__` as an ordinary entry, the way Prometheus holds them. @@ -1487,3 +1487,248 @@ fn set_operators_of_sparse_operands_keep_their_own_positions() { expected[9] = vec![Some(90.0)]; assert_result_eq(Some(&result[0]), &expected, &lhs.metric); } + +/// A comparison filters, so the histogram it keeps is the left operand's, payload included. +/// +/// A non-`bool` `==` or `!=` between two histograms keeps the left sample where the comparison +/// holds, so the histogram that leaves is the one that arrived on the left, start timestamp +/// included, and never the right operand's timestamp. `==` holds only for equal values, so that +/// pair differs in reset hint and start timestamp alone; `!=` holds for differing values, so its +/// pair differs in count and sum as well. What this pins is that a comparison passes the stored +/// payload through, not a Prometheus result. A `bool` comparison answers in floats, so nothing it +/// hands out is a histogram and no start timestamp leaves it. +#[test] +fn a_histogram_comparison_filter_keeps_the_payload_of_the_operand_it_keeps() { + let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 1_000, 3).unwrap()).unwrap()); + let schema = Arc::new(schema(1, &grid)); + // The start timestamp each operand was read with: written into its samples by `operand` below + // and repeated in every expectation, never read back through the operators under test. + const LEFT_START: i64 = 1_500; + const RIGHT_START: i64 = 2_500; + // count 4, sum 6, one span of two buckets holding 1 and 3, reset hint 2 (`NotCounterReset`). + let left = "0/0/0/4010000000000000/4018000000000000/2/0_2/3ff0000000000000,4008000000000000///"; + // The same value with reset hint 1 (`CounterReset`): `==` ignores the hint, so the pair is kept + // and only the hint and the start timestamp say which side survived. + let equal = + "0/0/0/4010000000000000/4018000000000000/1/0_2/3ff0000000000000,4008000000000000///"; + // count 5, sum 8: a value that differs, which is what `!=` keeps. + let other = + "0/0/0/4014000000000000/4020000000000000/2/0_2/4000000000000000,4008000000000000///"; + let operand = |name: &str, job: &str, text: &str, start: i64| -> Series { + let labels = Labels::from([ + ("__name__".to_string(), name.to_string()), + ("job".to_string(), job.to_string()), + ]); + let histogram = OwnedHistogram::parse(text); + let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap(); + for point in [0, 2] { + let mut read = FloatHistogram::try_from_view(&histogram.view()).unwrap(); + read.start_timestamp = Some(start); + builder + .push_samples(point, &[Some(GridSample::Histogram(Box::new(read)))]) + .unwrap(); + } + builder.finish(metric(&labels)).unwrap().unwrap() + }; + // Two pairs, one match group each: `job=eq` holds histograms `==` keeps, `job=ne` ones `!=` + // keeps, so each comparison passes exactly one pair and its result is named by that pair. + let lhs = [ + operand("eq", "eq", left, LEFT_START), + operand("ne", "ne", left, LEFT_START), + ]; + let rhs = [ + operand("eq_right", "eq", equal, RIGHT_START), + operand("ne_right", "ne", other, RIGHT_START), + ]; + // Every histogram a comparison handed out, by the series and point it sits at, with the reset + // hint as the number the fixtures wrote. + let kept = |series: &[Series]| -> Vec<(String, usize, f64, f64, Option, i32)> { + let mut kept: Vec<(String, usize, f64, f64, Option, i32)> = Vec::new(); + for series in series { + let name = series.metric.name().unwrap_or_default().to_string(); + let mut cursor = series.fields.as_slice()[0].cursor(); + for (point, row) in series.time.positions() { + match cursor.at(row).expect("the axis and the fields are aligned") { + SampleRef::Histogram(view) => kept.push(( + name.clone(), + point, + view.count, + view.sum, + view.start_timestamp, + i32::from(view.reset_hint), + )), + _ => panic!("a comparison handed out a sample that is not a histogram"), + } + } + } + kept.sort_by(|a, b| (a.0.as_str(), a.1).cmp(&(b.0.as_str(), b.1))); + kept + }; + let matching = |op: BinaryOp| BinaryMatching { + op, + return_bool: false, + labels: MatchingLabels::Ignoring(Vec::new()), + cardinality: Cardinality::OneToOne, + fill: FillValues::default(), + }; + + let eq = binary_vectors(&matching(BinaryOp::Eq), &schema, &lhs, &schema, &rhs).unwrap(); + assert_eq!( + kept(&eq), + [ + ("eq".to_string(), 0, 4.0, 6.0, Some(LEFT_START), 2), + ("eq".to_string(), 2, 4.0, 6.0, Some(LEFT_START), 2), + ], + "`==` did not hand back the left histogram it kept: {eq:?}" + ); + + let ne = binary_vectors(&matching(BinaryOp::Ne), &schema, &lhs, &schema, &rhs).unwrap(); + assert_eq!( + kept(&ne), + [ + ("ne".to_string(), 0, 4.0, 6.0, Some(LEFT_START), 2), + ("ne".to_string(), 2, 4.0, 6.0, Some(LEFT_START), 2), + ], + "`!=` did not hand back the left histogram it kept: {ne:?}" + ); + + // A `bool` comparison answers in floats, so nothing it hands out is a histogram and no start + // timestamp leaves it: `job=eq` holds equal histograms, so it answers 1, and `job=ne` differs + // in count and sum, so it answers 0. + let bool_matching = BinaryMatching { + return_bool: true, + ..matching(BinaryOp::Eq) + }; + let compared = binary_vectors(&bool_matching, &schema, &lhs, &schema, &rhs).unwrap(); + let answers = |series: &[Series]| -> Vec<(String, usize, f64)> { + let mut answers: Vec<(String, usize, f64)> = Vec::new(); + for series in series { + let job = series.metric.get("job").unwrap_or_default().to_string(); + let mut cursor = series.fields.as_slice()[0].cursor(); + for (point, row) in series.time.positions() { + match cursor.at(row).expect("the axis and the fields are aligned") { + SampleRef::Float(value) => answers.push((job.clone(), point, value)), + _ => panic!("a `bool` comparison handed out a sample that is not a float"), + } + } + } + answers.sort_by(|a, b| (a.0.as_str(), a.1).cmp(&(b.0.as_str(), b.1))); + answers + }; + assert_eq!( + answers(&compared), + [ + ("eq".to_string(), 0, 1.0), + ("eq".to_string(), 2, 1.0), + ("ne".to_string(), 0, 0.0), + ("ne".to_string(), 2, 0.0), + ], + "a `bool` comparison of histograms did not answer in floats: {compared:?}" + ); +} + +/// A set operation hands back the sample it kept, start timestamp included. +/// +/// `and`, `unless` and `or` choose samples rather than compute them, so each histogram that leaves +/// them is the histogram that arrived, payload included. The two operands were read with different +/// start timestamps and hold different counts and sums, so the count, the sum and the timestamp +/// together name the side and the point a kept sample came from. The one exception is `or` at a +/// point where the left side holds no sample: the sample kept is the right side's, and it carries +/// the right side's own start timestamp. What this pins is that the operators pass the stored +/// payload through, not a Prometheus result. +#[test] +fn a_set_operation_keeps_the_payload_of_the_sample_it_keeps() { + let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 1_000, 4).unwrap()).unwrap()); + let schema = Arc::new(schema(1, &grid)); + // The start timestamp each operand was read with: written into its samples by `operand` below + // and repeated in every expectation, never read back through the operators under test. + const LHS_START: i64 = 1_500; + const RHS_START: i64 = 2_500; + // count 4, sum 6, one span of two buckets holding 1 and 3. + let lhs_text = + "0/0/0/4010000000000000/4018000000000000/2/0_2/3ff0000000000000,4008000000000000///"; + // count 5, sum 8, so the count and the sum name the side a kept sample came from as well. + let rhs_text = + "0/0/0/4014000000000000/4020000000000000/2/0_2/4000000000000000,4008000000000000///"; + let operand = |name: &str, text: &str, start: i64, points: &[usize]| -> Series { + let labels = Labels::from([ + ("__name__".to_string(), name.to_string()), + ("job".to_string(), "x".to_string()), + ]); + let histogram = OwnedHistogram::parse(text); + let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap(); + for &point in points { + let mut read = FloatHistogram::try_from_view(&histogram.view()).unwrap(); + read.start_timestamp = Some(start); + builder + .push_samples(point, &[Some(GridSample::Histogram(Box::new(read)))]) + .unwrap(); + } + builder.finish(metric(&labels)).unwrap().unwrap() + }; + // The left side holds 0 and 1, the right side 0 and 2, so each operator keeps a different set: + // the shared point for `and`, the left-only point for `unless`, and the union for `or`. + let lhs = operand("l", lhs_text, LHS_START, &[0, 1]); + let rhs = operand("r", rhs_text, RHS_START, &[0, 2]); + // Every histogram an operator handed out, by the series and point it sits at. + let kept = |series: &[Series]| -> Vec<(String, usize, f64, f64, Option)> { + let mut kept: Vec<(String, usize, f64, f64, Option)> = Vec::new(); + for series in series { + let name = series.metric.name().unwrap_or_default().to_string(); + let mut cursor = series.fields.as_slice()[0].cursor(); + for (point, row) in series.time.positions() { + match cursor.at(row).expect("the axis and the fields are aligned") { + SampleRef::Histogram(view) => kept.push(( + name.clone(), + point, + view.count, + view.sum, + view.start_timestamp, + )), + _ => panic!("a set operator handed out a sample that is not a histogram"), + } + } + } + kept.sort_by(|a, b| (a.0.as_str(), a.1).cmp(&(b.0.as_str(), b.1))); + kept + }; + let labels = MatchingLabels::Ignoring(Vec::new()); + let eval = |op: SetOp| { + set_vectors( + op, + &labels, + &schema, + std::slice::from_ref(&lhs), + &schema, + std::slice::from_ref(&rhs), + ) + .unwrap() + }; + + let and = eval(SetOp::And); + assert_eq!( + kept(&and), + [("l".to_string(), 0, 4.0, 6.0, Some(LHS_START))], + "`and` did not hand back the left sample it kept: {and:?}" + ); + + let unless = eval(SetOp::Unless); + assert_eq!( + kept(&unless), + [("l".to_string(), 1, 4.0, 6.0, Some(LHS_START))], + "`unless` did not hand back the left sample it kept: {unless:?}" + ); + + // The right side comes back at point 2, where the left side holds no sample, and carries the + // right side's own start timestamp -- the one case where the payload kept is not the left one. + let or = eval(SetOp::Or); + assert_eq!( + kept(&or), + [ + ("l".to_string(), 0, 4.0, 6.0, Some(LHS_START)), + ("l".to_string(), 1, 4.0, 6.0, Some(LHS_START)), + ("r".to_string(), 2, 5.0, 8.0, Some(RHS_START)), + ], + "`or` did not hand back each side's own sample: {or:?}" + ); +}