From a8bb74cbf9a1659afd2dac881d7873dc21d931df Mon Sep 17 00:00:00 2001
From: discord9 <55937128+discord9@users.noreply.github.com>
Date: Thu, 15 Oct 2026 13:32:32 +0800
Subject: [PATCH 1/6] feat(storage): define raw series and optional scanner
 contract

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
---
 src/datatypes/src/series.rs           |   1 +
 src/datatypes/src/series/raw.rs       | 140 +++++++++++++
 src/datatypes/src/series/raw/tests.rs | 286 ++++++++++++++++++++++++++
 src/store-api/src/region_engine.rs    | 136 +++++++++++-
 4 files changed, 562 insertions(+), 1 deletion(-)
 create mode 100644 src/datatypes/src/series/raw.rs
 create mode 100644 src/datatypes/src/series/raw/tests.rs

diff --git a/src/datatypes/src/series.rs b/src/datatypes/src/series.rs
index d5d11bfd5ec..ad3bf52aa52 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;
 
diff --git a/src/datatypes/src/series/raw.rs b/src/datatypes/src/series/raw.rs
new file mode 100644
index 00000000000..1ff8cc873e0
--- /dev/null
+++ b/src/datatypes/src/series/raw.rs
@@ -0,0 +1,140 @@
+// 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};
+
+/// Explicit source timestamps in milliseconds for a raw series chunk.
+#[derive(Clone, Debug)]
+pub enum RawTimeAxis {
+    /// Source timestamps in strictly increasing order.
+    Explicit(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> {
+        if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) {
+            return invalid_series("raw timestamps must be strictly increasing");
+        }
+        Ok(Self::Explicit(timestamps))
+    }
+
+    /// Returns the number of timestamps.
+    pub fn len(&self) -> usize {
+        match self {
+            Self::Explicit(values) => values.len(),
+        }
+    }
+
+    /// Returns whether the axis has no timestamps.
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// 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(),
+        }
+    }
+
+    /// 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 time axis 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))),
+        }
+    }
+}
+
+/// 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.
+    Contiguous,
+    /// Chunks are ordered by read key and then by timestamp.
+    ReadKeyThenTime,
+}
+
+/// One raw source batch, with fields aligned to its timestamps.
+///
+/// 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 read_key: Buffer,
+    /// `None` leaves labels unmaterialized in the complete read key; `Some` contains
+    /// complete explicit labels.
+    pub metric: Option<MetricName>,
+    /// Source timestamps aligned with every field.
+    pub time: RawTimeAxis,
+    /// 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.
+    pub is_last: bool,
+}
+
+impl RawSeriesChunk {
+    /// Creates a nonempty chunk after checking its explicit 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,
+        metric: Option<MetricName>,
+        time: RawTimeAxis,
+        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");
+        }
+        let RawTimeAxis::Explicit(timestamps) = &time;
+        if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) {
+            return invalid_series("raw series timestamps must be strictly increasing");
+        }
+        Ok(Self {
+            read_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..3229f647fd9
--- /dev/null
+++ b/src/datatypes/src/series/raw/tests.rs
@@ -0,0 +1,286 @@
+// 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>) -> RawTimeAxis {
+    RawTimeAxis::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!(RawTimeAxis::try_new(vec![1_i64, 1].into()).is_err());
+    assert!(RawTimeAxis::try_new(vec![2_i64, 1].into()).is_err());
+
+    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()
+    );
+
+    // 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(),
+        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 = RawTimeAxis::try_new(time_owner.clone()).unwrap();
+    let time = time_source.slice(1, 4).unwrap();
+    let RawTimeAxis::Explicit(time_buffer) = &time;
+    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 read_key = key_source.slice_with_length(1, 4);
+    assert_eq!(read_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();
+    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.read_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_time_axis_checks_bounds_and_retains_exact_end_slices() {
+    let source: ScalarBuffer<i64> = vec![4, 8, 12].into();
+    let axis = RawTimeAxis::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.read_key.as_slice(), &[0, 0x12, 0x34, 0x56]);
+    assert!(native.metric.is_none());
+    assert!(tagless.read_key.is_empty());
+    assert!(tagless.metric.is_none());
+    assert_eq!(explicit.read_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..82183f77ec3 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,9 @@ use common_time::Timestamp;
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_plan::{DisplayAs, DisplayFormatType, PhysicalExpr};
 use datatypes::schema::SchemaRef;
+use datatypes::series::raw::{RawSeriesChunk, RawSeriesOrdering};
+use datatypes::series::samples::SeriesSchema;
+use futures::Stream;
 use futures::future::join_all;
 use serde::{Deserialize, Serialize};
 use tokio::sync::Semaphore;
@@ -38,7 +42,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 +471,49 @@ 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.
+    Explicit,
+}
+
+/// 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.
+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 a complete metric name and labels.
+    fn label_source(&self) -> &RawLabelSource;
+    /// Returns this stream's ordering guarantee.
+    fn ordering(&self) -> RawSeriesOrdering;
+}
+
+/// A pinned raw-series stream that can be sent between threads.
+pub type SendableRawSeriesStream = Pin<Box<dyn RawSeriesStream>>;
+
 /// 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 +546,21 @@ pub trait RegionScanner: Debug + DisplayAs + Send {
         partition: usize,
     ) -> Result<SendableRecordBatchStream, BoxedError>;
 
