diff --git a/src/datatypes/src/series.rs b/src/datatypes/src/series.rs
index d5d11bfd5ec..860358eff20 100644
--- a/src/datatypes/src/series.rs
+++ b/src/datatypes/src/series.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub mod histogram;
+pub mod raw;
 pub mod samples;
 pub mod time;
 
@@ -23,6 +24,27 @@ use std::sync::Arc;
 
 use crate::error::{InvalidSeriesSnafu, Result};
 
+/// Ordering guarantee within one stream and each of its output partitions.
+///
+/// Every series has strictly increasing sample positions, including across chunks; positions
+/// are raw timestamps or, for evaluated series, current-stage evaluation times or grid positions.
+/// They need not all be explicit times on the grid. No mode permits
+/// duplicate positions. `Interleaved` permits A1, B1, A2; only the latter two modes keep each
+/// series contiguous. `KeyThenTime` also orders by this stream's series identity, then time:
+/// raw streams use the complete bound-compatible identity (not just a TSID or hash), while
+/// evaluated streams use current name/label identity, not storage primary keys or object address/hash.
+/// This ordering
+/// is not a query-global ordering, partition-ownership rule, or finality guarantee.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum SeriesOrdering {
+    /// Series chunks may interleave; positions within each series remain strictly increasing.
+    Interleaved,
+    /// Each series is contiguous; distinct series need not be ordered.
+    Contiguous,
+    /// Series are ordered by stream identity and then by strictly increasing position.
+    KeyThenTime,
+}
+
 /// A label with shared UTF-8 name and value strings.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Label {
