diff --git a/src/datatypes/src/series/samples.rs b/src/datatypes/src/series/samples.rs index 42b1d4bba0b..2efb63059e7 100644 --- a/src/datatypes/src/series/samples.rs +++ b/src/datatypes/src/series/samples.rs @@ -262,6 +262,9 @@ impl Samples { } } /// Returns the sample, treating an out-of-range index as missing. + /// + /// A chunked field walks its descriptors from the start, so reading row after row with this + /// costs a scan per row: a sequential reader should use [`Self::cursor`] instead. pub fn sample(&self, index: usize) -> SampleRef<'_> { if index >= self.len() { return SampleRef::Missing; @@ -312,6 +315,23 @@ impl Samples { remaining, } } + /// A forward reader of this field addressed by compact row; see [`RowCursor`]. + pub fn cursor(&self) -> RowCursor<'_> { + let (chunks, remaining) = match &self.storage { + SampleStorage::Chunks(chunks, len) => (Some(chunks.as_ref()), *len), + _ => (None, self.len()), + }; + RowCursor { + iter: SamplesIter { + samples: self, + chunks, + chunk_index: 0, + local_index: 0, + remaining, + }, + row: 0, + } + } /// Return a checked zero-copy slice (across chunks only intersected descriptors are retained). pub fn slice(&self, offset: usize, len: usize) -> Result { let end = offset.checked_add(len).context(InvalidSeriesSnafu { @@ -531,6 +551,53 @@ impl ExactSizeIterator for SamplesIter<'_> { } } +/// A forward reader of one field's samples, addressed by compact row. +/// +/// [`Samples::sample`] rescans the chunk descriptors from the start on every call, so reading a +/// chunked field with it row by row costs a scan per row. A cursor resumes where it stopped: the +/// rows asked for must not go backwards, and reading `n` rows in order costs one pass over the +/// field. That is what every operator walking a [`TimeAxis`](crate::series::time::TimeAxis) needs, +/// since the axis says which rows exist and the cursor turns them into values. +/// +/// ``` +/// use arrow::array::Float64Array; +/// use datatypes::series::samples::{SampleRef, Samples}; +/// +/// let samples = Samples::try_from_chunks(vec![ +/// Samples::from_float(Float64Array::from(vec![1.0, 2.0])), +/// Samples::from_float(Float64Array::from(vec![3.0])), +/// ]) +/// .unwrap(); +/// let mut cursor = samples.cursor(); +/// assert!(matches!(cursor.at(2), Some(SampleRef::Float(3.0)))); +/// assert!(cursor.at(0).is_none(), "a cursor never goes back"); +/// ``` +pub struct RowCursor<'a> { + iter: SamplesIter<'a>, + /// Row the next [`RowCursor::at`] call must start from. + row: usize, +} +impl<'a> RowCursor<'a> { + /// Reads the sample at compact `row`, which must not be behind the previous read. + /// + /// Returns `None` when the row is behind the cursor's position or at or past the field's end. + /// Such a call leaves the cursor where it was, so one out-of-range row does not consume the + /// rows after it. + pub fn at(&mut self, row: usize) -> Option> { + if row < self.row { + return None; + } + // Bound-check before advancing: the iterator is exact-size, so draining it would make it + // answer `None` for every later row as well. + if row - self.row >= self.iter.len() { + return None; + } + let sample = self.iter.nth(row - self.row)?; + self.row = row + 1; + Some(sample) + } +} + #[derive(Clone, Debug)] enum Fields { One(Samples), @@ -630,18 +697,18 @@ impl Series { if fields.iter().any(|f| f.len() != time_len) { return invalid("series field length does not match time axis"); } - match (&schema.eval_grid, &time) { - (Some(expected), TimeAxis::Grid(range)) if expected.spec() == range.grid().spec() => {} - (Some(_), TimeAxis::Grid(_)) => { + match (&schema.eval_grid, time.grid()) { + (Some(expected), Some(grid)) if expected.spec() == grid.spec() => {} + (Some(_), Some(_)) => { return invalid("series grid does not match schema evaluation grid"); } - (Some(_), _) => { + (Some(_), None) => { return invalid("explicit series time axis cannot use an evaluation grid"); } - (None, TimeAxis::Grid(_)) => { + (None, Some(_)) => { return invalid("grid series requires a schema evaluation grid"); } - (None, TimeAxis::Explicit(_) | TimeAxis::ExplicitChunks(_)) => {} + (None, None) => {} } Ok(Self { metric, diff --git a/src/datatypes/src/series/samples/tests.rs b/src/datatypes/src/series/samples/tests.rs index 62539f885fa..10ed6b7693d 100644 --- a/src/datatypes/src/series/samples/tests.rs +++ b/src/datatypes/src/series/samples/tests.rs @@ -20,7 +20,7 @@ use arrow::buffer::{BooleanBuffer, Buffer, NullBuffer, ScalarBuffer}; use crate::series::histogram::{BucketSideView, CounterResetHint, HistogramBatchBuilder}; -use crate::series::time::{EvalGrid, GridRange, GridSpec, TimeAxis}; +use crate::series::time::{EvalGrid, GridRange, GridRanges, GridSpan, GridSpec, TimeAxis}; fn batch(sum: f64, count: f64, buckets: &[f64]) -> Arc { let offsets = if buckets.is_empty() { vec![] } else { vec![0] }; @@ -1306,3 +1306,131 @@ fn short_multifield_series_uses_its_grid_range_and_checks_explicit_chunk_order() .is_err() ); } + +#[test] +fn series_accepts_a_sparse_axis_on_its_schema_grid_only() { + let spec = GridSpec::try_new(100, 10, 5).unwrap(); + let schema_grid = Arc::new(EvalGrid::try_new(spec).unwrap()); + let metric = MetricName::try_new(None, vec![]).unwrap(); + let schema = SeriesSchema { + fields: Arc::from([FieldSpec { + name: Arc::from("value"), + }]), + eval_grid: Some(schema_grid.clone()), + }; + let values = Samples::from_float(Float64Array::from(vec![1.0, 2.0])); + let fields = SeriesFields::one(values.clone()); + let ranges = || { + GridRanges::try_new( + Arc::new(EvalGrid::try_new(spec).unwrap()), + vec![GridSpan::new(1, 1), GridSpan::new(4, 1)], + ) + .unwrap() + }; + + let series = Series::try_new( + &schema, + metric.clone(), + TimeAxis::GridRanges(ranges()), + fields.clone(), + ) + .unwrap(); + assert_eq!(series.time.len(), 2); + assert_eq!(series.time.grid().map(|grid| grid.spec()), Some(spec)); + + // Another grid specification is not interchangeable, however the rows line up. + let other = GridRanges::try_new( + Arc::new(EvalGrid::try_new(GridSpec::try_new(100, 5, 5).unwrap()).unwrap()), + vec![GridSpan::new(1, 1), GridSpan::new(4, 1)], + ) + .unwrap(); + assert!( + Series::try_new( + &schema, + metric.clone(), + TimeAxis::GridRanges(other), + fields.clone() + ) + .is_err() + ); + + // A schema without an evaluation grid cannot carry grid rows at all. + let explicit_schema = SeriesSchema { + fields: schema.fields.clone(), + eval_grid: None, + }; + assert!( + Series::try_new( + &explicit_schema, + metric.clone(), + TimeAxis::GridRanges(ranges()), + fields.clone() + ) + .is_err() + ); + + // An empty sparse axis still names its grid, so it is accepted with empty fields. + let empty_fields = SeriesFields::one(Samples::from_float(Float64Array::new_null(0))); + let empty = GridRanges::empty(schema_grid.clone()); + assert!( + Series::try_new( + &schema, + metric.clone(), + TimeAxis::GridRanges(empty), + empty_fields.clone() + ) + .is_ok() + ); + assert!( + Series::try_new( + &explicit_schema, + metric, + TimeAxis::GridRanges(GridRanges::empty(Arc::new( + EvalGrid::try_new(spec).unwrap() + ))), + empty_fields.clone() + ) + .is_err() + ); + // Field length still has to match the number of carried rows. + assert!( + Series::try_new( + &schema, + MetricName::try_new(None, vec![]).unwrap(), + TimeAxis::GridRanges(ranges()), + empty_fields + ) + .is_err() + ); +} + +#[test] +fn row_cursor_reads_a_chunked_field_forward_only() { + let samples = Samples::try_from_chunks(vec![ + Samples::from_float(Float64Array::from(vec![1.0, 2.0])), + Samples::from_float(Float64Array::new_null(1)), + Samples::from_float(Float64Array::from(vec![4.0])), + ]) + .unwrap(); + let mut cursor = samples.cursor(); + assert!(matches!(cursor.at(0), Some(SampleRef::Float(1.0)))); + // Skipping rows is allowed: the cursor resumes from its position. + assert!(matches!(cursor.at(2), Some(SampleRef::Missing))); + assert!(matches!(cursor.at(3), Some(SampleRef::Float(4.0)))); + assert!(cursor.at(4).is_none()); + // Going backwards is not a scan: it is refused. + assert!(cursor.at(1).is_none()); +} + +#[test] +fn row_cursor_keeps_its_place_after_a_row_past_the_end() { + let samples = Samples::from_float(Float64Array::from(vec![1.0, 2.0, 3.0])); + let mut cursor = samples.cursor(); + // A row at and past the end is refused without draining the field. + assert!(cursor.at(3).is_none()); + assert!(cursor.at(usize::MAX / 2).is_none()); + // The cursor still sits at row 0, so the valid rows that follow are still readable. + assert!(matches!(cursor.at(1), Some(SampleRef::Float(2.0)))); + assert!(matches!(cursor.at(2), Some(SampleRef::Float(3.0)))); + assert!(cursor.at(3).is_none()); +} diff --git a/src/datatypes/src/series/time.rs b/src/datatypes/src/series/time.rs index 8003b461393..5bff2d45493 100644 --- a/src/datatypes/src/series/time.rs +++ b/src/datatypes/src/series/time.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::borrow::Cow; use std::sync::Arc; use arrow::buffer::ScalarBuffer; @@ -164,20 +165,335 @@ impl PartialEq for GridRange { } impl Eq for GridRange {} +/// One contiguous run of positions on a shared evaluation grid. +/// +/// The span is half-open: it names the absolute grid indices `start..start + len`. A span is +/// never empty, so `start + len - 1` is always its last position. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct GridSpan { + start: usize, + len: usize, +} + +// A span is never empty (the checked constructors reject a zero length), so there is no +// `is_empty` to pair with `len`. +#[allow(clippy::len_without_is_empty)] +impl GridSpan { + /// Creates a span of `len` positions starting at absolute grid index `start`. + /// + /// [`TimeAxis::from_grid_spans`] validates it against its grid; this constructor only records + /// it. + pub fn new(start: usize, len: usize) -> Self { + Self { start, len } + } + + /// Absolute grid index of the first position. + pub fn start(&self) -> usize { + self.start + } + /// Number of positions in the span. + pub fn len(&self) -> usize { + self.len + } +} + +/// Several ordered, non-overlapping runs of one shared evaluation grid. +/// +/// The runs are the rows an axis carries: everything outside them holds no value, so an axis over +/// a few scattered positions costs a few spans instead of a row per empty position. The spans are +/// strictly increasing and never overlap (`prev.start + prev.len <= next.start`); they may sit +/// next to each other, and nothing is stored between them. +/// +/// An axis with no rows is an empty span list that still carries its grid, so a consumer can tell +/// "nothing here" from "not on a grid". +#[derive(Clone, Debug)] +pub struct GridRanges { + grid: Arc, + /// The runs, in absolute grid order; no span is empty. + spans: Arc<[GridSpan]>, + /// Compact row of each span's first row, parallel to `spans`: prefix sums of the lengths, + /// kept so that compact-row lookups need a binary search and not a scan. + rows: Arc<[usize]>, + /// Number of carried rows, cached so that [`Self::len`] never walks the spans. + len: usize, +} + +// `len` is the number of carried rows; "carries no row" is answered by `TimeAxis::is_empty`. +#[allow(clippy::len_without_is_empty)] +impl GridRanges { + /// Creates a validated list of runs over `grid`. + /// + /// Rejects an empty span, a span past the end of the grid, spans that are out of order or + /// overlap, and any length or index that overflows `usize`. + /// + /// Not public: a list of one span is not normalized into the single run a [`TimeAxis`] fast + /// path expects, so callers outside this crate build axes through + /// [`TimeAxis::from_grid_spans`], which keeps [`TimeAxis::Grid`] for a single span. + pub(crate) fn try_new(grid: Arc, spans: Vec) -> Result { + let grid_len = grid.timestamps().len(); + let mut rows = Vec::new(); + rows.try_reserve_exact(spans.len()) + .map_err(|_| invalid_series("grid span allocation failed"))?; + let mut len = 0usize; + let mut previous_end: Option = None; + for span in &spans { + if span.len == 0 { + return invalid("grid spans must not be empty"); + } + let end = span + .start + .checked_add(span.len) + .context(InvalidSeriesSnafu { + reason: "grid span overflow".to_string(), + })?; + if end > grid_len { + return invalid("grid span exceeds its grid"); + } + if previous_end.is_some_and(|previous_end| span.start < previous_end) { + return invalid("grid spans must be ordered and must not overlap"); + } + rows.push(len); + len = len.checked_add(span.len).context(InvalidSeriesSnafu { + reason: "grid span length overflow".to_string(), + })?; + previous_end = Some(end); + } + Ok(Self { + grid, + spans: spans.into(), + rows: rows.into(), + len, + }) + } + + /// Creates an axis over `grid` that carries no row. + pub fn empty(grid: Arc) -> Self { + Self { + grid, + spans: Arc::from([]), + rows: Arc::from([]), + len: 0, + } + } + + /// Returns the shared grid the spans refer to. + pub fn grid(&self) -> &Arc { + &self.grid + } + + /// Returns the number of carried rows, not the distance from the first to the last position. + pub fn len(&self) -> usize { + self.len + } + + /// Maps an absolute grid position to its compact row, or `None` when it holds no row. + /// + /// Binary search over the span starts: a point lookup never scans. + pub fn row_of(&self, position: usize) -> Option { + let index = self + .spans + .partition_point(|span| span.start <= position) + .checked_sub(1)?; + let span = self.spans[index]; + let offset = position - span.start; + (offset < span.len).then(|| self.rows[index] + offset) + } + + /// Returns the span holding compact row `index`, with the offset of the row inside it. + fn span_of(&self, index: usize) -> Option<(GridSpan, usize)> { + let span_index = self + .rows + .partition_point(|&row| row <= index) + .checked_sub(1)?; + let span = self.spans[span_index]; + let offset = index - self.rows[span_index]; + (offset < span.len).then_some((span, offset)) + } + + /// Returns the span that starts at or after compact row `index`, as an index into `spans`. + fn span_index_at(&self, index: usize) -> usize { + self.rows + .partition_point(|&row| row <= index) + .saturating_sub(1) + } + + /// The spans covering compact rows `offset..offset + len`, renumbered from zero. + fn spans_between(&self, offset: usize, len: usize) -> Vec { + let end = offset + len; + let mut sliced = Vec::new(); + for index in self.span_index_at(offset)..self.spans.len() { + let row = self.rows[index]; + if row >= end { + break; + } + let span = self.spans[index]; + let from = offset.max(row) - row; + let to = (end - row).min(span.len); + if from < to { + sliced.push(GridSpan::new(span.start + from, to - from)); + } + } + sliced + } + + /// The absolute position of compact row `offset`, or the position after the last row when + /// `offset` is one past the end; `None` when the axis carries no row at all. + fn position_at(&self, offset: usize) -> Option { + let last = self.spans.last()?; + if offset >= self.len { + return Some(last.start + last.len); + } + let (span, inside) = self.span_of(offset)?; + Some(span.start + inside) + } + + /// Cuts compact rows `offset..offset + len`, keeping the grid. + /// + /// The result is a [`TimeAxis::Grid`] when it stays within one span, including every empty + /// result over a non-empty axis, and [`TimeAxis::GridRanges`] otherwise. + fn slice(&self, offset: usize, len: usize) -> Result { + if len == 0 { + return Ok(match self.position_at(offset) { + Some(position) => { + TimeAxis::Grid(GridRange::try_new(self.grid.clone(), position, 0)?) + } + None => TimeAxis::GridRanges(self.clone()), + }); + } + let spans = self.spans_between(offset, len); + match spans.as_slice() { + [span] => Ok(TimeAxis::Grid(GridRange::try_new( + self.grid.clone(), + span.start, + span.len, + )?)), + _ => Ok(TimeAxis::GridRanges(Self::try_new( + self.grid.clone(), + spans, + )?)), + } + } +} + +impl PartialEq for GridRanges { + fn eq(&self, other: &Self) -> bool { + self.grid.spec == other.grid.spec && self.spans == other.spans && self.len == other.len + } +} +impl Eq for GridRanges {} + /// A timestamp axis represented as a grid or explicit shared Arrow buffers. /// /// Checked constructors validate timestamp ordering. Public payloads are validated again by the /// downstream series boundary. +/// +/// Two coordinate conventions meet here, and every method says which one it speaks: +/// +/// * a **compact row** is an index into the axis and into the fields aligned with it, and it +/// counts only the rows the axis carries: [`TimeAxis::len`], [`TimeAxis::timestamp`], +/// [`TimeAxis::slice`] and the row half of [`TimeAxis::positions`] speak compact rows; +/// * an **absolute grid position** is an index into the shared [`EvalGrid`], holes included; +/// [`TimeAxis::positions`], [`TimeAxis::row_of`], [`TimeAxis::first_position`] and +/// [`TimeAxis::last_position`] speak absolute grid positions. #[derive(Clone, Debug)] pub enum TimeAxis { /// A slice of a shared regular evaluation grid. Grid(GridRange), + /// Several ordered runs of one shared evaluation grid. + GridRanges(GridRanges), /// An explicit timestamp sequence. Explicit(ScalarBuffer), /// Explicit timestamps held in ordered chunks without flattening. ExplicitChunks(Arc<[ScalarBuffer]>), } +/// Sequential traversal of an axis, as returned by [`TimeAxis::positions`]. +/// +/// Yields `(absolute grid position, compact row)` pairs in increasing position order, and walks +/// the runs of a [`TimeAxis::GridRanges`] with one cursor, so a row never costs a scan over the +/// spans before it. +pub struct Positions<'a> { + source: PositionSource<'a>, +} + +enum PositionSource<'a> { + /// A dense grid range: position and row advance together. + Grid { + start: usize, + row: usize, + len: usize, + }, + /// Several runs: the cursor is the span index plus the offset inside it. + GridRanges { + spans: &'a [GridSpan], + span: usize, + offset: usize, + row: usize, + len: usize, + }, + /// No grid: a row sits at its own position. + Rows { row: usize, len: usize }, +} + +impl Iterator for Positions<'_> { + type Item = (usize, usize); + + fn next(&mut self) -> Option<(usize, usize)> { + match &mut self.source { + PositionSource::Grid { start, row, len } => { + if *row >= *len { + return None; + } + let item = (*start + *row, *row); + *row += 1; + Some(item) + } + PositionSource::GridRanges { + spans, + span, + offset, + row, + len, + } => { + if *row >= *len { + return None; + } + loop { + let current = spans.get(*span)?; + if *offset < current.len { + let item = (current.start + *offset, *row); + *offset += 1; + *row += 1; + return Some(item); + } + *span += 1; + *offset = 0; + } + } + PositionSource::Rows { row, len } => { + if *row >= *len { + return None; + } + let item = (*row, *row); + *row += 1; + Some(item) + } + } + } + + fn size_hint(&self) -> (usize, Option) { + let (row, len) = match &self.source { + PositionSource::Grid { row, len, .. } => (*row, *len), + PositionSource::GridRanges { row, len, .. } => (*row, *len), + PositionSource::Rows { row, len } => (*row, *len), + }; + let remaining = len - row; + (remaining, Some(remaining)) + } +} + +impl ExactSizeIterator for Positions<'_> {} + impl TimeAxis { /// Validates an explicit, strictly increasing timestamp sequence. pub fn try_new(timestamps: ScalarBuffer) -> Result { @@ -198,10 +514,27 @@ impl TimeAxis { Ok(axis) } + /// Builds an axis over `grid` from the runs `spans`, keeping the single-run fast path. + /// + /// This is what a producer holding only the rows it decided should use: one span becomes + /// [`Self::Grid`], several become [`Self::GridRanges`], and no span becomes an empty + /// [`Self::GridRanges`] that still carries the grid. In particular it never fills the + /// positions between two spans, which is what keeps a sparse result from costing a row per + /// empty position. + pub fn from_grid_spans(grid: Arc, spans: Vec) -> Result { + match spans.as_slice() { + [] => Ok(Self::GridRanges(GridRanges::empty(grid))), + [span] => Ok(Self::Grid(GridRange::try_new(grid, span.start, span.len)?)), + _ => Ok(Self::GridRanges(GridRanges::try_new(grid, spans)?)), + } + } + /// Returns the validated timestamp count. pub(crate) fn try_len(&self) -> Result { match self { Self::Grid(range) => Ok(range.len()), + // `GridRanges` is only reachable through its checked constructor. + Self::GridRanges(ranges) => Ok(ranges.len()), Self::Explicit(values) => { if values.windows(2).any(|pair| pair[0] >= pair[1]) { return invalid("timestamps must be strictly increasing"); @@ -231,10 +564,13 @@ impl TimeAxis { } } - /// Returns the timestamp count. + /// Returns the compact row count, not the distance from the first to the last position. + /// + /// A [`Self::GridRanges`] answers from its cached total and does not walk its spans. pub fn len(&self) -> usize { match self { Self::Grid(range) => range.len, + Self::GridRanges(ranges) => ranges.len(), Self::Explicit(values) => values.len(), Self::ExplicitChunks(chunks) => chunks.iter().map(|chunk| chunk.len()).sum(), } @@ -244,11 +580,15 @@ impl TimeAxis { self.len() == 0 } - /// Returns a timestamp by checked index. + /// Returns a timestamp by compact row index. pub fn timestamp(&self, index: usize) -> Option { match self { - Self::Grid(range) if index < range.len => { - range.grid.timestamps.get(range.start + index).copied() + Self::Grid(range) => (index < range.len) + .then(|| range.grid.timestamps.get(range.start + index).copied()) + .flatten(), + Self::GridRanges(ranges) => { + let (span, offset) = ranges.span_of(index)?; + ranges.grid.timestamps.get(span.start + offset).copied() } Self::Explicit(values) => values.get(index).copied(), Self::ExplicitChunks(chunks) => { @@ -261,11 +601,14 @@ impl TimeAxis { } None } - _ => None, } } /// Returns a zero-copy slice, retaining Arrow owners; chunked axes remain chunked. + /// + /// `offset` and `len` count compact rows, so the result stays aligned with the fields of the + /// series the axis belongs to. A slice that stays within one run of a [`Self::GridRanges`] + /// comes back as [`Self::Grid`]; every empty slice keeps the grid it was cut from. pub fn slice(&self, offset: usize, len: usize) -> Result { let end = offset.checked_add(len).context(InvalidSeriesSnafu { reason: "timestamp slice overflow".to_string(), @@ -279,6 +622,7 @@ impl TimeAxis { range.start + offset, len, )?)), + Self::GridRanges(ranges) => ranges.slice(offset, len), Self::Explicit(values) => Ok(Self::Explicit(values.slice(offset, len))), Self::ExplicitChunks(chunks) => { let mut remain = len; @@ -303,6 +647,349 @@ impl TimeAxis { } } } + + /// Returns the single contiguous range this axis covers, if it covers exactly one. + /// + /// [`Self::Grid`] always answers `Some`, including for an empty range that still names the + /// position it sits at. A [`Self::GridRanges`] holding several runs, and every explicit axis, + /// answer `None`: their rows are not one range, and a caller that assumed one would read the + /// holes between them as rows. + pub fn as_grid_range(&self) -> Option<&GridRange> { + match self { + Self::Grid(range) => Some(range), + Self::GridRanges(_) | Self::Explicit(_) | Self::ExplicitChunks(_) => None, + } + } + + /// Returns the evaluation grid the axis is addressed on, if it has one. + /// + /// Both [`Self::Grid`] and [`Self::GridRanges`] answer `Some`; the explicit axes answer `None`. + pub fn grid(&self) -> Option<&Arc> { + match self { + Self::Grid(range) => Some(range.grid()), + Self::GridRanges(ranges) => Some(ranges.grid()), + Self::Explicit(_) | Self::ExplicitChunks(_) => None, + } + } + + /// Maps an absolute grid position to its compact row, or `None` when the axis carries no row + /// there. + /// + /// A point lookup: a [`Self::GridRanges`] binary-searches its spans, it does not scan them. + /// On an explicit axis a row sits at its own position, so only the length bounds it. + pub fn row_of(&self, position: usize) -> Option { + match self { + Self::Grid(range) => (position >= range.start + && position + .checked_sub(range.start) + .is_some_and(|row| row < range.len)) + .then(|| position - range.start), + Self::GridRanges(ranges) => ranges.row_of(position), + Self::Explicit(_) | Self::ExplicitChunks(_) => { + (position < self.len()).then_some(position) + } + } + } + + /// Walks the axis once, yielding `(absolute grid position, compact row)` in position order. + /// + /// This is the traversal every consumer of a grid axis should use: a helper walks the runs of + /// a [`Self::GridRanges`] with a single cursor, so a row never rescans the spans before it. + /// On an explicit axis there is no grid, so a row sits at its own position. + pub fn positions(&self) -> Positions<'_> { + let len = self.len(); + let source = match self { + Self::Grid(range) => PositionSource::Grid { + start: range.start, + row: 0, + len, + }, + Self::GridRanges(ranges) => PositionSource::GridRanges { + spans: &ranges.spans, + span: 0, + offset: 0, + row: 0, + len, + }, + Self::Explicit(_) | Self::ExplicitChunks(_) => PositionSource::Rows { row: 0, len }, + }; + Positions { source } + } + + /// Returns the absolute grid position of the first carried row. + pub fn first_position(&self) -> Option { + match self { + Self::Grid(range) => (!range.is_empty()).then_some(range.start), + Self::GridRanges(ranges) => ranges.spans.first().map(GridSpan::start), + Self::Explicit(_) | Self::ExplicitChunks(_) => (!self.is_empty()).then_some(0), + } + } + + /// Returns the absolute grid position of the last carried row. + pub fn last_position(&self) -> Option { + match self { + Self::Grid(range) => (!range.is_empty()).then(|| range.start + range.len - 1), + Self::GridRanges(ranges) => ranges.spans.last().map(|span| span.start + span.len - 1), + Self::Explicit(_) | Self::ExplicitChunks(_) => self.len().checked_sub(1), + } + } + + /// Returns whether the carried rows form one run, which is what a fast path can rely on. + /// + /// A [`Self::GridRanges`] answers `true` only for a single span, so an empty span list reports + /// `false`: it names no run, and therefore no position a fast path could start at. + pub fn is_contiguous(&self) -> bool { + match self { + Self::Grid(_) => true, + Self::GridRanges(ranges) => ranges.spans.len() == 1, + Self::Explicit(_) => true, + Self::ExplicitChunks(chunks) => chunks.len() <= 1, + } + } + + /// Returns the timestamps of the carried rows, in compact-row order. + /// + /// A single-run grid axis lends its buffer without copying; an axis of several runs, and a + /// chunked explicit axis, allocate one array of the carried rows. + pub fn timestamps(&self) -> Cow<'_, [i64]> { + match self { + Self::Grid(range) => { + Cow::Borrowed(&range.grid.timestamps()[range.start()..range.start() + range.len()]) + } + Self::GridRanges(ranges) => { + let mut timestamps = Vec::with_capacity(ranges.len()); + for (position, _) in self.positions() { + match ranges.grid.timestamps.get(position) { + Some(timestamp) => timestamps.push(*timestamp), + // Unreachable: the constructor checks every span against the grid. + None => break, + } + } + Cow::Owned(timestamps) + } + Self::Explicit(values) => Cow::Borrowed(values), + Self::ExplicitChunks(_) => Cow::Owned( + (0..self.len()) + .filter_map(|row| self.timestamp(row)) + .collect(), + ), + } + } + + /// Returns one run per stored span of carried rows, as [`GridRun`]s. + /// + /// One run for a contiguous axis, one per stored span for [`Self::GridRanges`], and none when + /// no row is carried. Adjacent stored spans stay two runs, so a run is a maximal run only when + /// the axis was built from maximal spans (as the series builder does). A run is where bulk + /// work pays off: the rows inside it are consecutive in both coordinates, so its samples can + /// be sliced and zipped instead of read one by one. + pub fn runs(&self) -> GridRuns<'_> { + let (one, spans) = match self { + Self::Grid(range) => ( + Some(GridRun { + position: range.start, + row: 0, + len: range.len, + }), + &[][..], + ), + Self::GridRanges(ranges) => (None, &ranges.spans[..]), + Self::Explicit(_) | Self::ExplicitChunks(_) => ( + Some(GridRun { + position: 0, + row: 0, + len: self.len(), + }), + &[][..], + ), + }; + GridRuns { + one: one.filter(|run| run.len > 0), + spans, + span: 0, + row: 0, + } + } + + /// Intersects the carried rows of two axes, yielding one shared run per pair of overlapping + /// stored runs as `(first absolute position, row in `self`, row in `other`, length)`. + /// + /// A single merge pass over the two run lists, so the cost is the number of runs and not their + /// product. This is how a pairwise operator aligns two axes: it never builds the dense + /// envelope `first_position()..last_position()`, which would visit -- and densify -- the holes, + /// and it never re-derives the row of a position by subtracting starts, which is wrong as soon + /// as either side skips a row. + /// + /// Both axes must be addressed on the same [`GridSpec`]: the positions it yields are indices + /// into one shared [`EvalGrid`], so two different grids line up only by coincidence. An + /// explicit axis carries no grid and must not be passed at all -- its rows are not absolute + /// positions on any shared grid, so the pairs would be silent nonsense. Debug builds assert + /// the precondition and therefore panic on a violation; release builds trust the caller. + pub fn intersect_runs<'a>(&'a self, other: &'a Self) -> impl Iterator + 'a { + debug_assert!( + matches!( + (self.grid(), other.grid()), + (Some(lhs), Some(rhs)) if lhs.spec() == rhs.spec() + ), + "intersect_runs requires two axes on one grid, \ + but an explicit axis carries none and two grids only meet by coincidence", + ); + let mut lhs = self.runs().peekable(); + let mut rhs = other.runs().peekable(); + std::iter::from_fn(move || { + loop { + let left = *lhs.peek()?; + let right = *rhs.peek()?; + let start = left.position.max(right.position); + let end = (left.position + left.len).min(right.position + right.len); + if start >= end { + // No shared position in this pair of runs: drop the one that ends first. + if left.position + left.len <= right.position + right.len { + lhs.next(); + } else { + rhs.next(); + } + continue; + } + let len = end - start; + let pair = GridRunPair { + position: start, + row: left.row + (start - left.position), + other_row: right.row + (start - right.position), + len, + }; + // Consume the shared piece from both sides, keeping the remainder for the next + // round. + if left.position + left.len <= end { + lhs.next(); + } + if right.position + right.len <= end { + rhs.next(); + } + return Some(pair); + } + }) + } +} + +/// A run of carried rows: `len` rows starting at absolute position `position` and compact +/// `row`, both of which advance by one per row. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct GridRun { + /// Absolute grid position of the run's first row. + pub position: usize, + /// Compact row of the run's first row. + pub row: usize, + /// Number of rows in the run; never zero. + pub len: usize, +} + +/// A run shared by two axes, with the compact row of each. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct GridRunPair { + /// Absolute grid position of the shared run's first row. + pub position: usize, + /// Compact row of the first shared position in the axis `intersect_runs` was called on. + pub row: usize, + /// Compact row of the first shared position in the other axis. + pub other_row: usize, + /// Number of rows in the shared run; never zero. + pub len: usize, +} + +/// Lazily produced runs of one axis; see [`TimeAxis::runs`]. +/// +/// A contiguous axis holds its one run inline; a sparse axis walks its spans with a cursor. +pub struct GridRuns<'a> { + one: Option, + spans: &'a [GridSpan], + span: usize, + row: usize, +} + +impl Iterator for GridRuns<'_> { + type Item = GridRun; + + fn next(&mut self) -> Option { + if let Some(run) = self.one.take() { + return Some(run); + } + let span = *self.spans.get(self.span)?; + self.span += 1; + let run = GridRun { + position: span.start, + row: self.row, + len: span.len, + }; + self.row += span.len; + Some(run) + } +} + +/// Walks every position at least one of `axes` carries, once, in increasing order. +/// +/// A merge over one cursor per axis: each step costs one comparison per axis, and no position is +/// visited twice. This is how an operator covering several axes (a histogram's bucket series, an +/// aggregate over an input set) visits the positions that hold something, instead of scanning the +/// envelope from the earliest to the latest position and densifying every hole in between. +/// +/// Every axis must be addressed on the same [`GridSpec`]: a position names a point of one shared +/// [`EvalGrid`], so merging axes of two different grids would interleave unrelated positions. An +/// explicit axis carries no grid and must not be passed at all -- its rows are not absolute +/// positions on any shared grid. Debug builds assert the precondition and therefore panic on a +/// violation; release builds trust the caller. +pub fn union_positions<'a>(axes: impl IntoIterator) -> UnionPositions<'a> { + // Materialized once so that the debug assert above can see every axis' grid before the first + // cursor is built; the cursors below borrow these references, not the list. + let axes: Vec<&'a TimeAxis> = axes.into_iter().collect(); + debug_assert!( + { + let mut specs = axes.iter().map(|axis| axis.grid().map(|grid| grid.spec())); + match specs.next() { + // No axis at all: the merge yields nothing and reads no grid. + None => true, + Some(None) => false, + Some(Some(first)) => specs.all(|spec| spec == Some(first)), + } + }, + "union_positions requires every axis on one grid, \ + but an explicit axis carries none and two grids only meet by coincidence", + ); + let cursors: Vec> = axes.into_iter().map(TimeAxis::positions).collect(); + let pending = vec![None; cursors.len()]; + UnionPositions { cursors, pending } +} + +/// The positions of several axes, merged; see [`union_positions`]. +pub struct UnionPositions<'a> { + cursors: Vec>, + /// The next `(position, row)` of each cursor, buffered so that it can be peeked without a + /// second cursor type. + pending: Vec>, +} + +impl Iterator for UnionPositions<'_> { + type Item = usize; + + fn next(&mut self) -> Option { + let mut lowest: Option = None; + for (cursor, pending) in self.cursors.iter_mut().zip(&mut self.pending) { + if pending.is_none() { + *pending = cursor.next(); + } + if let Some((position, _)) = *pending { + lowest = Some(lowest.map_or(position, |lowest| lowest.min(position))); + } + } + let lowest = lowest?; + // Every axis standing at this position advances; the others keep their buffered position. + for pending in &mut self.pending { + if matches!(*pending, Some((position, _)) if position == lowest) { + *pending = None; + } + } + Some(lowest) + } } fn invalid_series(reason: &str) -> crate::error::Error { diff --git a/src/datatypes/src/series/time/tests.rs b/src/datatypes/src/series/time/tests.rs index f3ee539a624..de3ba29f621 100644 --- a/src/datatypes/src/series/time/tests.rs +++ b/src/datatypes/src/series/time/tests.rs @@ -19,7 +19,7 @@ use std::sync::{ use arrow::buffer::{Buffer, ScalarBuffer}; -use super::{EvalGrid, GridRange, GridSpec, TimeAxis}; +use super::{EvalGrid, GridRange, GridRanges, GridSpan, GridSpec, TimeAxis}; #[test] fn grid_specs_validate_and_ranges_are_inclusive_without_rounding_up() { @@ -247,3 +247,623 @@ fn sliced_explicit_and_chunk_buffers_retain_and_release_real_allocations() { assert_eq!(first_drops.load(Ordering::SeqCst), 1); assert_eq!(second_drops.load(Ordering::SeqCst), 1); } + +/// The axis of the sparse tests: positions 1..3 and 5..6 of a six-point grid. +fn sparse_grid() -> Arc { + Arc::new(EvalGrid::try_new(GridSpec::try_new(10, 10, 6).unwrap()).unwrap()) +} + +fn sparse_axis() -> TimeAxis { + TimeAxis::GridRanges( + GridRanges::try_new( + sparse_grid(), + vec![GridSpan::new(1, 2), GridSpan::new(5, 1)], + ) + .unwrap(), + ) +} + +#[test] +fn sparse_axis_counts_carried_rows_and_maps_them_to_timestamps() { + let axis = sparse_axis(); + // Three rows over the six positions from the first to the last, so `len` is not the envelope. + assert_eq!(axis.len(), 3); + assert_eq!(axis.first_position(), Some(1)); + assert_eq!(axis.last_position(), Some(5)); + assert!(!axis.is_contiguous()); + assert_eq!( + axis.grid().map(|grid| grid.spec()), + Some(sparse_grid().spec()) + ); + assert_eq!(axis.timestamp(0), Some(20)); + assert_eq!(axis.timestamp(1), Some(30)); + assert_eq!(axis.timestamp(2), Some(60)); + assert_eq!(axis.timestamp(3), None); + + let dense = TimeAxis::Grid(GridRange::try_new(sparse_grid(), 1, 2).unwrap()); + assert_eq!(dense.len(), 2); + assert!(dense.is_contiguous()); + assert_eq!( + (dense.first_position(), dense.last_position()), + (Some(1), Some(2)) + ); +} + +#[test] +fn sparse_axis_slices_stay_compact_and_keep_the_grid() { + let axis = sparse_axis(); + // Inside one span: the result is the single-range fast path. + let inside = axis.slice(2, 1).unwrap(); + let TimeAxis::Grid(range) = &inside else { + panic!("a slice inside one span is not a grid range") + }; + assert_eq!((range.start(), range.len()), (5, 1)); + assert_eq!(inside.timestamp(0), Some(60)); + + // Across the gap: rows 1 and 2 are positions 2 and 5, not positions 2, 3, 4 and 5. + let across = axis.slice(1, 2).unwrap(); + let TimeAxis::GridRanges(_) = &across else { + panic!("a slice across a gap stayed a grid range") + }; + assert_eq!( + across + .runs() + .map(|run| (run.position, run.row, run.len)) + .collect::>(), + vec![(2, 0, 1), (5, 1, 1)] + ); + assert_eq!(across.len(), 2); + assert_eq!( + (across.timestamp(0), across.timestamp(1)), + (Some(30), Some(60)) + ); + + // A slice that starts and ends in the same span still shares the grid. + let whole = axis.slice(0, 2).unwrap(); + let TimeAxis::Grid(range) = &whole else { + panic!("a slice inside the first span is not a grid range") + }; + assert_eq!((range.start(), range.len()), (1, 2)); + + // Empty slices keep the grid: at a row, and right after the last row. + for (offset, position) in [(0, 1), (2, 5), (3, 6)] { + let empty = axis.slice(offset, 0).unwrap(); + let TimeAxis::Grid(range) = &empty else { + panic!("an empty slice lost the grid range it was cut at") + }; + assert!(range.is_empty()); + assert_eq!(range.start(), position); + assert_eq!(range.grid().spec(), sparse_grid().spec()); + } + let (offset, len) = (3, 0); + assert!(axis.slice(offset, len).is_ok()); + assert!(axis.slice(3, 1).is_err()); + assert!(axis.slice(4, 0).is_err()); + assert!(axis.slice(2, usize::MAX).is_err()); + + // A nested slice of a sparse slice keeps working. + let nested = across.slice(1, 1).unwrap(); + assert_eq!(nested.timestamp(0), Some(60)); +} + +#[test] +fn sparse_axis_positions_walk_every_row_once_in_order() { + let axis = sparse_axis(); + assert_eq!( + axis.positions().collect::>(), + vec![(1, 0), (2, 1), (5, 2)] + ); + let dense = TimeAxis::Grid(GridRange::try_new(sparse_grid(), 2, 2).unwrap()); + assert_eq!(dense.positions().collect::>(), vec![(2, 0), (3, 1)]); + let explicit = TimeAxis::try_new(ScalarBuffer::from(vec![1_i64, 3])).unwrap(); + // Without a grid a row sits at its own position. + assert_eq!( + explicit.positions().collect::>(), + vec![(0, 0), (1, 1)] + ); + let mut positions = axis.positions(); + assert_eq!(positions.len(), axis.len()); + assert_eq!(positions.next(), Some((1, 0))); + assert_eq!(positions.len(), 2, "a consumed row is no longer counted"); + // The last row of the first span. + assert_eq!(positions.next(), Some((2, 1))); + assert_eq!(positions.len(), 1); + // The next row is in the second span, across the gap. + assert_eq!(positions.next(), Some((5, 2))); + assert_eq!(positions.len(), 0); + assert_eq!(positions.next(), None); + assert_eq!(positions.len(), 0, "an exhausted walk reports nothing left"); +} + +#[test] +fn sparse_axis_row_of_answers_carried_positions_only() { + let axis = sparse_axis(); + for (position, row) in [(1, 0), (2, 1), (5, 2)] { + assert_eq!(axis.row_of(position), Some(row)); + } + for position in [0, 3, 4, 6, 9] { + assert_eq!(axis.row_of(position), None, "position {position}"); + } + let explicit = TimeAxis::try_new(ScalarBuffer::from(vec![1_i64, 3])).unwrap(); + assert_eq!(explicit.row_of(1), Some(1)); + assert_eq!(explicit.row_of(2), None); +} + +#[test] +fn sparse_spans_are_validated() { + let grid = sparse_grid(); + let build = |spans: Vec| GridRanges::try_new(grid.clone(), spans); + assert!(build(vec![GridSpan::new(0, 1), GridSpan::new(2, 3)]).is_ok()); + // Adjacent runs are allowed. + assert!(build(vec![GridSpan::new(0, 2), GridSpan::new(2, 2)]).is_ok()); + // A span may end at the grid end, which is index 6 on this grid. + assert!(build(vec![GridSpan::new(4, 2)]).is_ok()); + assert!(build(vec![GridSpan::new(5, 1)]).is_ok()); + + // Overlapping and out-of-order spans. + assert!(build(vec![GridSpan::new(1, 2), GridSpan::new(2, 1)]).is_err()); + assert!(build(vec![GridSpan::new(2, 1), GridSpan::new(1, 1)]).is_err()); + // Past the grid end. + assert!(build(vec![GridSpan::new(5, 2)]).is_err()); + assert!(build(vec![GridSpan::new(6, 1)]).is_err()); + // Zero-length spans, alone and in a list. + assert!(build(vec![GridSpan::new(1, 0)]).is_err()); + assert!(build(vec![GridSpan::new(1, 1), GridSpan::new(3, 0)]).is_err()); + // Index and length arithmetic is checked. + assert!(build(vec![GridSpan::new(usize::MAX, 1)]).is_err()); + assert!(build(vec![GridSpan::new(usize::MAX - 1, 2)]).is_err()); + assert!(build(vec![GridSpan::new(usize::MAX, usize::MAX)]).is_err()); +} + +#[test] +fn empty_sparse_axis_keeps_its_grid() { + let axis = TimeAxis::GridRanges(GridRanges::empty(sparse_grid())); + assert!(axis.is_empty()); + assert_eq!(axis.len(), 0); + assert!(!axis.is_contiguous()); + assert_eq!(axis.first_position(), None); + assert_eq!(axis.last_position(), None); + assert_eq!( + axis.grid().map(|grid| grid.spec()), + Some(sparse_grid().spec()) + ); + assert_eq!(axis.timestamp(0), None); + assert_eq!(axis.row_of(0), None); + assert_eq!(axis.positions().collect::>(), Vec::new()); + + // Slicing an empty axis keeps it empty and on its grid. + let empty = axis.slice(0, 0).unwrap(); + let TimeAxis::GridRanges(ranges) = &empty else { + panic!("an empty sparse slice invented a position") + }; + assert_eq!(ranges.len(), 0); + assert_eq!(ranges.grid().spec(), sparse_grid().spec()); +} + +#[test] +fn sparse_runs_are_the_carried_rows_and_intersect_by_a_merge_walk() { + let axis = sparse_axis(); + let runs: Vec<_> = axis.runs().collect(); + assert_eq!(runs.len(), 2); + assert_eq!((runs[0].position, runs[0].row, runs[0].len), (1, 0, 2)); + assert_eq!((runs[1].position, runs[1].row, runs[1].len), (5, 2, 1)); + + // An axis sharing the middle row and the last position with `axis`. + let other = TimeAxis::GridRanges( + GridRanges::try_new( + sparse_grid(), + vec![GridSpan::new(2, 1), GridSpan::new(5, 1)], + ) + .unwrap(), + ); + let shared: Vec<_> = axis.intersect_runs(&other).collect(); + assert_eq!(shared.len(), 2); + assert_eq!( + ( + shared[0].position, + shared[0].row, + shared[0].other_row, + shared[0].len + ), + (2, 1, 0, 1) + ); + assert_eq!( + ( + shared[1].position, + shared[1].row, + shared[1].other_row, + shared[1].len + ), + (5, 2, 1, 1) + ); + + // An overlapping stretch of several rows is one shared run with two row offsets. + let left = TimeAxis::GridRanges( + GridRanges::try_new(sparse_grid(), vec![GridSpan::new(0, 4)]).unwrap(), + ); + let right = TimeAxis::GridRanges( + GridRanges::try_new(sparse_grid(), vec![GridSpan::new(1, 3)]).unwrap(), + ); + let shared: Vec<_> = left.intersect_runs(&right).collect(); + assert_eq!(shared.len(), 1); + assert_eq!( + ( + shared[0].position, + shared[0].row, + shared[0].other_row, + shared[0].len + ), + (1, 1, 0, 3) + ); + + // Disjoint axes share nothing, and an empty axis shares nothing either. + let far = TimeAxis::GridRanges( + GridRanges::try_new(sparse_grid(), vec![GridSpan::new(3, 2)]).unwrap(), + ); + assert_eq!(axis.intersect_runs(&far).count(), 0); + let empty = TimeAxis::GridRanges(GridRanges::empty(sparse_grid())); + assert_eq!(axis.intersect_runs(&empty).count(), 0); + assert_eq!(empty.intersect_runs(&axis).count(), 0); + + // A dense axis is one run, so its intersection with a sparse axis is the sparse rows. + let dense = TimeAxis::Grid(GridRange::try_new(sparse_grid(), 0, 6).unwrap()); + let shared: Vec<_> = dense.intersect_runs(&axis).collect(); + assert_eq!( + shared + .iter() + .map(|pair| (pair.position, pair.row, pair.other_row, pair.len)) + .collect::>(), + vec![(1, 1, 0, 2), (5, 5, 2, 1)] + ); +} + +#[test] +fn from_grid_spans_keeps_the_single_span_fast_path() { + let grid = sparse_grid(); + let single = TimeAxis::from_grid_spans(grid.clone(), vec![GridSpan::new(1, 2)]).unwrap(); + let TimeAxis::Grid(range) = &single else { + panic!("one span is a grid range") + }; + assert_eq!((range.start(), range.len()), (1, 2)); + + let several = + TimeAxis::from_grid_spans(grid.clone(), vec![GridSpan::new(1, 1), GridSpan::new(5, 1)]) + .unwrap(); + assert!(matches!(several, TimeAxis::GridRanges(_))); + assert_eq!(several.len(), 2); + + let none = TimeAxis::from_grid_spans(grid, Vec::new()).unwrap(); + assert!(none.is_empty()); + assert_eq!( + none.grid().map(|grid| grid.spec()), + Some(sparse_grid().spec()) + ); + // The same validation as the checked constructor. + assert!(TimeAxis::from_grid_spans(sparse_grid(), vec![GridSpan::new(5, 3)]).is_err()); +} + +// Tests of the fragmented grid axis as an end-to-end contract: what a slice, a walk and a point +// lookup say about a series that carries only the positions holding a value. +mod acceptance { + use std::sync::Arc; + + use arrow::array::Float64Array; + + use super::super::{EvalGrid, GridRanges, GridSpan, GridSpec, TimeAxis}; + use crate::series::MetricName; + use crate::series::samples::{ + FieldSpec, SampleRef, Samples, Series, SeriesFields, SeriesSchema, + }; + + /// The grid of the fragmented-series fixture: twelve positions, ten milliseconds apart. + fn fixture_grid() -> Arc { + Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 10, 12).unwrap()).unwrap()) + } + + /// A two-field series over positions 1, 2, 3, 6, 7 and 9 of [`fixture_grid`]: three runs, and two + /// fields that disagree about which rows hold a value. + /// + /// The first field holds a value everywhere; the second has a hole, a NaN -- a *value*, not a + /// hole -- and another hole, so a slice that shifts by one row changes what it reads. + fn fragmented_series() -> (Arc, Series) { + let grid = fixture_grid(); + let schema = Arc::new(SeriesSchema { + fields: vec![ + FieldSpec { + name: Arc::from("a"), + }, + FieldSpec { + name: Arc::from("b"), + }, + ] + .into(), + eval_grid: Some(grid.clone()), + }); + let time = TimeAxis::from_grid_spans( + grid, + vec![ + GridSpan::new(1, 3), + GridSpan::new(6, 2), + GridSpan::new(9, 1), + ], + ) + .unwrap(); + let a = Samples::from_float(Float64Array::from(vec![11.0, 12.0, 13.0, 16.0, 17.0, 19.0])); + let b = Samples::from_float(Float64Array::from(vec![ + None, + Some(f64::NAN), + Some(13.0), + Some(16.0), + None, + Some(19.0), + ])); + let series = Series::try_new( + &schema, + MetricName::try_new(Some(Arc::from("m")), vec![]).unwrap(), + time, + SeriesFields::try_from_vec(vec![a, b]).unwrap(), + ) + .unwrap(); + (schema, series) + } + + /// One sample read of two fields is the same reading: same kind, same bits, and any NaN as NaN, + /// the way results are compared elsewhere in the tree. + fn same_reading(a: SampleRef<'_>, b: SampleRef<'_>) -> bool { + match (a, b) { + (SampleRef::Missing, SampleRef::Missing) => true, + (SampleRef::Float(a), SampleRef::Float(b)) if a.is_nan() || b.is_nan() => { + a.is_nan() && b.is_nan() + } + (SampleRef::Float(a), SampleRef::Float(b)) => a.to_bits() == b.to_bits(), + (SampleRef::Histogram(_), SampleRef::Histogram(_)) => false, + _ => false, + } + } + + /// Slicing a fragmented series cuts the axis and every field at the same compact offset, keeps the + /// absolute positions of the rows, and keeps the grid the rows were addressed on. + /// + /// The cases cover a slice inside one run (which is the single-range fast path), one that starts + /// on the last row of a run and ends on the first row of the next, one spanning two runs, one at + /// the very end of the axis, and the two empty slices -- inside the axis and one past its last + /// row. A slice taken by the span start rather than by the row offset would read a neighbour's + /// rows and fail the timestamp and field comparisons. + #[test] + fn slicing_a_fragmented_series_cuts_every_field_with_its_axis() { + let (schema, series) = fragmented_series(); + assert_eq!(series.time.len(), 6); + assert!(!series.time.is_contiguous()); + assert_eq!( + series + .time + .runs() + .map(|run| (run.position, run.len)) + .collect::>(), + vec![(1, 3), (6, 2), (9, 1)] + ); + assert_eq!( + series.time.positions().collect::>(), + vec![(1, 0), (2, 1), (3, 2), (6, 3), (7, 4), (9, 5)] + ); + + let cases: [(usize, usize, &str); 6] = [ + (1, 2, "inside one run"), + (2, 2, "exactly across a run boundary"), + (1, 4, "spanning two runs"), + (4, 2, "at the very end"), + (2, 0, "empty, between two rows"), + (6, 0, "empty, one past the last row"), + ]; + for (offset, len, context) in cases { + let time = series + .time + .slice(offset, len) + .unwrap_or_else(|error| panic!("{context}: slicing the axis failed: {error}")); + let fields = series + .fields + .iter() + .map(|field| { + field.slice(offset, len).unwrap_or_else(|error| { + panic!("{context}: slicing a field failed: {error}") + }) + }) + .collect::>(); + let sliced = Series::try_new( + &schema, + series.metric.clone(), + time, + SeriesFields::try_from_vec(fields).expect("the fields stay equally long"), + ) + .unwrap_or_else(|error| { + panic!("{context}: the sliced series is inconsistent: {error}") + }); + + assert_eq!(sliced.time.len(), len, "{context}: the axis length"); + assert_eq!(sliced.fields.len(), 2, "{context}: a field was lost"); + for (index, field) in sliced.fields.iter().enumerate() { + assert_eq!(field.len(), len, "{context}: field {index} length"); + } + assert_eq!( + sliced.time.grid().map(|grid| grid.spec()), + series.time.grid().map(|grid| grid.spec()), + "{context}: the slice lost its grid" + ); + for index in 0..len { + assert_eq!( + sliced.time.timestamp(index), + series.time.timestamp(offset + index), + "{context}: timestamp {index} moved" + ); + } + // The rows of the slice sit at the absolute positions the original rows sat at, so a + // consumer that reads through the axis reaches the same instants. + let original_positions: Vec = series + .time + .positions() + .skip(offset) + .take(len) + .map(|(position, _)| position) + .collect(); + assert_eq!( + sliced + .time + .positions() + .map(|(position, _)| position) + .collect::>(), + original_positions, + "{context}: the slice moved the rows" + ); + for (position, row) in sliced.time.positions() { + assert_eq!( + sliced.time.row_of(position), + Some(row), + "{context}: row_of disagrees with the walk at {position}" + ); + } + + // Every field reads the samples the original field reads at `offset + row`. + for (field_index, sliced_field) in sliced.fields.iter().enumerate() { + let original_field = &series.fields.as_slice()[field_index]; + let mut original_cursor = original_field.cursor(); + let mut sliced_cursor = sliced_field.cursor(); + for row in 0..len { + let want = original_cursor + .at(offset + row) + .expect("the original field is long enough"); + let got = sliced_cursor.at(row).expect("the slice is long enough"); + assert!( + same_reading(want, got), + "{context}: field {field_index} row {row} (original row {}) reads differently", + offset + row + ); + } + } + if len == 0 { + // An empty slice still names the grid it was cut from, and the series built on it is + // accepted: `Series::try_new` checks the field lengths against the axis. + assert!(sliced.time.is_empty(), "{context}: an empty slice is empty"); + assert!( + sliced + .time + .grid() + .is_some_and(|grid| grid.spec() == fixture_grid().spec()), + "{context}: an empty slice lost the grid" + ); + } + } + + // The two shapes the slice must keep: inside one run it stays the single-range fast path, and + // across a boundary it becomes a span list -- never a filled-in range. + assert!( + series.time.slice(1, 2).unwrap().as_grid_range().is_some(), + "a slice inside one run is a grid range" + ); + let across = series.time.slice(2, 2).unwrap(); + assert!( + across.as_grid_range().is_none(), + "a slice across a run boundary is not one range: {across:?}" + ); + assert_eq!(across.len(), 2); + assert_eq!( + across + .runs() + .map(|run| (run.position, run.len)) + .collect::>(), + vec![(3, 1), (6, 1)] + ); + assert!( + series.time.slice(6, 1).is_err(), + "a slice past the last row is rejected" + ); + assert!( + series.time.slice(usize::MAX, 1).is_err(), + "an overflowing slice is rejected" + ); + } + + /// A highly fragmented axis: `k` runs of one to three rows, `V` rows in total, on a grid far + /// longer than both. + /// + /// The descriptors scale with the runs and the rows with the values; nothing scales with the + /// grid. What this test can observe is that shape: the run count, the row count, the number of + /// items a walk yields, and which positions a point lookup answers for. It cannot observe the + /// cost of the loops behind them, and it deliberately makes no timing assertion: an inner-loop + /// cost is not a property of the API, so asserting on it would only measure the machine. + #[test] + fn a_fragmented_axis_costs_its_runs_and_rows_not_its_grid() { + const GRID: usize = 100_000; + let grid = + Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 1, GRID as u64).unwrap()).unwrap()); + // Runs of one to three rows, spread so that consecutive runs are always separated by at + // least one position holding no row. + let mut spans = Vec::new(); + let mut position = 0; + let mut run_index = 0; + while position < GRID && spans.len() < 20_000 { + let len = 1 + run_index % 3; + spans.push(GridSpan::new(position, len)); + position += len + 1 + run_index % 5; + run_index += 1; + } + let runs = spans.len(); + let rows: usize = spans.iter().map(GridSpan::len).sum(); + assert_eq!(runs, 20_000, "the fixture is as fragmented as intended"); + assert!(rows > runs, "{rows} rows over {runs} runs"); + assert!( + rows < GRID / 2, + "{rows} rows over a grid of {GRID}: the axis carries a small part of it" + ); + let axis = TimeAxis::GridRanges(GridRanges::try_new(grid.clone(), spans.clone()).unwrap()); + + // Descriptors scale with the runs, not with the grid. + assert_eq!(axis.runs().count(), runs, "one descriptor per run"); + assert_eq!(axis.len(), rows, "the axis carries one row per value"); + assert!(axis.len() < GRID, "the axis is not the grid"); + assert!(!axis.is_contiguous()); + + // One walk visits exactly the carried rows, once each and in order: a walk that + // materialised the holes would yield GRID items. + let walked: Vec<(usize, usize)> = axis.positions().collect(); + assert_eq!(walked.len(), rows, "the walk yields the carried rows only"); + assert_ne!(walked.len(), GRID, "the walk did not visit the grid"); + assert_eq!(axis.positions().len(), rows, "the walk knows its length"); + assert!(walked.windows(2).all(|pair| pair[0].0 < pair[1].0)); + assert!(walked.iter().enumerate().all(|(row, &(_, r))| row == r)); + let expected_positions: Vec = spans + .iter() + .flat_map(|span| span.start()..span.start() + span.len()) + .collect(); + assert_eq!( + walked + .iter() + .map(|&(position, _)| position) + .collect::>(), + expected_positions + ); + + // Point lookups answer at the carried positions and nowhere else. + for (row, &position) in expected_positions.iter().enumerate() { + assert_eq!( + axis.row_of(position), + Some(row), + "carried position {position}" + ); + } + for (index, span) in spans.iter().enumerate().take(200) { + let hole = span.start() + span.len(); + assert_eq!( + axis.row_of(hole), + None, + "run {index}: position {hole} holds no row" + ); + } + assert_eq!(axis.row_of(GRID), None, "past the grid"); + assert_eq!(axis.row_of(0), Some(0), "the first run starts at 0"); + // Every carried row resolves to a timestamp on the shared grid, which is what a consumer + // reads instead of a timestamp per position. + let last = *expected_positions.last().expect("the fixture carries rows"); + assert_eq!(axis.timestamp(rows - 1), Some(last as i64)); + assert_eq!(axis.timestamp(rows), None, "past the last row"); + } +}