rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 23:06:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(datatypes): support sparse evaluation-grid axes (#9762)

Browse Source 
* feat(datatypes): support sparse evaluation-grid axes

A series that holds a value at only some grid positions had no way to say so:
its time axis was one contiguous range, and an explicit timestamp list is not
the same thing -- it loses the evaluation grid the series is validated
against, and cannot answer which grid position a row belongs to.

`TimeAxis` already owns positions, length, timestamp access and slicing, and
already has a chunked form that keeps several Arrow buffers unflattened. This
gives grids the same treatment. `GridSpan` is one run of absolute positions;
`GridRanges` is an ordered, non-overlapping list of spans on one shared
`EvalGrid`; `TimeAxis::GridRanges` carries it beside the contiguous
`TimeAxis::Grid(GridRange)` fast path, which keeps its contract including the
zero-copy `timestamps()` slice.

The coordinates are fixed by the axis and documented on it. `len()` counts
carried rows, not the distance from the first position to the last;
`timestamp`, `slice` and `row_of` speak compact rows; `positions()` walks
`(absolute position, compact row)` pairs in one pass with one cursor per span;
`intersect_runs` merges two run lists without forming their product;
`union_positions` merges several axes. Spans are validated at construction --
ordered, non-overlapping, inside the grid, arithmetic checked -- and an axis
with no rows still owns its grid, so nothing unwraps a first span to find it.
`row_of` is a binary search and `slice` rebuilds through the checked
constructor. `Samples` gains a forward `RowCursor` so a walk reads rows in
order without scanning chunk descriptors per row.

`Series::try_new` accepts the new form under the same rule as the contiguous
one: its grid must match `SeriesSchema::eval_grid`, and an explicit axis still
may not carry an evaluation grid.

Nothing produces a fragmented axis yet. The type and its invariants land
first, and every existing reader keeps its behaviour: the places that require
a contiguous axis already return an error for anything else, so they reject
the new form rather than misreading it. Wiring the readers changes what those
places must do with their results, which is why it follows separately.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

* fix(datatypes): check the shared grid, keep one constructor, and stop a cursor from poisoning itself

`intersect_runs` and `union_positions` walk two axes together but never said
they must be addressed on one grid; on an explicit axis every position is 0, so
the pairs were silent nonsense. Both now state the precondition and assert it in
debug builds.

`GridRanges::try_new` accepted a single span, which is not normalized into the
one run a `TimeAxis` fast path expects, so a public caller could build a range
that reports `is_contiguous()` and still has no grid range. It is now
`pub(crate)`; `TimeAxis::from_grid_spans` stays the public way in and keeps
normalizing a single span.

`Samples::RowCursor::at` drained its iterator on an out-of-range row, so one bad
call made every later call answer `None`. It now bound-checks first and leaves
the cursor where it was.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

---------

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-10-20 15:53:33 +08:00
committed by GitHub
parent 05659b1094
commit 82c7d145b7
4 changed files with 1515 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/datatypes/src/series/samples.rs
+73 -6
View File
@@ -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<Self> {
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<SampleRef<'a>> {
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,
src/datatypes/src/series/samples/tests.rs
+129 -1
View File
@@ -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<HistogramBatch> {
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());
}
src/datatypes/src/series/time.rs
+692 -5
View File
@@ -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<EvalGrid>,
/// 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<EvalGrid>, spans: Vec<GridSpan>) -> Result<Self> {
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<usize> = 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<EvalGrid>) -> Self {
Self {
grid,
spans: Arc::from([]),
rows: Arc::from([]),
len: 0,
}
}
/// Returns the shared grid the spans refer to.
pub fn grid(&self) -> &Arc<EvalGrid> {
&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<usize> {
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<GridSpan> {
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<usize> {
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<TimeAxis> {
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<i64>),
/// Explicit timestamps held in ordered chunks without flattening.
ExplicitChunks(Arc<[ScalarBuffer<i64>]>),
}
/// 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<usize>) {
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<i64>) -> Result<Self> {
@@ -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<EvalGrid>, spans: Vec<GridSpan>) -> Result<Self> {
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<usize> {
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<i64> {
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<Self> {
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<EvalGrid>> {
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<usize> {
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<usize> {
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<usize> {
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<Item = GridRunPair> + '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<GridRun>,
spans: &'a [GridSpan],
span: usize,
row: usize,
}
impl Iterator for GridRuns<'_> {
type Item = GridRun;
fn next(&mut self) -> Option<GridRun> {
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<Item = &'a TimeAxis>) -> 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<Positions<'a>> = 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<Positions<'a>>,
/// The next `(position, row)` of each cursor, buffered so that it can be peeked without a
/// second cursor type.
pending: Vec<Option<(usize, usize)>>,
}
impl Iterator for UnionPositions<'_> {
type Item = usize;
fn next(&mut self) -> Option<usize> {
let mut lowest: Option<usize> = 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 {
src/datatypes/src/series/time/tests.rs
+621 -1
View File
@@ -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<EvalGrid> {
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<_>>(),
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<_>>(),
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<_>>(), 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<_>>(),
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<GridSpan>| 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<_>>(), 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<_>>(),
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<EvalGrid> {
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<SeriesSchema>, 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<_>>(),
vec![(1, 3), (6, 2), (9, 1)]
);
assert_eq!(
series.time.positions().collect::<Vec<_>>(),
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::<Vec<_>>();
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<usize> = series
.time
.positions()
.skip(offset)
.take(len)
.map(|(position, _)| position)
.collect();
assert_eq!(
sliced
.time
.positions()
.map(|(position, _)| position)
.collect::<Vec<_>>(),
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<_>>(),
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<usize> = spans
.iter()
.flat_map(|span| span.start()..span.start() + span.len())
.collect();
assert_eq!(
walked
.iter()
.map(|&(position, _)| position)
.collect::<Vec<_>>(),
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");
}
}