From cfc777f55ce517905bfeb76dc8103c710f3552de Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Fri, 16 Oct 2026 15:00:04 +0800 Subject: [PATCH] feat(storage): separate prepared series selection from raw readers Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/datatypes/src/series/raw.rs | 101 ++--- src/datatypes/src/series/raw/tests.rs | 51 +-- src/store-api/src/region_engine.rs | 75 +++- .../region_engine/prepared_series_tests.rs | 405 ++++++++++++++++++ 4 files changed, 546 insertions(+), 86 deletions(-) create mode 100644 src/store-api/src/region_engine/prepared_series_tests.rs diff --git a/src/datatypes/src/series/raw.rs b/src/datatypes/src/series/raw.rs index 6c5b04ceac7..9c18ae2941a 100644 --- a/src/datatypes/src/series/raw.rs +++ b/src/datatypes/src/series/raw.rs @@ -22,59 +22,51 @@ use crate::error::{InvalidSeriesSnafu, Result}; use crate::series::samples::{SeriesFields, SeriesSchema}; use crate::series::{MetricName, invalid_series}; -/// Explicit source timestamps in milliseconds for a raw series chunk. -#[non_exhaustive] +/// Strictly increasing source timestamps in milliseconds for a raw series chunk. +/// +/// Its private buffer ensures timestamps are validated once at construction. Direct +/// construction is intentionally unavailable: +/// +/// ```compile_fail +/// use arrow::buffer::ScalarBuffer; +/// use datatypes::series::raw::RawTimestamps; +/// +/// let timestamps = RawTimestamps { +/// values: ScalarBuffer::from(vec![1_i64, 2]), +/// }; +/// ``` #[derive(Clone, Debug)] -pub enum RawTimeAxis { - /// Source timestamps in strictly increasing order. - Explicit(ScalarBuffer), +pub struct RawTimestamps { + values: ScalarBuffer, } -impl RawTimeAxis { - /// Validates and wraps strictly increasing timestamps; an empty axis is valid on its own. - pub fn try_new(timestamps: ScalarBuffer) -> Result { - let axis = Self::Explicit(timestamps); - axis.try_len()?; - Ok(axis) +impl RawTimestamps { + /// Validates and wraps strictly increasing timestamps; an empty wrapper is valid. + pub fn try_new(values: ScalarBuffer) -> Result { + if values.windows(2).any(|pair| pair[0] >= pair[1]) { + return invalid_series("raw timestamps must be strictly increasing"); + } + Ok(Self { values }) } - /// Returns the explicit timestamp buffer, if this axis uses explicit timestamps. - /// - /// Future axis variants may not have an explicit buffer. - pub fn as_explicit(&self) -> Option<&ScalarBuffer> { - match self { - Self::Explicit(values) => Some(values), - } - } - - fn try_len(&self) -> Result { - match self { - Self::Explicit(values) => { - if values.windows(2).any(|pair| pair[0] >= pair[1]) { - return invalid_series("raw timestamps must be strictly increasing"); - } - Ok(values.len()) - } - } + /// Returns the underlying timestamp buffer. + pub fn as_buffer(&self) -> &ScalarBuffer { + &self.values } /// Returns the number of timestamps. pub fn len(&self) -> usize { - match self { - Self::Explicit(values) => values.len(), - } + self.values.len() } - /// Returns whether the axis has no timestamps. + /// Returns whether there are no timestamps. pub fn is_empty(&self) -> bool { - self.len() == 0 + self.values.is_empty() } /// Returns the timestamp at `index`, or `None` when out of bounds. pub fn timestamp(&self, index: usize) -> Option { - match self { - Self::Explicit(values) => values.get(index).copied(), - } + self.values.get(index).copied() } /// Returns a zero-copy slice, reporting overflowing or out-of-range bounds as an error. @@ -82,23 +74,23 @@ impl RawTimeAxis { let end = offset .checked_add(len) .with_context(|| InvalidSeriesSnafu { - reason: "raw time axis slice bounds overflow".to_string(), + reason: "raw timestamps slice bounds overflow".to_string(), })?; if end > self.len() { - return invalid_series("raw time axis slice is out of bounds"); - } - match self { - Self::Explicit(values) => Ok(Self::Explicit(values.slice(offset, len))), + return invalid_series("raw timestamps slice is out of bounds"); } + Ok(Self { + values: self.values.slice(offset, len), + }) } } /// Ordering guarantee for chunks in one input stream. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum RawSeriesOrdering { - /// Each read key's chunks are contiguous; distinct read keys need not be ordered. + /// Each complete primary key's chunks are contiguous; distinct keys need not be ordered. Contiguous, - /// Chunks are ordered by read key and then by timestamp. + /// Chunks are ordered by complete primary key and then by timestamp. ReadKeyThenTime, } @@ -109,27 +101,29 @@ pub enum RawSeriesOrdering { #[derive(Clone, Debug)] pub struct RawSeriesChunk { /// Complete encoded identity of the source series, not a TSID or hash alone. - pub read_key: Buffer, - /// `None` leaves labels unmaterialized in the complete read key; `Some` contains - /// complete explicit labels. + pub primary_key: Buffer, + /// `None` denotes an encoded key under bound-label mode; `Some` contains complete + /// explicit labels and retains their identity. pub metric: Option, /// Source timestamps aligned with every field. - pub time: RawTimeAxis, + pub time: RawTimestamps, /// Field values aligned with the source timestamps. pub fields: SeriesFields, - /// Whether this is the final chunk for this input stream's series, not all regions. + /// Whether this is the final chunk for this series in the input stream exposing it. + /// Merge code must recompute finalization globally across all relevant inputs before + /// downstream finalization. pub is_last: bool, } impl RawSeriesChunk { - /// Creates a nonempty chunk after checking its explicit time axis and schema alignment. + /// Creates a nonempty chunk after checking its time axis and schema alignment. /// /// Validation is local to this chunk and does not verify ordering between chunks. pub fn try_new( schema: &SeriesSchema, - read_key: Buffer, + primary_key: Buffer, metric: Option, - time: RawTimeAxis, + time: RawTimestamps, fields: SeriesFields, is_last: bool, ) -> Result { @@ -145,9 +139,8 @@ impl RawSeriesChunk { if time.is_empty() { return invalid_series("raw series chunks must not be empty"); } - time.try_len()?; Ok(Self { - read_key, + primary_key, metric, time, fields, diff --git a/src/datatypes/src/series/raw/tests.rs b/src/datatypes/src/series/raw/tests.rs index ee982148d73..ea2d4f80193 100644 --- a/src/datatypes/src/series/raw/tests.rs +++ b/src/datatypes/src/series/raw/tests.rs @@ -46,8 +46,8 @@ fn metric() -> MetricName { .unwrap() } -fn times(values: Vec) -> RawTimeAxis { - RawTimeAxis::try_new(values.into()).unwrap() +fn times(values: Vec) -> RawTimestamps { + RawTimestamps::try_new(values.into()).unwrap() } fn float_fields(first: Vec>, second: Option>>) -> SeriesFields { @@ -64,8 +64,13 @@ fn constructors_reject_invalid_chunk_boundaries_and_unordered_timestamps() { let key = Buffer::from(vec![1_u8]); let fields = float_fields(vec![Some(1.0), Some(2.0)], None); - assert!(RawTimeAxis::try_new(vec![1_i64, 1].into()).is_err()); - assert!(RawTimeAxis::try_new(vec![2_i64, 1].into()).is_err()); + assert!(RawTimestamps::try_new(vec![1_i64, 1].into()).is_err()); + assert!(RawTimestamps::try_new(vec![2_i64, 1].into()).is_err()); + assert!( + RawTimestamps::try_new(ScalarBuffer::from(Vec::::new())) + .unwrap() + .is_empty() + ); assert!( RawSeriesChunk::try_new( @@ -101,22 +106,6 @@ fn constructors_reject_invalid_chunk_boundaries_and_unordered_timestamps() { .is_err() ); - // Public enum construction bypasses RawTimeAxis::try_new, so chunk construction - // must independently reject duplicate and descending timestamps. - for unordered in [vec![1_i64, 1], vec![2_i64, 1]] { - assert!( - RawSeriesChunk::try_new( - &schema, - key.clone(), - None, - RawTimeAxis::Explicit(unordered.into()), - fields.clone(), - true, - ) - .is_err() - ); - } - let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(10, 5, 2).unwrap()).unwrap()); let grid_schema = SeriesSchema { fields: schema.fields.clone(), @@ -131,9 +120,9 @@ fn constructors_reject_invalid_chunk_boundaries_and_unordered_timestamps() { #[test] fn sliced_axes_and_chunks_keep_shared_buffers_and_values_alive() { let time_owner: ScalarBuffer = vec![0, 10, 20, 30, 40, 80].into(); - let time_source = RawTimeAxis::try_new(time_owner.clone()).unwrap(); + let time_source = RawTimestamps::try_new(time_owner.clone()).unwrap(); let time = time_source.slice(1, 4).unwrap(); - let time_buffer = time.as_explicit().unwrap(); + let time_buffer = time.as_buffer(); assert_eq!(time_buffer.as_ptr(), time_owner.as_ptr().wrapping_add(1)); let nan_payload = f64::from_bits(0x7ff8_0000_0000_0042); @@ -160,12 +149,12 @@ fn sliced_axes_and_chunks_keep_shared_buffers_and_values_alive() { ]) .unwrap(); let key_source = Buffer::from(vec![0_u8, 0xA1, 0xB2, 0xC3, 0xD4, 0xFF]); - let read_key = key_source.slice_with_length(1, 4); - assert_eq!(read_key.as_ptr(), key_source.as_ptr().wrapping_add(1)); + let primary_key = key_source.slice_with_length(1, 4); + assert_eq!(primary_key.as_ptr(), key_source.as_ptr().wrapping_add(1)); let schema = schema(&["first", "second"]); let chunk = - RawSeriesChunk::try_new(&schema, read_key, Some(metric()), time, fields, true).unwrap(); + RawSeriesChunk::try_new(&schema, primary_key, Some(metric()), time, fields, true).unwrap(); let Some(crate::series::samples::DenseSamplesRef::Float(first)) = chunk.fields.get(0).unwrap().dense() else { @@ -181,7 +170,7 @@ fn sliced_axes_and_chunks_keep_shared_buffers_and_values_alive() { drop(time_owner); drop(float_owner); drop(second_owner); - assert_eq!(chunk.read_key.as_slice(), &[0xA1, 0xB2, 0xC3, 0xD4]); + assert_eq!(chunk.primary_key.as_slice(), &[0xA1, 0xB2, 0xC3, 0xD4]); assert_eq!( (0..4) .map(|i| chunk.time.timestamp(i).unwrap()) @@ -224,9 +213,9 @@ fn sliced_axes_and_chunks_keep_shared_buffers_and_values_alive() { } #[test] -fn raw_time_axis_checks_bounds_and_retains_exact_end_slices() { +fn raw_timestamps_check_bounds_and_retain_exact_end_slices() { let source: ScalarBuffer = vec![4, 8, 12].into(); - let axis = RawTimeAxis::try_new(source.clone()).unwrap(); + let axis = RawTimestamps::try_new(source.clone()).unwrap(); assert!(axis.slice(usize::MAX, 1).is_err()); assert!(axis.slice(2, usize::MAX).is_err()); assert!(axis.slice(3, 1).is_err()); @@ -272,11 +261,11 @@ fn raw_chunks_preserve_native_unmaterialized_and_explicit_metrics() { false, ) .unwrap(); - assert_eq!(native.read_key.as_slice(), &[0, 0x12, 0x34, 0x56]); + assert_eq!(native.primary_key.as_slice(), &[0, 0x12, 0x34, 0x56]); assert!(native.metric.is_none()); - assert!(tagless.read_key.is_empty()); + assert!(tagless.primary_key.is_empty()); assert!(tagless.metric.is_none()); - assert_eq!(explicit.read_key.as_slice(), &[0, 0xAB, 0xCD, 0xEF]); + assert_eq!(explicit.primary_key.as_slice(), &[0, 0xAB, 0xCD, 0xEF]); assert_eq!(explicit.metric.as_ref().unwrap().name(), Some("cpu")); assert_eq!( explicit.metric.as_ref().unwrap().get("host"), diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs index 0fe6ae767c1..ec829f7e587 100644 --- a/src/store-api/src/region_engine.rs +++ b/src/store-api/src/region_engine.rs @@ -499,7 +499,11 @@ pub enum RawLabelSource { /// A stream of raw series chunks with stable schema, label binding, and ordering. /// /// Chunks for a read key are contiguous, with strictly increasing timestamps within and across -/// chunks. `is_last` marks the last chunk of that key in this stream, not across streams. +/// chunks. `is_last` marks the last chunk of that key in this stream, not across streams; merge +/// code must recompute finalization globally across all relevant inputs before downstream +/// per-input-series finalization (not aggregate or vector-match state). `RawLabelSource` keeps +/// the stream's label interpretation self-contained through transport rather than relying on a +/// local source handle. pub trait RawSeriesStream: Stream> + Send { /// Returns the schema shared by all chunks; raw schemas have no evaluation grid. fn schema(&self) -> &Arc; @@ -514,6 +518,48 @@ pub trait RawSeriesStream: Stream> + S /// A pinned raw-series stream that can be sent between threads. pub type SendableRawSeriesStream = Pin>; +/// Stable identity metadata for one ordinal in a prepared series source. +#[derive(Clone, Debug)] +pub struct SeriesInfo { + /// Complete encoded primary key, not a TSID or hash alone. + pub primary_key: datatypes::arrow::buffer::Buffer, + /// Complete explicit metric and labels, or `None` for primary-key label binding. + pub metric: Option, +} + +/// An immutable, finite selected set of series prepared for independent range reads. +/// +/// The count may include series with no samples after filtering. Ordinals are stable local +/// indexes, not protocol identifiers; each has exactly one complete identity returned by +/// [`PreparedSeriesSource::series_info`], and that identity matches chunks for that ordinal. +/// Every reader exposes the same fixed schema, label binding, and ordering before polling, +/// including for an empty range, and describes labels independently of this source's lifetime. `PrimaryKey` output has `metric: None`; `Explicit` output has a +/// complete `Some(metric)` matching each chunk. [`RawSeriesOrdering::Contiguous`] and +/// [`RawSeriesOrdering::ReadKeyThenTime`] describe stream order, not ordinal enumeration order. +/// +/// Preparation, identity lookup, and reader creation do not load sample payloads or perform +/// hidden blocking I/O. The selected set is bound to the partition's immutable MVCC view, +/// predicates, authorization, projection, labels, and existing scan resources. Readers own +/// resources needed to continue after this source or scanner is dropped. Dropping one reader +/// does not cancel sibling readers; normal ownership cleanup applies without detached work. +/// +/// A range must satisfy `start <= end <= num_series()`. Invalid ranges and invalid ordinals +/// return errors; an empty in-bounds range returns an ordinary empty stream. Repeated and +/// overlapping reads are independent and do not consume shared iteration state. +pub trait PreparedSeriesSource: Send + Sync { + /// Returns the exact, stable number of selected local series ordinals. + fn num_series(&self) -> usize; + + /// Returns identity metadata for an ordinal without polling sample data. + fn series_info(&self, ordinal: usize) -> Result; + + /// Creates an independent reader for the half-open ordinal range. + fn read_series( + &self, + range: std::ops::Range, + ) -> Result; +} + /// A scanner that provides a way to scan the region concurrently. /// /// The scanner splits the region into partitions so that each partition can be scanned concurrently. @@ -575,6 +621,22 @@ pub trait RegionScanner: Debug + DisplayAs + Send { todo!() } + /// Prepares a finite, stable set of series for independent ordinal-range reads. + /// + /// `None` means unsupported before consumption, so callers may use the legacy record-batch + /// scan. `Some`, including a source with zero series, selects this path; errors propagate + /// without replay. Choose exactly one entrypoint per partition per execution: this prepared + /// source, [`Self::scan_series_partition`], or [`Self::scan_partition`]. In particular, do + /// not call the legacy raw-scan default merely because this returns `None`. + fn prepare_series_partition( + &self, + _ctx: &QueryScanContext, + _metrics_set: &ExecutionPlanMetricsSet, + _partition: usize, + ) -> Result>, BoxedError> { + Ok(None) + } + /// Check if there is any predicate exclude region partition exprs that may be executed in this scanner. fn has_predicate_without_region(&self) -> bool; @@ -813,6 +875,10 @@ impl RegionStatistic { } } +#[cfg(test)] +#[path = "region_engine/prepared_series_tests.rs"] +mod prepared_series_tests; + #[cfg(test)] mod tests { use common_recordbatch::{RecordBatch, RecordBatches}; @@ -868,6 +934,13 @@ mod tests { let ctx = QueryScanContext::default(); let metrics = ExecutionPlanMetricsSet::new(); + assert!( + scanner + .prepare_series_partition(&ctx, &metrics, 0) + .unwrap() + .is_none() + ); + // SinglePartitionScanner uses the default raw-scan implementation, which panics. let raw_scan = catch_unwind(AssertUnwindSafe(|| { scanner.scan_series_partition(&ctx, &metrics, 0, &RawLabelSource::Explicit) diff --git a/src/store-api/src/region_engine/prepared_series_tests.rs b/src/store-api/src/region_engine/prepared_series_tests.rs new file mode 100644 index 00000000000..05882ddcc70 --- /dev/null +++ b/src/store-api/src/region_engine/prepared_series_tests.rs @@ -0,0 +1,405 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::VecDeque; +use std::pin::Pin; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; + +use api::v1::SemanticType; +use common_error::ext::BoxedError; +use datatypes::arrow::array::Float64Array; +use datatypes::arrow::buffer::Buffer; +use datatypes::error::InvalidSeriesSnafu; +use datatypes::prelude::ConcreteDataType; +use datatypes::schema::ColumnSchema; +use datatypes::series::raw::{RawSeriesChunk, RawSeriesOrdering, RawTimestamps}; +use datatypes::series::samples::{FieldSpec, Samples, SeriesFields, SeriesSchema}; +use datatypes::series::{Label, MetricName}; +use futures::{Stream, StreamExt}; + +use crate::region_engine::{ + PreparedSeriesSource, RawLabelSource, RawSeriesStream, SendableRawSeriesStream, SeriesInfo, +}; + +#[derive(Clone)] +struct TestSeries { + info: SeriesInfo, + chunks: Vec, +} + +struct MemorySource { + series: Arc<[TestSeries]>, + schema: Arc, + label_source: RawLabelSource, + polls: Arc, + drops: Arc, +} + +impl PreparedSeriesSource for MemorySource { + fn num_series(&self) -> usize { + self.series.len() + } + + fn series_info(&self, ordinal: usize) -> Result { + self.series + .get(ordinal) + .map(|series| series.info.clone()) + .ok_or_else(|| { + BoxedError::new( + InvalidSeriesSnafu { + reason: "invalid series ordinal".to_string(), + } + .build(), + ) + }) + } + + fn read_series( + &self, + range: std::ops::Range, + ) -> Result { + if range.start > range.end || range.end > self.series.len() { + return Err(BoxedError::new( + InvalidSeriesSnafu { + reason: "invalid series range".to_string(), + } + .build(), + )); + } + let chunks = self.series[range] + .iter() + .flat_map(|series| series.chunks.iter().cloned()) + .collect(); + Ok(Box::pin(MemoryReader { + chunks, + schema: self.schema.clone(), + label_source: self.label_source.clone(), + polls: self.polls.clone(), + drops: self.drops.clone(), + })) + } +} + +struct MemoryReader { + chunks: VecDeque, + schema: Arc, + label_source: RawLabelSource, + polls: Arc, + drops: Arc, +} + +impl Stream for MemoryReader { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + let chunk = self.chunks.pop_front(); + if chunk.is_some() { + self.polls.fetch_add(1, Ordering::SeqCst); + } + Poll::Ready(chunk.map(Ok)) + } +} + +impl Drop for MemoryReader { + fn drop(&mut self) { + self.drops.fetch_add(1, Ordering::SeqCst); + } +} + +impl RawSeriesStream for MemoryReader { + fn schema(&self) -> &Arc { + &self.schema + } + + fn label_source(&self) -> &RawLabelSource { + &self.label_source + } + + fn ordering(&self) -> RawSeriesOrdering { + RawSeriesOrdering::Contiguous + } +} + +fn metric(host: &str) -> MetricName { + MetricName::try_new( + Some(Arc::from("cpu")), + vec![Label { + name: Arc::from("host"), + value: Arc::from(host), + }], + ) + .unwrap() +} + +fn schema() -> Arc { + Arc::new(SeriesSchema { + fields: vec![FieldSpec { + name: Arc::from("usage"), + }] + .into(), + eval_grid: None, + }) +} + +fn chunk( + key: &[u8], + metric: Option, + timestamp: i64, + value: f64, + last: bool, +) -> RawSeriesChunk { + RawSeriesChunk::try_new( + &schema(), + Buffer::from(key), + metric, + RawTimestamps::try_new(vec![timestamp].into()).unwrap(), + SeriesFields::one(Samples::from_float(Float64Array::from(vec![value]))), + last, + ) + .unwrap() +} + +fn primary_key_metadata() -> crate::metadata::RegionMetadata { + let mut builder = + crate::metadata::RegionMetadataBuilder::new(crate::storage::RegionId::new(1, 1)); + builder.push_column_metadata(crate::metadata::ColumnMetadata { + column_schema: ColumnSchema::new("host", ConcreteDataType::string_datatype(), false), + semantic_type: SemanticType::Tag, + column_id: 1, + }); + builder.push_column_metadata(crate::metadata::ColumnMetadata { + column_schema: ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 2, + }); + builder.primary_key(vec![1]); + builder.build().unwrap() +} + +fn empty_source() -> Arc { + Arc::new(MemorySource { + series: Arc::from([]), + schema: schema(), + label_source: RawLabelSource::Explicit, + polls: Arc::new(AtomicUsize::new(0)), + drops: Arc::new(AtomicUsize::new(0)), + }) +} + +fn source(explicit: bool) -> Arc { + source_from_keys( + explicit, + vec![ + Buffer::from(vec![0x10, 0x01]), + Buffer::from(vec![0x10, 0x02]), + ] + .into(), + ) +} + +fn source_from_keys(explicit: bool, keys: Arc<[Buffer]>) -> Arc { + assert_eq!(keys.len(), 2); + let first_metric = explicit.then(|| metric("a")); + let second_metric = explicit.then(|| metric("b")); + let series = vec![ + TestSeries { + info: SeriesInfo { + primary_key: keys[0].clone(), + metric: first_metric.clone(), + }, + chunks: vec![chunk(keys[0].as_slice(), first_metric, 10, 1.0, true)], + }, + TestSeries { + info: SeriesInfo { + primary_key: keys[1].clone(), + metric: second_metric.clone(), + }, + chunks: vec![ + chunk(keys[1].as_slice(), second_metric.clone(), 10, 2.0, false), + chunk(keys[1].as_slice(), second_metric, 20, 3.0, true), + ], + }, + ]; + Arc::new(MemorySource { + series: series.into(), + schema: schema(), + label_source: if explicit { + RawLabelSource::Explicit + } else { + RawLabelSource::PrimaryKey { + metric_name: Arc::from("cpu"), + physical_metadata: Arc::new(primary_key_metadata()), + logical_tags: Arc::from([]), + } + }, + polls: Arc::new(AtomicUsize::new(0)), + drops: Arc::new(AtomicUsize::new(0)), + }) +} + +#[tokio::test] +async fn prepared_identity_matches_chunks_in_both_label_modes_including_empty_reader() { + for explicit in [false, true] { + let source = source(explicit); + assert_eq!(source.num_series(), 2); + let first = source.series_info(0).unwrap(); + let second = source.series_info(1).unwrap(); + assert_eq!(first.primary_key, Buffer::from(vec![0x10, 0x01])); + assert_eq!(second.primary_key, Buffer::from(vec![0x10, 0x02])); + assert_eq!( + first.metric.as_ref().and_then(MetricName::name), + if explicit { Some("cpu") } else { None } + ); + assert_eq!( + first.metric.as_ref().and_then(|m| m.get("host")), + explicit.then_some("a") + ); + assert_eq!( + second.metric.as_ref().and_then(|m| m.get("host")), + explicit.then_some("b") + ); + + let empty = source.read_series(1..1).unwrap(); + assert_eq!(empty.schema().fields.len(), 1); + assert_eq!(empty.ordering(), RawSeriesOrdering::Contiguous); + assert_eq!( + matches!(empty.label_source(), RawLabelSource::Explicit), + explicit + ); + assert_eq!(source.polls.load(Ordering::SeqCst), 0); + assert!( + futures::TryStreamExt::try_collect::>(empty) + .await + .unwrap() + .is_empty() + ); + assert_eq!(source.polls.load(Ordering::SeqCst), 0); + let end = source.read_series(2..2).unwrap(); + assert_eq!(end.schema().fields.len(), 1); + assert_eq!( + matches!(end.label_source(), RawLabelSource::Explicit), + explicit + ); + + let chunks = + futures::TryStreamExt::try_collect::>(source.read_series(0..2).unwrap()) + .await + .unwrap(); + assert_eq!(chunks.len(), 3); + assert_eq!(source.polls.load(Ordering::SeqCst), 3); + assert_eq!(chunks[1].time.timestamp(0), Some(10)); + assert_eq!(chunks[2].time.timestamp(0), Some(20)); + assert!( + matches!(chunks[1].fields.get(0).unwrap().sample(0), datatypes::series::samples::SampleRef::Float(value) if value == 2.0) + ); + assert!( + matches!(chunks[2].fields.get(0).unwrap().sample(0), datatypes::series::samples::SampleRef::Float(value) if value == 3.0) + ); + for chunk in &chunks { + let ordinal = if chunk.primary_key == first.primary_key { + 0 + } else { + 1 + }; + let info = source.series_info(ordinal).unwrap(); + assert_eq!(chunk.primary_key, info.primary_key); + assert_eq!(chunk.metric, info.metric); + } + assert_eq!(chunks[0].metric.is_some(), explicit); + assert!(!chunks[1].is_last); + assert!(chunks[2].is_last); + } +} + +#[tokio::test] +async fn empty_prepared_source_supports_its_empty_range() { + let source = empty_source(); + assert_eq!(source.num_series(), 0); + let stream = source.read_series(0..0).unwrap(); + assert_eq!(stream.schema().fields.len(), 1); + assert!( + futures::TryStreamExt::try_collect::>(stream) + .await + .unwrap() + .is_empty() + ); + assert_eq!(source.polls.load(Ordering::SeqCst), 0); +} + +#[test] +fn invalid_ordinals_and_ranges_fail_without_polling_samples() { + let source = source(false); + assert!(source.series_info(2).is_err()); + assert!(source.read_series(2..3).is_err()); + assert!(source.read_series(2..1).is_err()); + assert_eq!(source.polls.load(Ordering::SeqCst), 0); +} + +#[tokio::test] +async fn overlapping_readers_are_independent_across_source_and_reader_drops() { + let source = source(false); + let drops = source.drops.clone(); + let mut left = source.read_series(0..2).unwrap(); + let mut right = source.read_series(1..2).unwrap(); + drop(source); + assert!(left.next().await.unwrap().unwrap().is_last); + assert_eq!( + right.next().await.unwrap().unwrap().primary_key, + Buffer::from(vec![0x10, 0x02]) + ); + drop(left); + assert_eq!( + futures::TryStreamExt::try_collect::>(right) + .await + .unwrap() + .len(), + 1 + ); + assert_eq!(drops.load(Ordering::SeqCst), 2); +} + +#[tokio::test] +async fn prepared_selection_is_independent_of_later_mutable_selection_changes() { + let selected = Arc::new(Mutex::new(vec![ + Buffer::from(vec![1_u8]), + Buffer::from(vec![2_u8]), + ])); + let prepared = source_from_keys(false, selected.lock().unwrap().clone().into()); + selected.lock().unwrap().push(Buffer::from(vec![3_u8])); + + assert_eq!(prepared.num_series(), 2); + assert_eq!( + prepared.series_info(0).unwrap().primary_key, + Buffer::from(vec![1_u8]) + ); + assert_eq!( + prepared.series_info(1).unwrap().primary_key, + Buffer::from(vec![2_u8]) + ); + let chunks = futures::TryStreamExt::try_collect::>(prepared.read_series(0..2).unwrap()) + .await + .unwrap(); + assert_eq!(chunks.len(), 3); + assert!( + chunks + .iter() + .all(|chunk| chunk.primary_key != Buffer::from(vec![3_u8])) + ); +}