diff --git a/src/datatypes/src/series/raw.rs b/src/datatypes/src/series/raw.rs
new file mode 100644
index 00000000000..da4ba487204
--- /dev/null
+++ b/src/datatypes/src/series/raw.rs
@@ -0,0 +1,144 @@
+// 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.
+
+#[cfg(test)]
+mod tests;
+
+use arrow::buffer::{Buffer, ScalarBuffer};
+use snafu::OptionExt;
+
+use crate::error::{InvalidSeriesSnafu, Result};
+use crate::series::samples::{SeriesFields, SeriesSchema};
+use crate::series::{MetricName, invalid_series};
+
+/// 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 struct RawTimestamps {
+    values: ScalarBuffer<i64>,
+}
+
+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 underlying timestamp buffer.
+    pub fn as_buffer(&self) -> &ScalarBuffer<i64> {
+        &self.values
+    }
+
+    /// Returns the number of timestamps.
+    pub fn len(&self) -> usize {
+        self.values.len()
+    }
+
+    /// Returns whether there are no timestamps.
+    pub fn is_empty(&self) -> bool {
+        self.values.is_empty()
+    }
+
+    /// Returns the timestamp at `index`, or `None` when out of bounds.
+    pub fn timestamp(&self, index: usize) -> Option<i64> {
+        self.values.get(index).copied()
+    }
+
+    /// Returns a zero-copy slice, reporting overflowing or out-of-range bounds as an error.
+    pub fn slice(&self, offset: usize, len: usize) -> Result<Self> {
+        let end = offset
+            .checked_add(len)
+            .with_context(|| InvalidSeriesSnafu {
+                reason: "raw timestamps slice bounds overflow".to_string(),
+            })?;
+        if end > self.len() {
+            return invalid_series("raw timestamps slice is out of bounds");
+        }
+        Ok(Self {
+            values: self.values.slice(offset, len),
+        })
+    }
+}
+
+/// One raw source batch, with fields aligned to its timestamps.
+///
+/// `metric: None` means labels are not yet decoded because identity is bound by the source; it
+/// does not mean the metric name is absent. An explicit `Some(metric)` is always complete.
+///
+/// Public fields allow direct access and construction, which bypasses the checks in
+/// [`RawSeriesChunk::try_new`]; callers crossing a trust boundary must validate input.
+#[derive(Clone, Debug)]
+pub struct RawSeriesChunk {
+    /// Complete encoded identity of the source series, not a TSID or hash alone.
+    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: RawTimestamps,
+    /// Field values aligned with the source timestamps.
+    pub fields: SeriesFields,
+    /// 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 time axis and schema alignment.
+    ///
+    /// Validation is local to this chunk and does not verify ordering between chunks.
+    pub fn try_new(
+        schema: &SeriesSchema,
+        primary_key: Buffer,
+        metric: Option<MetricName>,
+        time: RawTimestamps,
+        fields: SeriesFields,
+        is_last: bool,
+    ) -> Result<Self> {
+        if schema.eval_grid.is_some() {
+            return invalid_series("raw series schema cannot have an evaluation grid");
+        }
+        if fields.len() != schema.fields.len() {
+            return invalid_series("raw series field count does not match schema");
+        }
+        if fields.iter().any(|field| field.len() != time.len()) {
+            return invalid_series("raw series field length does not match time axis");
+        }
+        if time.is_empty() {
+            return invalid_series("raw series chunks must not be empty");
+        }
+        Ok(Self {
+            primary_key,
+            metric,
+            time,
+            fields,
+            is_last,
+        })
+    }
+}
diff --git a/src/datatypes/src/series/raw/tests.rs b/src/datatypes/src/series/raw/tests.rs
new file mode 100644
index 00000000000..ea2d4f80193
--- /dev/null
+++ b/src/datatypes/src/series/raw/tests.rs
@@ -0,0 +1,275 @@
+// 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::sync::Arc;
+
+use arrow::array::Float64Array;
+use arrow::buffer::{Buffer, ScalarBuffer};
+
+use super::*;
+use crate::series::samples::{FieldSpec, SampleRef, Samples, SeriesFields, SeriesSchema};
+use crate::series::time::{EvalGrid, GridSpec};
+use crate::series::{Label, MetricName};
+
+fn schema(names: &[&str]) -> SeriesSchema {
+    SeriesSchema {
+        fields: names
+            .iter()
+            .map(|name| FieldSpec {
+                name: Arc::from(*name),
+            })
+            .collect::<Vec<_>>()
+            .into(),
+        eval_grid: None,
+    }
+}
+
+fn metric() -> MetricName {
+    MetricName::try_new(
+        Some(Arc::from("cpu")),
+        vec![Label {
+            name: Arc::from("host"),
+            value: Arc::from("node-a"),
+        }],
+    )
+    .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 {
+    let mut fields = vec![Samples::from_float(Float64Array::from(first))];
+    if let Some(second) = second {
+        fields.push(Samples::from_float(Float64Array::from(second)));
+    }
+    SeriesFields::try_from_vec(fields).unwrap()
+}
+
+#[test]
+fn constructors_reject_invalid_chunk_boundaries_and_unordered_timestamps() {
+    let schema = schema(&["value"]);
+    let key = Buffer::from(vec![1_u8]);
+    let fields = float_fields(vec![Some(1.0), Some(2.0)], None);
+
+    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(
+            &schema,
+            key.clone(),
+            None,
+            times(vec![1, 2]),
+            float_fields(vec![Some(1.0), Some(2.0)], Some(vec![Some(3.0), Some(4.0)])),
+            true,
+        )
+        .is_err()
+    );
+    assert!(
+        RawSeriesChunk::try_new(
+            &schema,
+            key.clone(),
+            None,
+            times(vec![1, 2]),
+            float_fields(vec![Some(1.0)], None),
+            true,
+        )
+        .is_err()
+    );
+    assert!(
+        RawSeriesChunk::try_new(
+            &schema,
+            key.clone(),
+            None,
+            times(vec![]),
+            float_fields(vec![], None),
+            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(),
+        eval_grid: Some(grid),
+    };
+    assert!(
+        RawSeriesChunk::try_new(&grid_schema, key, None, times(vec![10, 15]), fields, true,)
+            .is_err()
+    );
+}
+
+#[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 = RawTimestamps::try_new(time_owner.clone()).unwrap();
+    let time = time_source.slice(1, 4).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);
+    let stale = f64::from_bits(0x7ff0_0000_0000_0002);
+    let float_owner = Float64Array::from(vec![
+        Some(90.0),
+        Some(1.0),
+        None,
+        Some(nan_payload),
+        Some(stale),
+        Some(80.0),
+    ]);
+    let second_owner = Float64Array::from(vec![
+        Some(190.0),
+        Some(11.0),
+        Some(12.0),
+        None,
+        Some(14.0),
+        Some(180.0),
+    ]);
+    let fields = SeriesFields::try_from_vec(vec![
+        Samples::from_float(float_owner.slice(1, 4)),
+        Samples::from_float(second_owner.slice(1, 4)),
+    ])
+    .unwrap();
+    let key_source = Buffer::from(vec![0_u8, 0xA1, 0xB2, 0xC3, 0xD4, 0xFF]);
+    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, primary_key, Some(metric()), time, fields, true).unwrap();
+    let Some(crate::series::samples::DenseSamplesRef::Float(first)) =
+        chunk.fields.get(0).unwrap().dense()
+    else {
+        panic!()
+    };
+    assert_eq!(
+        first.values().as_ptr(),
+        float_owner.values().as_ptr().wrapping_add(1)
+    );
+
+    drop(key_source);
+    drop(time_source);
+    drop(time_owner);
+    drop(float_owner);
+    drop(second_owner);
+    assert_eq!(chunk.primary_key.as_slice(), &[0xA1, 0xB2, 0xC3, 0xD4]);
+    assert_eq!(
+        (0..4)
+            .map(|i| chunk.time.timestamp(i).unwrap())
+            .collect::<Vec<_>>(),
+        [10, 20, 30, 40]
+    );
+    assert!(matches!(
+        chunk.fields.get(0).unwrap().sample(0),
+        SampleRef::Float(1.0)
+    ));
+    assert!(matches!(
+        chunk.fields.get(0).unwrap().sample(1),
+        SampleRef::Missing
+    ));
+    assert!(
+        matches!(chunk.fields.get(0).unwrap().sample(2), SampleRef::Float(v) if v.to_bits() == nan_payload.to_bits())
+    );
+    assert!(
+        matches!(chunk.fields.get(0).unwrap().sample(3), SampleRef::Float(v) if v.to_bits() == stale.to_bits())
+    );
+    assert!(matches!(
+        chunk.fields.get(1).unwrap().sample(0),
+        SampleRef::Float(11.0)
+    ));
+    assert!(matches!(
+        chunk.fields.get(1).unwrap().sample(1),
+        SampleRef::Float(12.0)
+    ));
+    assert!(matches!(
+        chunk.fields.get(1).unwrap().sample(2),
+        SampleRef::Missing
+    ));
+    assert!(matches!(
+        chunk.fields.get(1).unwrap().sample(3),
+        SampleRef::Float(14.0)
+    ));
+    assert_eq!(chunk.metric.as_ref().unwrap().name(), Some("cpu"));
+    assert_eq!(chunk.metric.as_ref().unwrap().get("host"), Some("node-a"));
+    assert!(chunk.is_last);
+}
+
+#[test]
+fn raw_timestamps_check_bounds_and_retain_exact_end_slices() {
+    let source: ScalarBuffer<i64> = vec![4, 8, 12].into();
+    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());
+    assert!(axis.slice(usize::MAX, 0).is_err());
+    let empty = axis.slice(3, 0).unwrap();
+    assert!(empty.is_empty());
+    assert_eq!(axis.timestamp(3), None);
+    let end = axis.slice(1, 2).unwrap();
+    drop(axis);
+    drop(source);
+    assert_eq!(end.timestamp(0), Some(8));
+    assert_eq!(end.timestamp(1), Some(12));
+}
+
+#[test]
+fn raw_chunks_preserve_native_unmaterialized_and_explicit_metrics() {
+    let schema = schema(&["value"]);
+    let fields = float_fields(vec![Some(2.0)], None);
+    let native = RawSeriesChunk::try_new(
+        &schema,
+        Buffer::from(vec![0_u8, 0x12, 0x34, 0x56]),
+        None,
+        times(vec![7]),
+        fields.clone(),
+        false,
+    )
+    .unwrap();
+    let tagless = RawSeriesChunk::try_new(
+        &schema,
+        Buffer::from(Vec::<u8>::new()),
+        None,
+        times(vec![7]),
+        fields.clone(),
+        false,
+    )
+    .unwrap();
+    let explicit = RawSeriesChunk::try_new(
+        &schema,
+        Buffer::from(vec![0_u8, 0xAB, 0xCD, 0xEF]),
+        Some(metric()),
+        times(vec![7]),
+        fields,
+        false,
+    )
+    .unwrap();
+    assert_eq!(native.primary_key.as_slice(), &[0, 0x12, 0x34, 0x56]);
+    assert!(native.metric.is_none());
+    assert!(tagless.primary_key.is_empty());
+    assert!(tagless.metric.is_none());
+    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"),
+        Some("node-a")
+    );
+    assert!(!explicit.is_last);
+}
diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs
index 6577880b415..59a959c09f5 100644
--- a/src/store-api/src/region_engine.rs
+++ b/src/store-api/src/region_engine.rs
@@ -17,6 +17,7 @@
 use std::any::Any;
 use std::collections::HashMap;
 use std::fmt::{Debug, Display};
