diff --git a/src/datatypes/src/series.rs b/src/datatypes/src/series.rs index f00051ce46f..0de711e2da3 100644 --- a/src/datatypes/src/series.rs +++ b/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 { diff --git a/src/datatypes/src/series/raw.rs b/src/datatypes/src/series/raw.rs index 130f59565a9..da4ba487204 100644 --- a/src/datatypes/src/series/raw.rs +++ b/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)] 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> + Send { /// Returns the schema shared by all chunks; raw schemas have no evaluation grid. fn schema(&self) -> &Arc; /// 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>; 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, } @@ -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; 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 { 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 { Arc::new(SeriesSchema { fields: vec![FieldSpec { @@ -246,7 +278,11 @@ fn source_from_keys(explicit: bool, keys: Arc<[Buffer]>) -> Arc { 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 { } #[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::>(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::>(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::>(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);