feat(mito2): produce typed raw series from RegionScanner with truthful ordering (#9669)

I hereby agree to the terms of the [GreptimeDB CLA](https://github.com/GreptimeTeam/.github/blob/main/CLA.md). ## Refer to a related PR or issue link (optional) Part of the RFC24 native PromQL raw-series stack; tracked in GreptimeTeam/internal-docs#25. All prerequisites are merged into `narthex` (#9650 labels, #9651 preflight, #9653 merge, #9654 source). ## What's changed and what's your intention? Add truthful ordering and shared preflight to mito2 raw series production: - `SeriesScan::prepare_raw_scan`: hoists raw-label/field/predicate preflight out of `scan_series_partition` into a shared method, returning `RawScanPreparation` with schema, fields, and predicates. Rejects dynamic filters explicitly. - `RegionScanner::raw_series_ordering`: returns `KeyThenTime` for Legacy mode, `Contiguous` for TwoPhase — the declared ordering is now truthful instead of hard-coded. - `RegionScanner::raw_series_schema`: schema-only probe for planning-time preflight without I/O. - `RawStream` stores and reports the declared ordering; primary-key monotonicity guard keyed off `KeyThenTime`. **5 files, +1769/-212 lines.** The layer is rebased onto the current `narthex` and carries four commits: the feature, its test commit, and two review fixes. ### Verification - `cargo clippy -p mito2 -p promql --all-targets -- -D warnings` ✅ - `cargo nextest run -p mito2` ✅ - End-to-end: this layer plus the local query entry point executed a real `TQL EVAL (1, 3, '1s')` query through `SeriesScanExec`; the enabled plan contains `NativeInstantManipulateExec` → `SeriesScanExec: source=primary_key` with `output_rows: 4`, while the disabled instance runs the legacy `PromSeriesDivideExec` path and returns identical rows. ## PR Checklist - [x] I have written the necessary rustdoc comments. - [x] I have added the necessary unit tests and integration tests. - [ ] This PR requires documentation updates. - [x] API changes are backward compatible. - [x] Schema or data changes are backward compatible. - [ ] This PR needs to be backported to release branches, and I have added the `backport-<target>` labels (e.g. `backport-v1.3` targets `release/v1.3`).
2026-11-14 06:46:56 +00:00 · 2026-10-21 01:11:44 +08:00
parent 3ad5f96328 e1dba11eea
commit 33832ae99e
5 changed files with 1739 additions and 182 deletions
@@ -31,13 +31,15 @@ use common_recordbatch::{RecordBatchStreamWrapper, SendableRecordBatchStream};
 use common_telemetry::tracing::{self, Instrument};
 use common_telemetry::{debug, warn};
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
-use datafusion::physical_plan::{DisplayAs, DisplayFormatType};
+use datafusion::physical_plan::{DisplayAs, DisplayFormatType, PhysicalExpr};
 use datafusion_common::tree_node::TreeNode;
 use datafusion_expr::{Expr, Volatility};
 use datatypes::arrow::array::BinaryArray;
 use datatypes::arrow::datatypes::DataType as ArrowDataType;
 use datatypes::arrow::record_batch::RecordBatch;
+use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::SchemaRef;
+use datatypes::series::SeriesOrdering;
 use datatypes::series::samples::{FieldSpec, SeriesSchema};
 use datatypes::types::TimestampType;
 use futures::{StreamExt, TryStreamExt};
@@ -79,6 +81,49 @@ const SEND_TIMEOUT: Duration = Duration::from_micros(100);
 /// Maximum number of candidate series retained before distributing assignments.
 const CANDIDATE_SERIES_ASSIGNMENT_THRESHOLD: usize = 1_000_000;

+/// Test-only override of [`CANDIDATE_SERIES_ASSIGNMENT_THRESHOLD`], where `0` means "use the
+/// production constant".
+///
+/// The production build compiles this static, the setter below and the private accessor out, so
+/// the shipped threshold is always the constant above.
+#[cfg(test)]
+static CANDIDATE_SERIES_ASSIGNMENT_THRESHOLD_OVERRIDE: AtomicUsize = AtomicUsize::new(0);
+
+/// Test-only number of candidate assignment rounds sent since the override was last set.
+///
+/// A round is one `send_assignments` call, so a partition that received more than one assignment
+/// sees `>= 2` rounds here. Increments are compiled out of production builds.
+#[cfg(test)]
+static CANDIDATE_ASSIGNMENT_ROUNDS: AtomicUsize = AtomicUsize::new(0);
+
+/// Returns the candidate-series assignment threshold, honouring the test-only override.
+fn candidate_series_assignment_threshold() -> usize {
+    #[cfg(test)]
+    {
+        let threshold = CANDIDATE_SERIES_ASSIGNMENT_THRESHOLD_OVERRIDE.load(Ordering::Relaxed);
+        if threshold > 0 {
+            return threshold;
+        }
+    }
+    CANDIDATE_SERIES_ASSIGNMENT_THRESHOLD
+}
+
+/// Test-only: overrides the candidate-series assignment threshold and resets the round counter.
+///
+/// Pass `0` to restore the production threshold. Tests must reset the override before they
+/// return; a scope guard in `engine::scan_test` does that even on panic.
+#[cfg(test)]
+pub(crate) fn set_candidate_series_assignment_threshold(threshold: usize) {
+    CANDIDATE_SERIES_ASSIGNMENT_THRESHOLD_OVERRIDE.store(threshold, Ordering::Relaxed);
+    CANDIDATE_ASSIGNMENT_ROUNDS.store(0, Ordering::Relaxed);
+}
+
+/// Test-only: returns the number of assignment rounds sent since the override was last set.
+#[cfg(test)]
+pub(crate) fn candidate_assignment_rounds() -> usize {
+    CANDIDATE_ASSIGNMENT_ROUNDS.load(Ordering::Relaxed)
+}
+
 /// Legacy data receivers for output partitions.
 type LegacyReceiverList = Vec<Option<Receiver<Result<SeriesBatch>>>>;

@@ -88,6 +133,12 @@ type CandidateReceiverList = Vec<Option<CandidateReceiver>>;
 /// Candidate assignment senders for output partitions.
 type CandidateSenderList = Vec<Option<mpsc::UnboundedSender<Result<SeriesReaderInput>>>>;

+struct RawScanPreparation {
+    schema: Arc<SeriesSchema>,
+    fields: Vec<(usize, ConcreteDataType)>,
+    predicates: Vec<Arc<dyn PhysicalExpr>>,
+}
+
 struct CandidateReceiver {
    receiver: mpsc::UnboundedReceiver<Result<SeriesReaderInput>>,
    active_receivers: Arc<AtomicUsize>,
@@ -194,6 +245,189 @@ impl SeriesScan {
        }
    }

+    fn prepare_raw_scan(
+        &self,
+        label_source: &RawLabelSource,
+    ) -> Result<Option<RawScanPreparation>, BoxedError> {
+        let input = &self.stream_ctx.input;
+        let metadata = input.mapper.metadata();
+        match label_source {
+            RawLabelSource::PrimaryKey {
+                physical_metadata,
+                logical_tags,
+                ..
+            } => {
+                if !Arc::ptr_eq(metadata, physical_metadata) {
+                    debug!(
+                        "Raw series scan preflight rejected metadata snapshot: region_id={}, physical_region_id={}, schema_version={}, physical_schema_version={}",
+                        metadata.region_id,
+                        physical_metadata.region_id,
+                        metadata.schema_version,
+                        physical_metadata.schema_version,
+                    );
+                    return Err(BoxedError::new(
+                        InvalidRecordBatchSnafu {
+                            reason: "raw label source metadata does not match scanner snapshot"
+                                .to_string(),
+                        }
+                        .build(),
+                    ));
+                }
+                let mut ids = HashSet::new();
+                let mut names = HashSet::new();
+                for tag in logical_tags.iter() {
+                    let column = metadata.column_by_id(tag.column_id).filter(|column| {
+                        column.semantic_type == SemanticType::Tag
+                            && metadata.primary_key.contains(&tag.column_id)
+                    });
+                    let valid = column.is_some_and(|column| {
+                        column.column_schema.name == tag.name.as_ref()
+                            && tag.name.as_ref() != "__name__"
+                            && !is_internal_column(tag.name.as_ref())
+                            && !is_metric_engine_internal_column(tag.name.as_ref())
+                            && !ReservedColumnId::is_reserved(tag.column_id)
+                    });
+                    let unique_id = valid && ids.insert(tag.column_id);
+                    let unique_name = unique_id && names.insert(tag.name.as_ref());
+                    if !valid || !unique_id || !unique_name {
+                        debug!(
+                            "Raw series scan preflight rejected label binding: region_id={}, column_id={}, binding_valid={}, unique_id={}, unique_name={}",
+                            metadata.region_id, tag.column_id, valid, unique_id, unique_name,
+                        );
+                        return Err(BoxedError::new(
+                            InvalidRecordBatchSnafu {
+                                reason: "invalid raw label binding".to_string(),
+                            }
+                            .build(),
+                        ));
+                    }
+                }
+            }
+            RawLabelSource::Explicit => {
+                debug!(
+                    "Raw series scan preflight rejected explicit labels: region_id={}",
+                    metadata.region_id,
+                );
+                return Ok(None);
+            }
+        }
+        let selected = input.mapper.output_schema();
+        let mut fields = Vec::new();
+        let mut specs = Vec::new();
+        for output in selected.column_schemas() {
+            let Some(column) = metadata.column_metadatas.iter().find(|column| {
+                column.column_schema.name == output.name
+                    && column.semantic_type == SemanticType::Field
+            }) else {
+                continue;
+            };
+            let Some((index, (_, data_type))) = input
+                .mapper
+                .batch_schema()
+                .iter()
+                .enumerate()
+                .find(|(_, (id, _))| *id == column.column_id)
+            else {
+                debug!(
+                    "Raw series scan preflight missing mapper field: region_id={}, column_id={}",
+                    metadata.region_id, column.column_id,
+                );
+                return Err(BoxedError::new(
+                    InvalidRecordBatchSnafu {
+                        reason: format!(
+                            "selected field {} is missing from mapper batch schema",
+                            output.name
+                        ),
+                    }
+                    .build(),
+                ));
+            };
+            if !data_type.is_numeric() && !is_native_histogram_value_type(data_type) {
+                debug!(
+                    "Raw series scan preflight rejected field type: region_id={}, column_id={}",
+                    metadata.region_id, column.column_id,
+                );
+                return Ok(None);
+            }
+            fields.push((index, data_type.clone()));
+            specs.push(FieldSpec {
+                name: Arc::from(output.name.as_str()),
+            });
+        }
+        let time_is_millis = matches!(metadata.time_index_type(), TimestampType::Millisecond(_));
+        if fields.is_empty()
+            || input.append_mode
+            || input.series_row_selector.is_some()
+            || !time_is_millis
+        {
+            debug!(
+                "Raw series scan preflight rejected scan options: region_id={}, selected_fields={}, append_mode={}, row_selector={}, time_is_millis={}",
+                metadata.region_id,
+                fields.len(),
+                input.append_mode,
+                input.series_row_selector.is_some(),
+                time_is_millis,
+            );
+            return Ok(None);
+        }
+        let schema = Arc::new(SeriesSchema {
+            fields: specs.into(),
+            eval_grid: None,
+        });
+        let typed_schema = input.mapper.input_arrow_schema(false);
+        let mut predicates = Vec::new();
+        if let Some(predicate) = input.predicate_group().predicate_without_region() {
+            if !predicate.dyn_filters().is_empty() {
+                debug!(
+                    "Raw series scan preflight rejected dynamic predicate filters: region_id={}",
+                    metadata.region_id,
+                );
+                return Ok(None);
+            }
+            for expr in predicate.exprs() {
+                // Fresh execution properties cannot preserve Stable or query-dependent semantics.
+                if expr
+                    .exists(|node| {
+                        Ok(matches!(node, Expr::ScalarFunction(function)
+                            if function.func.signature().volatility != Volatility::Immutable)
+                            || matches!(node, Expr::ScalarVariable(..)))
+                    })
+                    .unwrap_or(true)
+                {
+                    debug!(
+                        "Raw series scan preflight rejected predicate volatility or variable: region_id={}",
+                        metadata.region_id,
+                    );
+                    return Ok(None);
+                }
+                // Evaluate residuals with original types before raw conversion to Float64.
+                let Ok(physical) = Predicate::to_physical_expr(expr, &typed_schema) else {
+                    debug!(
+                        "Raw series scan preflight failed predicate compilation: region_id={}",
+                        metadata.region_id,
+                    );
+                    return Ok(None);
+                };
+                if !matches!(
+                    physical.data_type(&typed_schema),
+                    Ok(ArrowDataType::Boolean | ArrowDataType::Null)
+                ) {
+                    debug!(
+                        "Raw series scan preflight rejected predicate type: region_id={}",
+                        metadata.region_id,
+                    );
+                    return Ok(None);
+                }
+                predicates.push(physical);
+            }
+        }
+        Ok(Some(RawScanPreparation {
+            schema,
+            fields,
+            predicates,
+        }))
+    }
+
    fn supports_two_phase(input: &ScanInput) -> bool {
        if !is_sparse_metric_metadata(input.region_metadata()) {
            return false;
@@ -578,6 +812,22 @@ impl RegionScanner for SeriesScan {
            .map_err(BoxedError::new)
    }

+    fn raw_series_ordering(&self) -> SeriesOrdering {
+        match self.mode {
+            SeriesScanMode::Legacy => SeriesOrdering::KeyThenTime,
+            SeriesScanMode::TwoPhase => SeriesOrdering::Contiguous,
+        }
+    }
+
+    fn raw_series_schema(
+        &self,
+        label_source: &RawLabelSource,
+    ) -> Result<Option<Arc<SeriesSchema>>, BoxedError> {
+        Ok(self
+            .prepare_raw_scan(label_source)?
+            .map(|preparation| preparation.schema))
+    }
+
    fn scan_series_partition(
        &self,
        ctx: &QueryScanContext,
@@ -585,171 +835,9 @@ impl RegionScanner for SeriesScan {
        partition: usize,
        label_source: &RawLabelSource,
    ) -> Result<Option<SendableRawSeriesStream>, BoxedError> {
-        let input = &self.stream_ctx.input;
-        let metadata = input.mapper.metadata();
-        match label_source {
-            RawLabelSource::PrimaryKey {
-                physical_metadata,
-                logical_tags,
-                ..
-            } => {
-                if !Arc::ptr_eq(metadata, physical_metadata) {
-                    debug!(
-                        "Raw series scan preflight rejected metadata snapshot: region_id={}, physical_region_id={}, schema_version={}, physical_schema_version={}",
-                        metadata.region_id,
-                        physical_metadata.region_id,
-                        metadata.schema_version,
-                        physical_metadata.schema_version,
-                    );
-                    return Err(BoxedError::new(
-                        InvalidRecordBatchSnafu {
-                            reason: "raw label source metadata does not match scanner snapshot"
-                                .to_string(),
-                        }
-                        .build(),
-                    ));
-                }
-                let mut ids = HashSet::new();
-                let mut names = HashSet::new();
-                for tag in logical_tags.iter() {
-                    let column = metadata.column_by_id(tag.column_id).filter(|column| {
-                        column.semantic_type == SemanticType::Tag
-                            && metadata.primary_key.contains(&tag.column_id)
-                    });
-                    let valid = column.is_some_and(|column| {
-                        column.column_schema.name == tag.name.as_ref()
-                            && tag.name.as_ref() != "__name__"
-                            && !is_internal_column(tag.name.as_ref())
-                            && !is_metric_engine_internal_column(tag.name.as_ref())
-                            && !ReservedColumnId::is_reserved(tag.column_id)
-                    });
-                    let unique_id = valid && ids.insert(tag.column_id);
-                    let unique_name = unique_id && names.insert(tag.name.as_ref());
-                    if !valid || !unique_id || !unique_name {
-                        debug!(
-                            "Raw series scan preflight rejected label binding: region_id={}, column_id={}, binding_valid={}, unique_id={}, unique_name={}",
-                            metadata.region_id, tag.column_id, valid, unique_id, unique_name,
-                        );
-                        return Err(BoxedError::new(
-                            InvalidRecordBatchSnafu {
-                                reason: "invalid raw label binding".to_string(),
-                            }
-                            .build(),
-                        ));
-                    }
-                }
-            }
-            RawLabelSource::Explicit => {
-                debug!(
-                    "Raw series scan preflight rejected explicit labels: region_id={}",
-                    metadata.region_id,
-                );
-                return Ok(None);
-            }
-        }
-        let selected = input.mapper.output_schema();
-        let mut fields = Vec::new();
-        let mut specs = Vec::new();
-        for output in selected.column_schemas() {
-            let Some(column) = metadata.column_metadatas.iter().find(|column| {
-                column.column_schema.name == output.name
-                    && column.semantic_type == SemanticType::Field
-            }) else {
-                continue;
-            };
-            let Some((index, (_, data_type))) = input
-                .mapper
-                .batch_schema()
-                .iter()
-                .enumerate()
-                .find(|(_, (id, _))| *id == column.column_id)
-            else {
-                debug!(
-                    "Raw series scan preflight missing mapper field: region_id={}, column_id={}",
-                    metadata.region_id, column.column_id,
-                );
-                return Err(BoxedError::new(
-                    InvalidRecordBatchSnafu {
-                        reason: format!(
-                            "selected field {} is missing from mapper batch schema",
-                            output.name
-                        ),
-                    }
-                    .build(),
-                ));
-            };
-            if !data_type.is_numeric() && !is_native_histogram_value_type(data_type) {
-                debug!(
-                    "Raw series scan preflight rejected field type: region_id={}, column_id={}",
-                    metadata.region_id, column.column_id,
-                );
-                return Ok(None);
-            }
-            fields.push((index, data_type.clone()));
-            specs.push(FieldSpec {
-                name: Arc::from(output.name.as_str()),
-            });
-        }
-        let time_is_millis = matches!(metadata.time_index_type(), TimestampType::Millisecond(_));
-        if fields.is_empty()
-            || input.append_mode
-            || input.series_row_selector.is_some()
-            || !time_is_millis
-        {
-            debug!(
-                "Raw series scan preflight rejected scan options: region_id={}, selected_fields={}, append_mode={}, row_selector={}, time_is_millis={}",
-                metadata.region_id,
-                fields.len(),
-                input.append_mode,
-                input.series_row_selector.is_some(),
-                time_is_millis,
-            );
+        let Some(preparation) = self.prepare_raw_scan(label_source)? else {
            return Ok(None);
-        }
-        let schema = Arc::new(SeriesSchema {
-            fields: specs.into(),
-            eval_grid: None,
-        });
-        let typed_schema = input.mapper.input_arrow_schema(false);
-        let mut predicates = Vec::new();
-        if let Some(predicate) = input.predicate_group().predicate_without_region() {
-            for expr in predicate.exprs() {
-                // Fresh execution properties cannot preserve Stable or query-dependent semantics.
-                if expr
-                    .exists(|node| {
-                        Ok(matches!(node, Expr::ScalarFunction(function)
-                            if function.func.signature().volatility != Volatility::Immutable)
-                            || matches!(node, Expr::ScalarVariable(..)))
-                    })
-                    .unwrap_or(true)
-                {
-                    debug!(
-                        "Raw series scan preflight rejected predicate volatility or variable: region_id={}",
-                        metadata.region_id,
-                    );
-                    return Ok(None);
-                }
-                // Evaluate residuals with original types before raw conversion to Float64.
-                let Ok(physical) = Predicate::to_physical_expr(expr, &typed_schema) else {
-                    debug!(
-                        "Raw series scan preflight failed predicate compilation: region_id={}",
-                        metadata.region_id,
-                    );
-                    return Ok(None);
-                };
-                if !matches!(
-                    physical.data_type(&typed_schema),
-                    Ok(ArrowDataType::Boolean | ArrowDataType::Null)
-                ) {
-                    debug!(
-                        "Raw series scan preflight rejected predicate type: region_id={}",
-                        metadata.region_id,
-                    );
-                    return Ok(None);
-                }
-                predicates.push(physical);
-            }
-        }
+        };
        let metrics = new_partition_metrics(
            &self.stream_ctx,
            ctx.explain_verbose,
@@ -760,13 +848,15 @@ impl RegionScanner for SeriesScan {
        let batches = self
            .scan_batch_in_partition(ctx, partition, metrics.clone(), metrics_set)
            .map_err(BoxedError::new)?;
+        let ordering = self.raw_series_ordering();
        Ok(Some(raw::make_stream(
            batches,
-            schema,
+            preparation.schema,
            label_source.clone(),
-            fields,
+            preparation.fields,
            metrics,
-            predicates,
+            preparation.predicates,
+            ordering,
        )))
    }

@@ -1096,7 +1186,7 @@ impl SeriesCandidateDistributor {
                return Ok(());
            }
            collector.push(batch);
-            if collector.len() >= CANDIDATE_SERIES_ASSIGNMENT_THRESHOLD {
+            if collector.len() >= candidate_series_assignment_threshold() {
                chunked = true;
                self.send_assignments(collector.finish(false), &partition_pruner);
                if !self.should_fetch_candidates() {
@@ -1120,6 +1210,9 @@ impl SeriesCandidateDistributor {
        assignments: Vec<AssignedSeriesBatch>,
        partition_pruner: &Arc<PartitionPruner>,
    ) {
+        // One round of candidate collection; a partition that receives two assignments sees two.
+        #[cfg(test)]
+        CANDIDATE_ASSIGNMENT_ROUNDS.fetch_add(1, Ordering::Relaxed);
        for (partition, assigned_series) in assignments.into_iter().enumerate() {
            if assigned_series.series().is_empty() {
                continue;
@@ -128,6 +128,7 @@ fn chunk(
 struct RawStream {
    schema: Arc<SeriesSchema>,
    label_source: RawLabelSource,
+    ordering: SeriesOrdering,
    inner: BoxStream<'static, Result<RawSeriesChunk, BoxedError>>,
 }

@@ -146,7 +147,7 @@ impl RawSeriesStream for RawStream {
        &self.label_source
    }
    fn ordering(&self) -> SeriesOrdering {
-        SeriesOrdering::Contiguous
+        self.ordering
    }
 }

@@ -157,15 +158,16 @@ pub(crate) fn make_stream(
    fields: Vec<(usize, ConcreteDataType)>,
    metrics: PartitionMetrics,
    predicates: Vec<Arc<dyn PhysicalExpr>>,
+    ordering: SeriesOrdering,
 ) -> SendableRawSeriesStream {
    let output_schema = schema.clone();
    let stream = async_stream::stream! {
        let mut input = inner;
        let mut pending: Option<RawSeriesChunk> = None;
        while let Some(item) = input.next().await {
-            let (records, is_legacy) = match item {
-                Ok(ScanBatch::Series(SeriesBatch::Flat(flat))) => (flat.batches.into_vec(), true),
-                Ok(ScanBatch::RecordBatch(batch)) => (vec![batch], false),
+            let records = match item {
+                Ok(ScanBatch::Series(SeriesBatch::Flat(flat))) => flat.batches.into_vec(),
+                Ok(ScanBatch::RecordBatch(batch)) => vec![batch],
                Err(error) => { yield Err(BoxedError::new(error)); return; }
            };
            for record in records {
@@ -238,7 +240,7 @@ pub(crate) fn make_stream(
                    };
                    metrics.inc_convert_batch_cost(begun.elapsed());
                    if let Some(old) = pending.as_ref()
-                        && is_legacy
+                        && ordering == SeriesOrdering::KeyThenTime
                        && old.primary_key.as_slice() != current.primary_key.as_slice()
                        && current.primary_key.as_slice() <= old.primary_key.as_slice()
                    {
@@ -265,6 +267,7 @@ pub(crate) fn make_stream(
    Box::pin(RawStream {
        schema,
        label_source,
+        ordering,
        inner: Box::pin(stream),
    })
 }
@@ -37,6 +37,7 @@ use datatypes::arrow::record_batch::RecordBatch;
 use datatypes::data_type::DataType as _;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::ColumnSchema;
+use datatypes::series::SeriesOrdering;
 use datatypes::series::raw::RawSeriesChunk;
 use datatypes::series::samples::{FieldSpec, SampleRef, SeriesSchema};
 use futures::{Stream, StreamExt, stream};
@@ -239,6 +240,7 @@ fn convert_with_source(
    schema: Arc<SeriesSchema>,
    source: RawLabelSource,
    field_type: ConcreteDataType,
+    ordering: SeriesOrdering,
 ) -> SendableRawSeriesStream {
    make_stream(
        inner,
@@ -247,6 +249,7 @@ fn convert_with_source(
        vec![(0, field_type)],
        metrics(),
        vec![],
+        ordering,
    )
 }

@@ -267,6 +270,17 @@ fn convert(inner: ScanBatchStream, schema: Arc<SeriesSchema>) -> SendableRawSeri
        schema,
        primary_key_source(metadata()),
        ConcreteDataType::int64_datatype(),
+        SeriesOrdering::Contiguous,
+    )
+}
+
+fn convert_ordered(inner: ScanBatchStream, schema: Arc<SeriesSchema>) -> SendableRawSeriesStream {
+    convert_with_source(
+        inner,
+        schema,
+        primary_key_source(metadata()),
+        ConcreteDataType::int64_datatype(),
+        SeriesOrdering::KeyThenTime,
    )
 }

@@ -370,6 +384,59 @@ async fn chunks_preserve_keys_values_times_and_final_flags_for_both_input_shapes
    }
 }

+#[tokio::test]
+async fn raw_stream_reports_and_enforces_declared_ordering() {
+    let stream = convert(input(vec![]), schema());
+    assert_eq!(stream.ordering(), SeriesOrdering::Contiguous);
+    drop(stream);
+    let stream = convert_ordered(input(vec![]), schema());
+    assert_eq!(stream.ordering(), SeriesOrdering::KeyThenTime);
+    drop(stream);
+
+    let mut output = convert_ordered(
+        input(vec![
+            Ok(ScanBatch::RecordBatch(dictionary_batch(
+                &[b"z"],
+                &[1],
+                &[Some(1)],
+            ))),
+            Ok(ScanBatch::RecordBatch(dictionary_batch(
+                &[b"a"],
+                &[1],
+                &[Some(2)],
+            ))),
+        ]),
+        schema(),
+    );
+    assert!(output.next().await.unwrap().is_err());
+    assert!(output.next().await.is_none());
+
+    let mut output = convert(
+        input(vec![
+            Ok(ScanBatch::RecordBatch(dictionary_batch(
+                &[b"z"],
+                &[1],
+                &[Some(1)],
+            ))),
+            Ok(ScanBatch::RecordBatch(dictionary_batch(
+                &[b"a"],
+                &[1],
+                &[Some(2)],
+            ))),
+        ]),
+        schema(),
+    );
+    assert_eq!(
+        output.next().await.unwrap().unwrap().primary_key.as_slice(),
+        b"z"
+    );
+    assert_eq!(
+        output.next().await.unwrap().unwrap().primary_key.as_slice(),
+        b"a"
+    );
+    assert!(output.next().await.is_none());
+}
+
 #[tokio::test]
 async fn typed_predicate_filters_before_numeric_samples_are_cast() {
    let threshold = 1_i64 << 53;
@@ -395,6 +462,7 @@ async fn typed_predicate_filters_before_numeric_samples_are_cast() {
        vec![(0, ConcreteDataType::int64_datatype())],
        metrics(),
        vec![predicate],
+        SeriesOrdering::Contiguous,
    );
    let chunk = output.next().await.unwrap().unwrap();
    assert!(chunk.is_last);
@@ -424,6 +492,7 @@ async fn predicate_error_is_terminal() {
        vec![(0, ConcreteDataType::int64_datatype())],
        metrics(),
        vec![predicate],
+        SeriesOrdering::Contiguous,
    );
    assert!(output.next().await.unwrap().is_err());
    assert!(output.next().await.is_none());
@@ -515,7 +584,7 @@ async fn legacy_series_requires_increasing_keys_but_record_batches_do_not() {
    let mut batches = smallvec::SmallVec::new();
    batches.push(dictionary_batch(&[b"b"], &[1], &[Some(1)]));
    batches.push(dictionary_batch(&[b"a"], &[2], &[Some(2)]));
-    let mut output = convert(
+    let mut output = convert_ordered(
        input(vec![Ok(ScanBatch::Series(SeriesBatch::Flat(
            FlatSeriesBatch { batches },
        )))]),
@@ -528,7 +597,7 @@ async fn legacy_series_requires_increasing_keys_but_record_batches_do_not() {
    batches.push(dictionary_batch(&[b"a"], &[1], &[Some(1)]));
    batches.push(dictionary_batch(&[b"b"], &[1], &[Some(2)]));
    batches.push(dictionary_batch(&[b"a"], &[2], &[Some(3)]));
-    let mut output = convert(
+    let mut output = convert_ordered(
        input(vec![Ok(ScanBatch::Series(SeriesBatch::Flat(
            FlatSeriesBatch { batches },
        )))]),
@@ -545,7 +614,7 @@ async fn legacy_series_requires_increasing_keys_but_record_batches_do_not() {
    let mut batches = smallvec::SmallVec::new();
    batches.push(dictionary_batch(&[b"a"], &[1], &[Some(1)]));
    batches.push(dictionary_batch(&[b"b"], &[1], &[Some(2)]));
-    let mut output = convert(
+    let mut output = convert_ordered(
        input(vec![Ok(ScanBatch::Series(SeriesBatch::Flat(
            FlatSeriesBatch { batches },
        )))]),
@@ -636,6 +705,7 @@ async fn native_histogram_adapter_preserves_complete_payload() {
            native_histogram_value_type().clone(),
        )),
        native_histogram_value_type().clone(),
+        SeriesOrdering::Contiguous,
    );
    assert!(matches!(
        raw.label_source(),
@@ -23,6 +23,7 @@ use std::task::{Context, Poll, Waker};
 use api::v1::SemanticType;
 use common_error::ext::{BoxedError, ErrorExt, PlainError};
 use common_error::status_code::StatusCode;
+use common_query::series_frame::{pack_raw_series, raw_series_frame_ordering, unpack_raw_series};
 use datatypes::arrow::array::Float64Array;
 use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
 use datatypes::data_type::ConcreteDataType;
@@ -431,6 +432,67 @@ async fn merges_full_encoded_keys_and_interleaved_chunks_with_global_finality()
    assert!(series[2].3.last().unwrap());
 }

+/// Two inputs that each went through the frame codec merge like plain streams: unpacking keeps
+/// the declared ordering, and the merge restores global key order and per-key finality across
+/// the unpacked frames.
+#[tokio::test]
+async fn merges_two_unpacked_frames_through_the_frame_codec() {
+    let rack_a_sensor_1 = read_key("rack-a", "sensor-1");
+    let rack_b_sensor_1 = read_key("rack-b", "sensor-1");
+    let framed = |labels: RawLabelSource, chunks: Vec<RawSeriesChunk>| {
+        let (stream, _, _) = stream(
+            labels,
+            SeriesOrdering::KeyThenTime,
+            chunks.into_iter().map(ready),
+        );
+        let schema = Arc::clone(stream.schema());
+        let binding = stream.label_source().clone();
+        let packed = pack_raw_series(stream).unwrap();
+        let ordering = raw_series_frame_ordering(packed.schema().as_ref()).unwrap();
+        assert_eq!(SeriesOrdering::KeyThenTime, ordering);
+        unpack_raw_series(packed, schema, binding, ordering).unwrap()
+    };
+    let left = framed(
+        label_source(1),
+        vec![
+            chunk(rack_a_sensor_1.clone(), &[10, 30], &[1.0, 3.0], true),
+            chunk(rack_b_sensor_1.clone(), &[5], &[5.0], true),
+        ],
+    );
+    let right = framed(
+        label_source(2),
+        vec![chunk(rack_a_sensor_1.clone(), &[20, 40], &[2.0, 4.0], true)],
+    );
+
+    let mut merged = merge_ordered_raw_series(vec![left, right]).unwrap();
+    assert_eq!(SeriesOrdering::KeyThenTime, merged.ordering());
+    let mut series = Vec::<(Buffer, Vec<i64>, Vec<bool>)>::new();
+    while let Some(item) = merged.next().await {
+        let item = item.unwrap();
+        if series
+            .last()
+            .is_none_or(|current| current.0 != item.primary_key)
+        {
+            series.push((item.primary_key.clone(), Vec::new(), Vec::new()));
+        }
+        let current = series.last_mut().unwrap();
+        current.1.extend(item.time.as_buffer().to_vec());
+        current.2.push(item.is_last);
+    }
+    assert_eq!(2, series.len());
+    assert_eq!(rack_a_sensor_1, series[0].0);
+    assert_eq!(vec![10, 20, 30, 40], series[0].1);
+    assert!(*series[0].2.last().unwrap());
+    assert!(
+        series[0].2[..series[0].2.len() - 1]
+            .iter()
+            .all(|last| !last)
+    );
+    assert_eq!(rack_b_sensor_1, series[1].0);
+    assert_eq!(vec![5], series[1].1);
+    assert!(*series[1].2.last().unwrap());
+}
+
 #[tokio::test]
 async fn rejects_equal_timestamp_from_separate_heads() {
    let key = read_key("rack-a", "sensor-1");