+use std::pin::Pin;
 use std::sync::{Arc, Mutex};
 
 use api::greptime_proto::v1::meta::{GrantedRegion as PbGrantedRegion, RegionRole as PbRegionRole};
@@ -29,6 +30,10 @@ use common_time::Timestamp;
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_plan::{DisplayAs, DisplayFormatType, PhysicalExpr};
 use datatypes::schema::SchemaRef;
+use datatypes::series::SeriesOrdering;
+use datatypes::series::raw::RawSeriesChunk;
+use datatypes::series::samples::SeriesSchema;
+use futures::Stream;
 use futures::future::join_all;
 use serde::{Deserialize, Serialize};
 use tokio::sync::Semaphore;
@@ -38,7 +43,7 @@ use crate::metadata::RegionMetadataRef;
 use crate::region_request::{
     BatchRegionDdlRequest, RegionCatchupRequest, RegionOpenRequest, RegionRequest,
 };
-use crate::storage::{FileId, RegionId, ScanRequest, SequenceNumber};
+use crate::storage::{ColumnId, FileId, RegionId, ScanRequest, SequenceNumber};
 
 /// The settable region role state.
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
@@ -467,6 +472,106 @@ pub struct QueryScanContext {
     pub explain_verbose: bool,
 }
 
+/// A tag bound to a physical column in the scan's metadata snapshot.
+#[derive(Clone, Debug)]
+pub struct BoundTag {
+    /// Physical region column identifier for this logical tag.
+    pub column_id: ColumnId,
+    /// Logical tag name visible to the query.
+    pub name: Arc<str>,
+}
+
+/// Bound source for materializing labels from raw series chunks.
+#[derive(Clone, Debug)]
+pub enum RawLabelSource {
+    /// Labels are encoded in the complete primary key and use this scan-bound metadata.
+    PrimaryKey {
+        /// Metric name associated with the bound logical source.
+        metric_name: Arc<str>,
+        /// Physical metadata snapshot used to decode primary-key values.
+        physical_metadata: RegionMetadataRef,
+        /// Logical tag names mapped to physical column identifiers.
+        logical_tags: Arc<[BoundTag]>,
+    },
+    /// The stream supplies a complete metric name and label set in each chunk, not a partial
+    /// label projection; `Some(metric)` is a complete name/label set (the name itself may be
+    /// absent).
+    Explicit,
+}
+
+/// A stream of raw series chunks with stable schema, label binding, and ordering.
+///
+/// Every series has strictly increasing timestamps within and across chunks; timestamps are
+/// unique per series. `Interleaved` permits a series to resume after another series; `Contiguous`
+/// and `KeyThenTime` guarantee each series' chunks are contiguous. `KeyThenTime` uses the complete
+/// stream identity compatible with its label binding, not just a TSID or hash. These are
+/// per-stream guarantees, not query-global order, partition ownership, or finality. `is_last`
+/// marks the last chunk of that series in this stream only; merging requires recomputing
+/// per-input-series finalization across all related inputs, not aggregate-group finality.
+/// `RawLabelSource` keeps label interpretation self-contained through transport. Stream finality
+/// applies to all related input series, not aggregate groups or vector-match state.
+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>;
+    /// Returns scan-bound label metadata. `PrimaryKey` chunks leave `metric` as `None`, meaning
+    /// labels are undecoded, not that the metric name is missing; `Explicit` chunks supply
+    /// `Some` with the complete name/label set, not a partial projection. The metric name itself
+    /// may be absent.
+    fn label_source(&self) -> &RawLabelSource;
+    /// Returns this stream's ordering guarantee.
+    fn ordering(&self) -> SeriesOrdering;
+}
+
+/// 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 name and labels, not a partial label projection; or `None` for
+    /// primary-key label binding (undecoded, not nameless). The metric name itself may be absent.
+    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` (undecoded); `Explicit` output has a complete
+/// `Some(metric)` matching each chunk. Ordering describes the stream, not ordinal enumeration.
+///
+/// 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 label binding shared by series metadata and every reader, without opening or
+    /// polling a reader; the binding is self-contained for remote consumers.
+    fn label_source(&self) -> &RawLabelSource;
+
+    /// 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.
@@ -499,6 +604,42 @@ pub trait RegionScanner: Debug + DisplayAs + Send {
         partition: usize,
     ) -> Result<SendableRecordBatchStream, BoxedError>;
 
+    /// Scans a partition as raw series when this source supports that path.
+    ///
+    /// Implementations may return `None` when raw capability is absent before execution, permitting
+    /// a legacy scan instead. `Some` selects this entrypoint, including for an empty successful
+    /// stream. Errors propagate; callers must not replay a failed raw scan through the legacy path.
+    /// Use only one entrypoint for a given scanner partition per execution.
+    ///
+    /// # Panics
+    ///
+    /// The default implementation panics. Implementations must override this method to support
+    /// raw series scanning.
+    fn scan_series_partition(
+        &self,
+        _ctx: &QueryScanContext,
+        _metrics_set: &ExecutionPlanMetricsSet,
+        _partition: usize,
+    ) -> Result<Option<SendableRawSeriesStream>, BoxedError> {
+        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;
 
@@ -737,12 +878,93 @@ impl RegionStatistic {
     }
 }
 
+#[cfg(test)]
+#[path = "region_engine/prepared_series_tests.rs"]
+mod prepared_series_tests;
+
 #[cfg(test)]
 mod tests {
+    use common_recordbatch::{RecordBatch, RecordBatches};
+    use datatypes::prelude::ConcreteDataType;
+    use datatypes::schema::ColumnSchema;
+    use datatypes::vectors::{Int32Vector, TimestampMillisecondVector, VectorRef};
+    use futures::TryStreamExt;
     use serde_json::json;
 
     use super::*;
 
+    #[tokio::test]
+    async fn single_partition_scanner_raw_scan_panics_and_preserves_legacy_read() {
+        use std::panic::{AssertUnwindSafe, catch_unwind};
+
+        use api::v1::SemanticType;
+        use datatypes::arrow::array::Int32Array;
+        use datatypes::schema::Schema as ArrowSchema;
+
+        use crate::metadata::{ColumnMetadata, RegionMetadataBuilder};
+
+        let schema = Arc::new(ArrowSchema::new(vec![
+            ColumnSchema::new("value", ConcreteDataType::int32_datatype(), false),
+            ColumnSchema::new(
+                "ts",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            ),
+        ]));
+        let columns: Vec<VectorRef> = vec![
+            Arc::new(Int32Vector::from_slice([42])),
+            Arc::new(TimestampMillisecondVector::from_slice([7])),
+        ];
+        let batch = RecordBatch::new(schema.clone(), columns).unwrap();
+        let batches = RecordBatches::try_new(schema, vec![batch]).unwrap();
+        let mut metadata_builder = RegionMetadataBuilder::new(RegionId::new(1234, 1));
+        metadata_builder.push_column_metadata(ColumnMetadata {
+            column_schema: ColumnSchema::new("value", ConcreteDataType::int32_datatype(), false),
+            semantic_type: SemanticType::Field,
+            column_id: 1,
+        });
+        metadata_builder.push_column_metadata(ColumnMetadata {
+            column_schema: ColumnSchema::new(
+                "ts",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            ),
+            semantic_type: SemanticType::Timestamp,
+            column_id: 2,
+        });
+        let metadata = Arc::new(metadata_builder.build().unwrap());
+        let scanner = SinglePartitionScanner::new(batches.as_stream(), false, metadata, None);
+        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)
+        }));
+        assert!(raw_scan.is_err());
+        let legacy = scanner.scan_partition(&ctx, &metrics, 0).unwrap();
+        let legacy_batches = legacy.try_collect::<Vec<_>>().await.unwrap();
+        assert_eq!(legacy_batches.len(), 1);
+        assert_eq!(legacy_batches[0].num_rows(), 1);
+        assert_eq!(
+            legacy_batches[0]
+                .column(0)
+                .as_ref()
+                .as_any()
+                .downcast_ref::<Int32Array>()
+                .unwrap()
+                .value(0),
+            42
+        );
+    }
+
     #[test]
     fn region_statistic_deserializes_without_query_stats() {
         let statistic: RegionStatistic = serde_json::from_value(json!({
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..b44b543a3ef
--- /dev/null
+++ b/src/store-api/src/region_engine/prepared_series_tests.rs
@@ -0,0 +1,470 @@
+// 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::SeriesOrdering;
+use datatypes::series::raw::{RawSeriesChunk, 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 label_source(&self) -> &RawLabelSource {
+        &self.label_source
+    }
+
+    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) -> SeriesOrdering {
+        SeriesOrdering::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 assert_same_label_source(expected: &RawLabelSource, actual: &RawLabelSource) {
+    match (expected, actual) {
+        (RawLabelSource::Explicit, RawLabelSource::Explicit) => {}
+        (
+            RawLabelSource::PrimaryKey {
+                metric_name: expected_name,
+                physical_metadata: expected_metadata,
+                logical_tags: expected_tags,
+            },
+            RawLabelSource::PrimaryKey {
+                metric_name: actual_name,
+                physical_metadata: actual_metadata,
+                logical_tags: actual_tags,
+            },
+        ) => {
+            assert_eq!(expected_name, actual_name);
+            assert!(Arc::ptr_eq(expected_metadata, actual_metadata));
+            assert_eq!(expected_tags.len(), actual_tags.len());
+            for (expected, actual) in expected_tags.iter().zip(actual_tags.iter()) {
+                assert_eq!(expected.name, actual.name);
+                assert_eq!(expected.column_id, actual.column_id);
+            }
+        }
+        _ => panic!("label source differs"),
+    }
+}
+
+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: vec![crate::region_engine::BoundTag {
+                    column_id: 1,
+                    name: Arc::from("host"),
+                }]
+                .into(),
+            }
+        },
+        polls: Arc::new(AtomicUsize::new(0)),
+        drops: Arc::new(AtomicUsize::new(0)),
+    })
+}
+
+#[tokio::test]
+async fn prepared_source_label_binding_is_available_before_readers_and_matches_chunks() {
+    for explicit in [false, true] {
+        let source = source(explicit);
+        assert_eq!(source.num_series(), 2);
+        match (explicit, source.label_source()) {
+            (true, RawLabelSource::Explicit) => {}
+            (
+                false,
+                RawLabelSource::PrimaryKey {
+                    metric_name,
+                    physical_metadata,
+                    logical_tags,
+                },
+            ) => {
+                assert_eq!(metric_name.as_ref(), "cpu");
+                assert_eq!(
+                    physical_metadata
+                        .column_by_id(1)
+                        .unwrap()
+                        .column_schema
+                        .name,
+                    "host"
+                );
+                assert_eq!(logical_tags.len(), 1);
+                assert_eq!(logical_tags[0].column_id, 1);
+                assert_eq!(logical_tags[0].name.as_ref(), "host");
+            }
+            _ => panic!("unexpected source label binding"),
+        }
+        assert_eq!(source.polls.load(Ordering::SeqCst), 0);
+        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(), SeriesOrdering::Contiguous);
+        assert_same_label_source(source.label_source(), empty.label_source());
+        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!(end.ordering(), SeriesOrdering::Contiguous);
+        assert_same_label_source(source.label_source(), end.label_source());
+        assert_eq!(source.polls.load(Ordering::SeqCst), 0);
+
+        let reader = source.read_series(0..2).unwrap();
+        assert_same_label_source(source.label_source(), reader.label_source());
+        assert_eq!(source.polls.load(Ordering::SeqCst), 0);
+        let chunks = futures::TryStreamExt::try_collect::<Vec<_>>(reader)
+            .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!(matches!(source.label_source(), RawLabelSource::Explicit));
+    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(std::ops::Range { start: 2, end: 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]))
+    );
+}