+    /// Scans a partition as raw series when this source supports that path.
+    ///
+    /// `None` means raw capability is absent before execution and permits 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.
+    fn scan_series_partition(
+        &self,
+        _ctx: &QueryScanContext,
+        _metrics_set: &ExecutionPlanMetricsSet,
+        _partition: usize,
+    ) -> Result<Option<SendableRawSeriesStream>, 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;
 
@@ -739,10 +801,82 @@ impl RegionStatistic {
 
 #[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_preserves_legacy_fallback_after_raw_capability_miss() {
+        let schema = Arc::new(datatypes::schema::Schema::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 =
+            crate::metadata::RegionMetadataBuilder::new(RegionId::new(1234, 1));
+        metadata_builder.push_column_metadata(crate::metadata::ColumnMetadata {
+            column_schema: ColumnSchema::new("value", ConcreteDataType::int32_datatype(), false),
+            semantic_type: api::v1::SemanticType::Field,
+            column_id: 1,
+        });
+        metadata_builder.push_column_metadata(crate::metadata::ColumnMetadata {
+            column_schema: ColumnSchema::new(
+                "ts",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            ),
+            semantic_type: api::v1::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
+                .scan_series_partition(&ctx, &metrics, 0)
+                .unwrap()
+                .is_none()
+        );
+        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::<datatypes::arrow::array::Int32Array>()
+                .unwrap()
+                .value(0),
+            42
+        );
+
+        let second = scanner
+            .scan_partition(&ctx, &metrics, 0)
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        assert!(second.is_empty());
+    }
+
     #[test]
     fn region_statistic_deserializes_without_query_stats() {
         let statistic: RegionStatistic = serde_json::from_value(json!({

From 4ad0ce71fe5f1caf4c34947c4a0a7fa862d2cb0b Mon Sep 17 00:00:00 2001
From: discord9 <55937128+discord9@users.noreply.github.com>
Date: Thu, 15 Oct 2026 19:35:48 +0800
Subject: [PATCH 2/6] fix(store-api): mark default raw series scan as todo

Leave the raw entrypoint explicitly unimplemented until a scanner overrides it. Document the default panic and keep the real scanner regression proving that its legacy input remains readable after the placeholder is caught.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
---
 src/store-api/src/region_engine.rs | 27 +++++++++++++++------------
 1 file changed, 15 insertions(+), 12 deletions(-)

diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs
index 82183f77ec3..746eacddc51 100644
--- a/src/store-api/src/region_engine.rs
+++ b/src/store-api/src/region_engine.rs
@@ -548,17 +548,22 @@ pub trait RegionScanner: Debug + DisplayAs + Send {
 
     /// Scans a partition as raw series when this source supports that path.
     ///
-    /// `None` means raw capability is absent before execution and permits 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.
+    /// 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> {
-        Ok(None)
+        todo!()
     }
 
     /// Check if there is any predicate exclude region partition exprs that may be executed in this scanner.
@@ -811,7 +816,7 @@ mod tests {
     use super::*;
 
     #[tokio::test]
-    async fn single_partition_scanner_preserves_legacy_fallback_after_raw_capability_miss() {
+    async fn single_partition_scanner_raw_scan_panics_and_preserves_legacy_read() {
         let schema = Arc::new(datatypes::schema::Schema::new(vec![
             ColumnSchema::new("value", ConcreteDataType::int32_datatype(), false),
             ColumnSchema::new(
@@ -847,12 +852,10 @@ mod tests {
         let ctx = QueryScanContext::default();
         let metrics = ExecutionPlanMetricsSet::new();
 
-        assert!(
-            scanner
-                .scan_series_partition(&ctx, &metrics, 0)
-                .unwrap()
-                .is_none()
-        );
+        let raw_scan = std::panic::catch_unwind(std::panic::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);

From 5fa4fed582381a7c7e0494b32eec49b61c9ad2ab Mon Sep 17 00:00:00 2001
From: discord9 <55937128+discord9@users.noreply.github.com>
Date: Thu, 15 Oct 2026 20:53:35 +0800
Subject: [PATCH 3/6] fix(datatypes): share raw axis validation and expose
 explicit buffer

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
---
 src/datatypes/src/series/raw.rs       | 31 ++++++++++++++++++------
 src/datatypes/src/series/raw/tests.rs |  2 +-
 src/store-api/src/region_engine.rs    | 34 +++++++++++++--------------
 3 files changed, 42 insertions(+), 25 deletions(-)

diff --git a/src/datatypes/src/series/raw.rs b/src/datatypes/src/series/raw.rs
index 1ff8cc873e0..6c5b04ceac7 100644
--- a/src/datatypes/src/series/raw.rs
+++ b/src/datatypes/src/series/raw.rs
@@ -23,6 +23,7 @@ use crate::series::samples::{SeriesFields, SeriesSchema};
 use crate::series::{MetricName, invalid_series};
 
 /// Explicit source timestamps in milliseconds for a raw series chunk.
+#[non_exhaustive]
 #[derive(Clone, Debug)]
 pub enum RawTimeAxis {
     /// Source timestamps in strictly increasing order.
@@ -32,10 +33,29 @@ pub enum RawTimeAxis {
 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> {
-        if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) {
-            return invalid_series("raw timestamps must be strictly increasing");
+        let axis = Self::Explicit(timestamps);
+        axis.try_len()?;
+        Ok(axis)
+    }
+
+    /// 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())
+            }
         }
-        Ok(Self::Explicit(timestamps))
     }
 
     /// Returns the number of timestamps.
@@ -125,10 +145,7 @@ impl RawSeriesChunk {
         if time.is_empty() {
             return invalid_series("raw series chunks must not be empty");
         }
-        let RawTimeAxis::Explicit(timestamps) = &time;
-        if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) {
-            return invalid_series("raw series timestamps must be strictly increasing");
-        }
+        time.try_len()?;
         Ok(Self {
             read_key,
             metric,
diff --git a/src/datatypes/src/series/raw/tests.rs b/src/datatypes/src/series/raw/tests.rs
index 3229f647fd9..ee982148d73 100644
--- a/src/datatypes/src/series/raw/tests.rs
+++ b/src/datatypes/src/series/raw/tests.rs
@@ -133,7 +133,7 @@ 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 = time_source.slice(1, 4).unwrap();
-    let RawTimeAxis::Explicit(time_buffer) = &time;
+    let time_buffer = time.as_explicit().unwrap();
     assert_eq!(time_buffer.as_ptr(), time_owner.as_ptr().wrapping_add(1));
 
     let nan_payload = f64::from_bits(0x7ff8_0000_0000_0042);
diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs
index 746eacddc51..48017c94f0f 100644
--- a/src/store-api/src/region_engine.rs
+++ b/src/store-api/src/region_engine.rs
@@ -817,7 +817,15 @@ mod tests {
 
     #[tokio::test]
     async fn single_partition_scanner_raw_scan_panics_and_preserves_legacy_read() {
-        let schema = Arc::new(datatypes::schema::Schema::new(vec![
+        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",
@@ -831,20 +839,19 @@ mod tests {
         ];
         let batch = RecordBatch::new(schema.clone(), columns).unwrap();
         let batches = RecordBatches::try_new(schema, vec![batch]).unwrap();
-        let mut metadata_builder =
-            crate::metadata::RegionMetadataBuilder::new(RegionId::new(1234, 1));
-        metadata_builder.push_column_metadata(crate::metadata::ColumnMetadata {
+        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: api::v1::SemanticType::Field,
+            semantic_type: SemanticType::Field,
             column_id: 1,
         });
-        metadata_builder.push_column_metadata(crate::metadata::ColumnMetadata {
+        metadata_builder.push_column_metadata(ColumnMetadata {
             column_schema: ColumnSchema::new(
                 "ts",
                 ConcreteDataType::timestamp_millisecond_datatype(),
                 false,
             ),
-            semantic_type: api::v1::SemanticType::Timestamp,
+            semantic_type: SemanticType::Timestamp,
             column_id: 2,
         });
         let metadata = Arc::new(metadata_builder.build().unwrap());
@@ -852,7 +859,8 @@ mod tests {
         let ctx = QueryScanContext::default();
         let metrics = ExecutionPlanMetricsSet::new();
 
-        let raw_scan = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+        // 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());
@@ -865,19 +873,11 @@ mod tests {
                 .column(0)
                 .as_ref()
                 .as_any()
-                .downcast_ref::<datatypes::arrow::array::Int32Array>()
+                .downcast_ref::<Int32Array>()
                 .unwrap()
                 .value(0),
             42
         );
-
-        let second = scanner
-            .scan_partition(&ctx, &metrics, 0)
-            .unwrap()
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-        assert!(second.is_empty());
     }
 
     #[test]

From d639f1569e317ef32c11b1b588f44048d93f0054 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 4/6] 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<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,
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<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"),
diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs
index 48017c94f0f..d9c4fa477f6 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<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.
@@ -566,6 +612,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;
 
@@ -804,6 +866,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};
@@ -859,6 +925,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)
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<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]))
+    );
+}

From 650ee58c226e827b21f9f75a3961f60e6102c5c8 Mon Sep 17 00:00:00 2001
From: discord9 <55937128+discord9@users.noreply.github.com>
Date: Fri, 16 Oct 2026 15:45:42 +0800
Subject: [PATCH 5/6] feat(datatypes): share series ordering and expose source
 label binding

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
---
 src/datatypes/src/series.rs                   | 21 ++++
 src/datatypes/src/series/raw.rs               | 12 +--
 src/store-api/src/region_engine.rs            | 40 +++++---
 .../region_engine/prepared_series_tests.rs    | 97 +++++++++++++++----
 4 files changed, 129 insertions(+), 41 deletions(-)

diff --git a/src/datatypes/src/series.rs b/src/datatypes/src/series.rs
index ad3bf52aa52..860358eff20 100644
--- a/src/datatypes/src/series.rs
+++ b/src/datatypes/src/series.rs
@@ -24,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
index 9c18ae2941a..da4ba487204 100644
--- a/src/datatypes/src/series/raw.rs
+++ b/src/datatypes/src/series/raw.rs
@@ -85,17 +85,11 @@ impl RawTimestamps {
     }
 }
 
-/// Ordering guarantee for chunks in one input stream.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum RawSeriesOrdering {
-    /// Each complete primary key's chunks are contiguous; distinct keys need not be ordered.
-    Contiguous,
-    /// Chunks are ordered by complete primary key and then by timestamp.
-    ReadKeyThenTime,
-}
-
 /// 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)]
diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs
index d9c4fa477f6..59a959c09f5 100644
--- a/src/store-api/src/region_engine.rs
+++ b/src/store-api/src/region_engine.rs
@@ -30,7 +30,8 @@ use common_time::Timestamp;
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_plan::{DisplayAs, DisplayFormatType, PhysicalExpr};
 use datatypes::schema::SchemaRef;
-use datatypes::series::raw::{RawSeriesChunk, RawSeriesOrdering};
+use datatypes::series::SeriesOrdering;
+use datatypes::series::raw::RawSeriesChunk;
 use datatypes::series::samples::SeriesSchema;
 use futures::Stream;
 use futures::future::join_all;
@@ -492,27 +493,33 @@ pub enum RawLabelSource {
         /// 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.
+    /// 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.
 ///
-/// 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; 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.
+/// 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 a complete metric name and labels.
+    /// `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) -> RawSeriesOrdering;
+    fn ordering(&self) -> SeriesOrdering;
 }
 
 /// A pinned raw-series stream that can be sent between threads.
@@ -523,7 +530,8 @@ pub type SendableRawSeriesStream = Pin<Box<dyn RawSeriesStream>>;
 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.
+    /// 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>,
 }
 
@@ -533,9 +541,9 @@ pub struct SeriesInfo {
 /// 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.
+/// 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,
@@ -550,6 +558,10 @@ 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>;
 
diff --git a/src/store-api/src/region_engine/prepared_series_tests.rs b/src/store-api/src/region_engine/prepared_series_tests.rs
index 05882ddcc70..a489598b567 100644
--- a/src/store-api/src/region_engine/prepared_series_tests.rs
+++ b/src/store-api/src/region_engine/prepared_series_tests.rs
@@ -25,7 +25,8 @@ 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::SeriesOrdering;
+use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
 use datatypes::series::samples::{FieldSpec, Samples, SeriesFields, SeriesSchema};
 use datatypes::series::{Label, MetricName};
 use futures::{Stream, StreamExt};
@@ -53,6 +54,10 @@ impl PreparedSeriesSource for MemorySource {
         self.series.len()
     }
 
+    fn label_source(&self) -> &RawLabelSource {
+        &self.label_source
+    }
+
     fn series_info(&self, ordinal: usize) -> Result<SeriesInfo, BoxedError> {
         self.series
             .get(ordinal)
@@ -128,8 +133,8 @@ impl RawSeriesStream for MemoryReader {
         &self.label_source
     }
 
-    fn ordering(&self) -> RawSeriesOrdering {
-        RawSeriesOrdering::Contiguous
+    fn ordering(&self) -> SeriesOrdering {
+        SeriesOrdering::Contiguous
     }
 }
 
@@ -144,6 +149,33 @@ fn metric(host: &str) -> MetricName {
     .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 {
@@ -246,7 +278,11 @@ fn source_from_keys(explicit: bool, keys: Arc<[Buffer]>) -> Arc<MemorySource> {
             RawLabelSource::PrimaryKey {
                 metric_name: Arc::from("cpu"),
                 physical_metadata: Arc::new(primary_key_metadata()),
-                logical_tags: Arc::from([]),
+                logical_tags: vec![crate::region_engine::BoundTag {
+                    column_id: 1,
+                    name: Arc::from("host"),
+                }]
+                .into(),
             }
         },
         polls: Arc::new(AtomicUsize::new(0)),
@@ -255,10 +291,36 @@ fn source_from_keys(explicit: bool, keys: Arc<[Buffer]>) -> Arc<MemorySource> {
 }
 
 #[tokio::test]
-async fn prepared_identity_matches_chunks_in_both_label_modes_including_empty_reader() {
+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]));
@@ -278,11 +340,8 @@ async fn prepared_identity_matches_chunks_in_both_label_modes_including_empty_re
 
         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!(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)
@@ -293,15 +352,16 @@ async fn prepared_identity_matches_chunks_in_both_label_modes_including_empty_re
         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
-        );
+        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 chunks =
-            futures::TryStreamExt::try_collect::<Vec<_>>(source.read_series(0..2).unwrap())
-                .await
-                .unwrap();
+        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));
@@ -331,6 +391,7 @@ async fn prepared_identity_matches_chunks_in_both_label_modes_including_empty_re
 #[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);

From 794389688084ee97a7760d1ae993bcf1c109489e Mon Sep 17 00:00:00 2001
From: discord9 <55937128+discord9@users.noreply.github.com>
Date: Fri, 16 Oct 2026 17:05:00 +0800
Subject: [PATCH 6/6] fix(store-api): retain reversed range test without clippy
 lint

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
---
 src/store-api/src/region_engine/prepared_series_tests.rs | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/src/store-api/src/region_engine/prepared_series_tests.rs b/src/store-api/src/region_engine/prepared_series_tests.rs
index a489598b567..b44b543a3ef 100644
--- a/src/store-api/src/region_engine/prepared_series_tests.rs
+++ b/src/store-api/src/region_engine/prepared_series_tests.rs
@@ -409,7 +409,11 @@ 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!(
+        source
+            .read_series(std::ops::Range { start: 2, end: 1 })
+            .is_err()
+    );
     assert_eq!(source.polls.load(Ordering::SeqCst), 0);
 }