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

@@ -25,6 +25,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 {

src/datatypes/src/series/raw.rs

@@ -85,20 +85,11 @@ impl RawTimestamps {
     }
 }
 
-/// Producer ordering guarantee for chunks in one input stream. Producers MUST emit all chunks
-/// for each read key contiguously; adapters validate adjacent chunk conditions but do not certify
-/// nonadjacent recurrence for [`Self::Contiguous`]. An `A/B/A` sequence violates the producer
-/// contract even when a consumer cannot detect that recurrence.
-#[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)]

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>;
 

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);