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

@@ -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<i64>),
+pub struct RawTimestamps {
+    values: ScalarBuffer<i64>,
 }
 
-impl RawTimeAxis {
-    /// Validates and wraps strictly increasing timestamps; an empty axis is valid on its own.
-    pub fn try_new(timestamps: ScalarBuffer<i64>) -> Result<Self> {
-        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<i64>) -> Result<Self> {
+        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<i64>> {
-        match self {
-            Self::Explicit(values) => Some(values),
-        }
-    }
-
-    fn try_len(&self) -> Result<usize> {
-        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<i64> {
+        &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<i64> {
-        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<MetricName>,
     /// 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<MetricName>,
-        time: RawTimeAxis,
+        time: RawTimestamps,
         fields: SeriesFields,
         is_last: bool,
     ) -> Result<Self> {
@@ -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,

src/datatypes/src/series/raw/tests.rs

@@ -46,8 +46,8 @@ fn metric() -> MetricName {
     .unwrap()
 }
 
-fn times(values: Vec<i64>) -> RawTimeAxis {
-    RawTimeAxis::try_new(values.into()).unwrap()
+fn times(values: Vec<i64>) -> RawTimestamps {
+    RawTimestamps::try_new(values.into()).unwrap()
 }
 
 fn float_fields(first: Vec<Option<f64>>, second: Option<Vec<Option<f64>>>) -> 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::<i64>::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<i64> = 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<i64> = 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"),

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<Item = Result<RawSeriesChunk, BoxedError>> + Send {
     /// Returns the schema shared by all chunks; raw schemas have no evaluation grid.
     fn schema(&self) -> &Arc<SeriesSchema>;
@@ -514,6 +518,48 @@ pub trait RawSeriesStream: Stream<Item = Result<RawSeriesChunk, BoxedError>> + S
 /// A pinned raw-series stream that can be sent between threads.
 pub type SendableRawSeriesStream = Pin<Box<dyn RawSeriesStream>>;
 
+/// 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<datatypes::series::MetricName>,
+}
+
+/// 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<SeriesInfo, BoxedError>;
+
+    /// Creates an independent reader for the half-open ordinal range.
+    fn read_series(
+        &self,
+        range: std::ops::Range<usize>,
+    ) -> Result<SendableRawSeriesStream, BoxedError>;
+}
+
 /// 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<Option<Arc<dyn PreparedSeriesSource>>, 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)

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<RawSeriesChunk>,
+}
+
+struct MemorySource {
+    series: Arc<[TestSeries]>,
+    schema: Arc<SeriesSchema>,
+    label_source: RawLabelSource,
+    polls: Arc<AtomicUsize>,
+    drops: Arc<AtomicUsize>,
+}
+
+impl PreparedSeriesSource for MemorySource {
+    fn num_series(&self) -> usize {
+        self.series.len()
+    }
+
+    fn series_info(&self, ordinal: usize) -> Result<SeriesInfo, BoxedError> {
+        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<usize>,
+    ) -> Result<SendableRawSeriesStream, BoxedError> {
+        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<RawSeriesChunk>,
+    schema: Arc<SeriesSchema>,
+    label_source: RawLabelSource,
+    polls: Arc<AtomicUsize>,
+    drops: Arc<AtomicUsize>,
+}
+
+impl Stream for MemoryReader {
+    type Item = Result<RawSeriesChunk, BoxedError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        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<SeriesSchema> {
+        &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<SeriesSchema> {
+    Arc::new(SeriesSchema {
+        fields: vec![FieldSpec {
+            name: Arc::from("usage"),
+        }]
+        .into(),
+        eval_grid: None,
+    })
+}
+
+fn chunk(
+    key: &[u8],
+    metric: Option<MetricName>,
+    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<MemorySource> {
+    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<MemorySource> {
+    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<MemorySource> {
+    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::<Vec<_>>(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::<Vec<_>>(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::<Vec<_>>(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::<Vec<_>>(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::<Vec<_>>(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]))
+    );
+}