diff --git a/src/promql/src/extension_plan/fused_series.rs b/src/promql/src/extension_plan/fused_series.rs
index b864dfe9460..9031611a313 100644
--- a/src/promql/src/extension_plan/fused_series.rs
+++ b/src/promql/src/extension_plan/fused_series.rs
@@ -16,8 +16,8 @@
 //!
 //! [`FusedSeries`] is the logical half: it holds the PromQL text the tree was compiled from, the
 //! tree itself and the binding its input was compiled against. [`FusedSeriesExec`] is the physical
-//! half: it owns one execution of that tree per partition and converts its input batches into raw
-//! series chunks and its evaluated chunks back into batches.
+//! half: it owns one execution of that tree per partition and decodes its input frames into raw
+//! series chunks and converts its evaluated chunks back into batches.
 //!
 //! The two halves show different things in `EXPLAIN`, which is the point of having both. The
 //! logical node prints the query a reader wrote ([`FusedSeries`]'s `fmt_for_explain`), and the
@@ -26,24 +26,26 @@
 //! can therefore see whether a query was compiled the way they expected without decoding node
 //! names.
 //!
-//! Scope of this layer: the input is a stream of ordinary record batches carrying one timestamp
-//! column, the series' field columns and the series' tag columns, with each complete label set
-//! contiguous and its timestamps ascending. The output is one batch per evaluated chunk, with the
-//! statement's grid timestamps in the declared time column. Everything the tree cannot do is
-//! rejected when the execution plan is built, not when the first batch arrives: a tree the native
-//! runtime does not execute, a binding that does not match the tree, or an input or output schema
-//! the conversion cannot classify.
+//! Scope of this layer: the input is a stream of native raw series frames
+//! ([`raw_series_frame_schema`](datatypes::series::frame::raw_series_frame_schema)) — one row per
+//! chunk, carrying the series' encoded key, its finality, its timestamps and its fields, plus the
+//! complete labels when the frame is explicit — with each series' rows contiguous and its
+//! timestamps ascending, which is what the caller's [`SeriesOrdering`] names. The output is one
+//! batch per evaluated chunk, with the statement's grid timestamps in the declared time column.
+//! Everything the tree cannot do is rejected when the execution plan is built, not when the first
+//! batch arrives: a tree the native runtime does not execute, a frame whose schema the tree's input
+//! was not compiled against, or an output schema the conversion cannot classify.
 
 use std::cmp::Ordering;
 use std::collections::hash_map::Entry;
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::collections::{HashMap, VecDeque};
 use std::fmt::{self, Formatter};
 use std::hash::{Hash, Hasher};
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
 
-use common_query::series_frame::{InputMapping, record_batches_to_raw_series};
+use common_query::series_frame::unpack_raw_series;
 use datafusion::arrow::array::{
     Array, ArrayRef, Float64Array, StringArray, TimestampMillisecondArray,
 };
@@ -59,24 +61,26 @@ use datafusion::execution::memory_pool::{
 };
 use datafusion::logical_expr::{Expr, LogicalPlan, UserDefinedLogicalNodeCore};
 use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr};
+use datafusion::physical_plan::expressions::Column as ColumnExpr;
 use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet};
+use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
 use datafusion::physical_plan::{
-    DisplayAs, DisplayFormatType, EmptyRecordBatchStream, ExecutionPlan, Partitioning,
-    PlanProperties, RecordBatchStream, SendableRecordBatchStream,
+    DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, InputDistributionRequirements,
+    Partitioning, PlanProperties, RecordBatchStream, SendableRecordBatchStream,
 };
-use datatypes::series::MetricName;
+use datatypes::series::SeriesOrdering;
 use datatypes::series::raw::RawSeriesChunk;
-use datatypes::series::samples::{
-    DenseSamplesRef, FieldSpec, SampleRef, Samples, Series, SeriesSchema,
-};
+use datatypes::series::samples::{DenseSamplesRef, SampleRef, Samples, Series, SeriesSchema};
 use datatypes::series::time::EvalGrid;
 use futures::Stream;
-use store_api::region_engine::SendableRawSeriesStream;
+use store_api::region_engine::{RawLabelSource, SendableRawSeriesStream};
 
 use crate::evaluated::EvaluatedChunk;
+use crate::grid_series::{LabelSet, label_set};
 use crate::native::execute::{NativeExecutionState, NativeExecutor};
 use crate::native::metrics::{self, NodePath};
 use crate::native::plan::{NativeExpr, NativeTree};
+use crate::source::materialize_raw_labels;
 
 #[cfg(test)]
 mod tests;
@@ -86,7 +90,7 @@ const METRIC_NAME: &str = "__name__";
 
 /// How many input steps one poll of [`FusedSeriesStream`] runs before it wakes itself.
 ///
-/// A step polls the input once and hands what it settled to the converter. Settling is not
+/// A step polls the input once and hands what it settled to the executor. Settling is not
 /// guaranteed: raw samples before the first evaluation point settle nothing, so an input of many
 /// such chunks would otherwise be read to its end inside one poll, holding the task's thread for
 /// an unbounded time and delaying both other tasks on it and the cancellation the consumer asked
@@ -98,14 +102,15 @@ const POLL_BUDGET: usize = 64;
 ///
 /// The binding travels with the tree: a tree is legal only against the schema it was bound to, and
 /// a reader of the output needs to know which grid a chunk's absolute positions are on. Keeping it
-/// in one value is what lets both halves of the node be built from the same description.
+/// in one value is what lets the logical node be built from the same description, and the physical
+/// node reads the same facts back off the tree rather than trusting a second copy of them.
 ///
-/// What the node checks is that the input's columns are exactly the ones the binding accounts for:
-/// the time column, the fields and the tags. What it cannot check is whether the caller meant a
-/// *different* column of the same type. A binding whose [`metric_name`](Self::metric_name) is not
-/// the metric the scan selects, or whose [`time_index`](Self::time_index) names another millisecond
-/// timestamp column of the input, is indistinguishable from a correct one from here: the node sees
-/// one timestamp column and one constant name, so picking them is the caller's obligation.
+/// What the logical node checks is that the schema the tree's input node declares is the one this
+/// binding names. What it cannot check is whether the caller meant *some other* schema with the
+/// same field names, or whether [`time_index`](Self::time_index) is the column the query reports
+/// positions in: from here both are indistinguishable from a correct binding, so picking them is
+/// the caller's obligation. The physical node reads neither the metric name nor a time column from
+/// the input — a raw frame carries the labels of each series and its own timestamps.
 #[derive(Clone, Debug)]
 pub struct FusedSeriesBinding {
     /// The raw series schema the tree's input node was bound to, without an evaluation grid.
@@ -114,14 +119,15 @@ pub struct FusedSeriesBinding {
     pub grid: Arc<EvalGrid>,
     /// The metric name every input series is bound to, when the input reads one named metric.
     ///
-    /// The name is a binding rather than a column because a raw series chunk carries one metric
-    /// name for all of its samples; a name that varied inside a series would not be one series.
+    /// The logical node requires it when [`tags`](Self::tags) declares [`METRIC_NAME`]: a reader of
+    /// the output has to be able to tell that a range function dropped the name, which needs a name
+    /// to drop. The physical node takes the names from the frames, which carry each series' own.
     pub metric_name: Option<Arc<str>>,
-    /// The input column holding the timestamps.
+    /// The output column holding the grid timestamps.
     pub time_index: String,
     /// The logical tag names the output carries, read from each series' labels.
     ///
-    /// [`METRIC_NAME`] is special: it is resolved to the bound metric name instead of a column,
+    /// [`METRIC_NAME`] is special: it is resolved to the series' metric name instead of a column,
     /// which is how a reader sees that a range function dropped it.
     pub tags: Vec<String>,
 }
@@ -254,19 +260,28 @@ impl FusedSeries {
 
     /// The physical node that runs `exec_input` as this node's input.
     ///
-    /// Building it rejects a tree the runtime cannot execute and a schema the conversion cannot
+    /// `label_source` and `ordering` are the binding of the frames `exec_input` produces: the
+    /// frame carries the encoded key and, in explicit mode, the complete labels, but no schema can
+    /// say which of the two a producer meant or how it ordered the series, so the caller that built
+    /// the frame passes both on. Building the node rejects a tree the runtime cannot execute, a
+    /// frame schema the tree's input was not compiled against, and a schema the conversion cannot
     /// classify, so a plan that is built is a plan that runs.
     pub fn to_execution_plan(
         &self,
         exec_input: Arc<dyn ExecutionPlan>,
+        label_source: RawLabelSource,
+        ordering: SeriesOrdering,
     ) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
         let output_schema: SchemaRef = self.output_schema.inner().clone();
         Ok(Arc::new(FusedSeriesExec::try_new(
             Arc::clone(&self.promql),
-            Arc::new(self.tree.clone()),
-            Arc::new(self.binding.clone()),
             exec_input,
+            label_source,
+            ordering,
+            Arc::new(self.tree.clone()),
             output_schema,
+            self.binding.time_index.clone(),
+            self.binding.tags.clone(),
         )?))
     }
 }
@@ -355,6 +370,64 @@ impl PartialOrd for FusedSeries {
     }
 }
 
+/// What the tree's input node reads, read off the tree itself.
+///
+/// The tree is the one authority on the schema its input was compiled against: a caller's binding
+/// can only agree or disagree with it, and the frame this node reads has to be decoded against the
+/// same schema, so both halves read it here instead of each carrying a copy.
+struct TreeInput {
+    /// The position the tree's raw chunks enter at: a position is how the runtime addresses a node,
+    /// so nothing on the node has to carry an identity that the tree itself does not.
+    path: NodePath,
+    /// The raw series schema the tree's input node was compiled against.
+    schema: Arc<SeriesSchema>,
+}
+
+/// Reads what the tree's input node was compiled against, rejecting a tree this node cannot feed.
+///
+/// A node feeds one raw input, and its output is addressed on a grid; a tree that does not have
+/// that shape is one this layer cannot hand chunks to or read positions from.
+fn tree_input(tree: &NativeTree) -> DataFusionResult<TreeInput> {
+    let mut sources = 0;
+    let mut input_path = None;
+    let mut schema = None;
+    metrics::walk(tree.root(), &mut |node, path| {
+        let NativeExpr::Input(input) = &node.expr else {
+            return;
+        };
+        sources += 1;
+        input_path.get_or_insert_with(|| path.clone());
+        // Raw, not evaluated: this node feeds the tree from a scan, so a tree whose input was
+        // compiled against an already-evaluated schema would read the wrong thing.
+        schema = input
+            .output
+            .samples()
+            .filter(|schema| schema.eval_grid.is_none())
+            .map(Arc::clone);
+    });
+    if sources != 1 {
+        return Err(DataFusionError::Plan(format!(
+            "FusedSeries runs a tree with {sources} inputs, and it needs exactly one"
+        )));
+    }
+    let (Some(path), Some(schema)) = (input_path, schema) else {
+        return Err(DataFusionError::Plan(
+            "FusedSeries runs a tree whose input is not raw series".to_string(),
+        ));
+    };
+    let evaluated = tree
+        .root()
+        .output
+        .samples()
+        .is_some_and(|schema| schema.eval_grid.is_some());
+    if !evaluated {
+        return Err(DataFusionError::Plan(
+            "FusedSeries runs a tree whose output is not evaluated on a grid".to_string(),
+        ));
+    }
+    Ok(TreeInput { path, schema })
+}
+
 /// Checks that `tree` reads the schema `binding` declares, on the grid it declares.
 ///
 /// The output grid is compared by [`EvalGrid::spec`], not by allocation: that is the rule
@@ -362,10 +435,12 @@ impl PartialOrd for FusedSeries {
 /// from its absolute grid positions, so a separately allocated grid with the same spec is the same
 /// grid to everything downstream.
 ///
-/// Returns the position of the tree's input node, which is what a caller names when it hands a raw
-/// chunk over: a position is how the runtime addresses a node, so nothing on the node has to carry
-/// an identity that the tree itself does not.
-fn validate_binding(tree: &NativeTree, binding: &FusedSeriesBinding) -> DataFusionResult<NodePath> {
+/// Answers with what the tree's input reads, which is what a caller names when it hands a raw chunk
+/// over.
+fn validate_binding(
+    tree: &NativeTree,
+    binding: &FusedSeriesBinding,
+) -> DataFusionResult<TreeInput> {
     if binding.series_schema.eval_grid.is_some() {
         return Err(DataFusionError::Plan(
             "FusedSeries binds a raw series schema, which cannot have an evaluation grid"
@@ -381,36 +456,13 @@ fn validate_binding(tree: &NativeTree, binding: &FusedSeriesBinding) -> DataFusi
     // The tree's own input node says which schema it reads, so the binding is checked against the
     // tree rather than trusted: a mismatch would evaluate one schema and label the result as
     // another.
-    let mut sources = 0;
-    let mut bound = false;
-    let mut input_path = None;
-    metrics::walk(tree.root(), &mut |node, path| {
-        let NativeExpr::Input(input) = &node.expr else {
-            return;
-        };
-        sources += 1;
-        input_path.get_or_insert_with(|| path.clone());
-        if let Some(schema) = input.output.samples() {
-            // Raw, not evaluated: this node feeds the tree from a scan, so a tree whose input was
-            // compiled against an already-evaluated schema would read the wrong thing.
-            bound = binding.same_fields(schema) && schema.eval_grid.is_none();
-        }
-    });
-    if sources != 1 {
-        return Err(DataFusionError::Plan(format!(
-            "FusedSeries runs a tree with {sources} inputs, and it needs exactly one"
-        )));
-    }
-    if !bound {
+    let input = tree_input(tree)?;
+    if !binding.same_fields(&input.schema) {
         return Err(DataFusionError::Plan(
             "FusedSeries binds a raw series schema the tree's input was not compiled against"
                 .to_string(),
         ));
     }
-    let input_path = input_path.ok_or_else(|| {
-        DataFusionError::Plan("FusedSeries runs a tree with no input node".to_string())
-    })?;
-
     match tree.root().output.samples() {
         Some(schema)
             if schema
@@ -418,7 +470,7 @@ fn validate_binding(tree: &NativeTree, binding: &FusedSeriesBinding) -> DataFusi
                 .as_ref()
                 .is_some_and(|grid| grid.spec() == binding.grid.spec()) =>
         {
-            Ok(input_path)
+            Ok(input)
         }
         _ => Err(DataFusionError::Plan(
             "FusedSeries runs a tree whose output is not evaluated on the statement's grid"
@@ -573,161 +625,29 @@ fn classify_output(
     Ok(classification)
 }
 
-/// Maps the input columns this node reads onto the raw-series roles it needs.
-///
-/// The input has to be exactly the columns the binding accounts for: the time column, the fields
-/// and the tags. A column outside them is part of some series' identity that the raw chunks would
-/// not carry, so two rows a Prometheus query keeps apart could be answered as one series with the
-/// wrong labels; such an input is refused rather than read past. Whether a *different* column of
-/// the same type was meant is not something this function can see — that is the caller commitment
-/// [`FusedSeriesBinding`] documents.
-fn input_mapping(
-    binding: &FusedSeriesBinding,
-    schema: &SchemaRef,
-) -> DataFusionResult<InputMapping> {
-    // A name that appears more than once says less than it looks: every one of those columns
-    // passes the coverage check below, because the check reads names, while the mapping reads
-    // `index_of`, which answers with one index. The other column would then be part of a series'
-    // identity that the raw chunks never carry -- rows a Prometheus query keeps apart answered as
-    // one series with the wrong labels, and no error. Which of the two the caller meant is not
-    // visible from here, so the schema is refused rather than one of them chosen.
-    let mut named = HashSet::with_capacity(schema.fields().len());
-    for field in schema.fields() {
-        if !named.insert(field.name()) {
-            return Err(DataFusionError::Plan(format!(
-                "FusedSeriesExec input has more than one column named {:?}: it cannot say which \
-                 one its binding maps, and the other would take no part in the series it holds",
-                field.name()
-            )));
-        }
-    }
-    if schema.column_with_name(METRIC_NAME).is_some() {
-        return Err(DataFusionError::Plan(format!(
-            "FusedSeriesExec cannot read a {METRIC_NAME:?} column: a raw series' metric name is \
-             bound on the node, not varied inside one series"
-        )));
-    }
-    let index_of = |name: &str| -> DataFusionResult<usize> {
-        schema.index_of(name).map_err(|_| {
-            DataFusionError::Plan(format!("FusedSeriesExec input has no column {name:?}"))
-        })
-    };
-
-    // Every column of the input has to be a role the binding reads. The binding cannot express
-    // "ignore this column": a column it does not read is one the raw series loses.
-    for field in schema.fields() {
-        let name = field.name();
-        let read = name == binding.time_index.as_str()
-            || binding.tags.iter().any(|tag| tag == name)
-            || binding
-                .series_schema
-                .fields
-                .iter()
-                .any(|spec| spec.name.as_ref() == name);
-        if !read {
-            return Err(DataFusionError::Plan(format!(
-                "FusedSeriesExec input column {name:?} is neither the time column {:?} nor one of \
-                 the fields and tags its binding maps, so it would take no part in the series' \
-                 identity",
-                binding.time_index
-            )));
-        }
-    }
-
-    let tags = binding
-        .tags
-        .iter()
-        .filter(|tag| tag.as_str() != METRIC_NAME)
-        .map(|tag| Ok((Arc::from(tag.as_str()), index_of(tag)?)))
-        .collect::<DataFusionResult<Vec<(Arc<str>, usize)>>>()?;
-    let fields = binding
-        .series_schema
-        .fields
-        .iter()
-        .map(|spec: &FieldSpec| {
-            let index = index_of(&spec.name)?;
-            // A native-histogram column would reach the output converter as a histogram sample,
-            // which only fails there, once execution has started. This node answers in floats, so
-            // the column it reads has to be one.
-            if schema.field(index).data_type() != &DataType::Float64 {
-                return Err(DataFusionError::Plan(format!(
-                    "FusedSeriesExec reads field {:?} as Float64, and the input column is {}",
-                    spec.name,
-                    schema.field(index).data_type()
-                )));
-            }
-            Ok((spec.clone(), index))
-        })
-        .collect::<DataFusionResult<Vec<(FieldSpec, usize)>>>()?;
-    let mapping = InputMapping {
-        time_index: index_of(&binding.time_index)?,
-        metric_name: binding.metric_name.clone(),
-        tags,
-        fields,
-    };
-    // The shared converter is the authority on which column types it accepts, so it is asked
-    // rather than copied: handing it an empty stream of this schema makes it validate the mapping
-    // with no batch to convert.
-    validate_mapping_types(&mapping, schema)?;
-    Ok(mapping)
-}
-
-/// Rejects a mapping the shared converter would refuse, naming the column at fault.
-///
-/// The types accepted here are the converter's own, not a copy of its rules: [`converter_accepts`]
-/// runs the converter's validation, so a time or tag type this node accepts is a type `execute`
-/// converts, and one it refuses can only be refused there too.
-fn validate_mapping_types(mapping: &InputMapping, schema: &SchemaRef) -> DataFusionResult<()> {
-    let Err(error) = converter_accepts(mapping, schema) else {
-        return Ok(());
-    };
-    // The converter reports the mapping as a whole, so the column at fault is found by asking it
-    // again with the time column alone and then with one tag at a time. The field columns are
-    // already checked above, and they are the only other columns this mapping maps.
-    let mut without_tags = mapping.clone();
-    let tags = std::mem::take(&mut without_tags.tags);
-    if converter_accepts(&without_tags, schema).is_err() {
-        return Err(DataFusionError::Plan(format!(
-            "FusedSeriesExec cannot read {:?} as its time column: {error}",
-            schema.field(mapping.time_index).name()
-        )));
-    }
-    for tag in tags {
-        let mut one = without_tags.clone();
-        one.tags = vec![tag.clone()];
-        if converter_accepts(&one, schema).is_err() {
-            return Err(DataFusionError::Plan(format!(
-                "FusedSeriesExec cannot read tag {:?} from its input column: {error}",
-                tag.0
-            )));
-        }
-    }
-    // Something the converter refuses that no single role accounts for: its own error says it.
-    Err(error)
-}
-
-/// Whether the shared converter validates `mapping` against `schema`.
-fn converter_accepts(mapping: &InputMapping, schema: &SchemaRef) -> DataFusionResult<()> {
-    record_batches_to_raw_series(
-        Box::pin(EmptyRecordBatchStream::new(Arc::clone(schema))),
-        mapping.clone(),
-    )
-    .map(|_| ())
-}
-
 /// The physical node that runs a compiled native tree over its input's raw series.
 pub struct FusedSeriesExec {
     /// The PromQL text the tree was compiled from, which is what the plan line shows.
     promql: Arc<str>,
     /// The tree this node runs, shared by every partition's execution.
     tree: Arc<NativeTree>,
-    /// What the tree's input was compiled against, and what the output columns are called.
-    binding: Arc<FusedSeriesBinding>,
+    /// The raw series schema the tree's input was compiled against, which every frame is decoded
+    /// against.
+    series_schema: Arc<SeriesSchema>,
+    /// How the input frames name their series: labels encoded in the key, or carried along.
+    label_source: RawLabelSource,
+    /// The ordering the input frames guarantee.
+    ordering: SeriesOrdering,
     /// The position of the tree's input node, where every raw chunk a partition reads enters.
     input_path: NodePath,
     /// Where each output column comes from.
     classification: Arc<[OutputColumn]>,
-    /// The plan producing the raw series.
+    /// The output column holding the grid timestamps, kept so this node can be rebuilt over a new
+    /// child without losing which column its time came from.
+    time_column: String,
+    /// The tag column names, indexed by [`OutputColumn::Tag`].
+    tags: Arc<[String]>,
+    /// The plan producing the raw series frames.
     input: Arc<dyn ExecutionPlan>,
     /// The columns this node produces.
     output_schema: SchemaRef,
@@ -744,14 +664,27 @@ pub struct FusedSeriesExec {
 
 impl FusedSeriesExec {
     /// Creates the node, rejecting whatever this node cannot execute or convert.
+    ///
+    /// `input` produces native raw series frames: the series' encoded key, its finality, its
+    /// timestamps and its fields, plus the complete labels when `label_source` is
+    /// [`RawLabelSource::Explicit`]. `time_column` and `tags` name the columns of the output, which
+    /// is a row table rather than a frame.
+    ///
+    /// The parameters stay separate because each is a fact its own caller has: the query, the plan
+    /// below, the frame's label binding and ordering, the tree, and the three things the output
+    /// schema is classified by. A struct over them would be the same list under another name.
+    #[allow(clippy::too_many_arguments)]
     pub fn try_new(
         promql: Arc<str>,
-        tree: Arc<NativeTree>,
-        binding: Arc<FusedSeriesBinding>,
         input: Arc<dyn ExecutionPlan>,
+        label_source: RawLabelSource,
+        ordering: SeriesOrdering,
+        tree: Arc<NativeTree>,
         output_schema: SchemaRef,
+        time_column: String,
+        tags: Vec<String>,
     ) -> DataFusionResult<Self> {
-        let input_path = validate_binding(&tree, &binding)?;
+        let tree_input = tree_input(&tree)?;
         // Building a state is what proves the runtime executes this tree: a node kind it does not
         // run yet is refused here, so a plan that is built is a plan that starts. The throwaway
         // state is dropped at once, returning its (unbounded) reservations.
@@ -761,24 +694,32 @@ impl FusedSeriesExec {
             &ExecutionPlanMetricsSet::new(),
             0,
         )?;
-        let classification = classify_output(
-            &output_schema,
-            &binding.series_schema,
-            &binding.tags,
-            &binding.time_index,
+        let classification =
+            classify_output(&output_schema, &tree_input.schema, &tags, &time_column)?;
+        // The frame is checked against what the tree's input reads, so a frame this node cannot
+        // decode is reported when the plan is built. The shared decoder is the authority on that
+        // layout, so it is asked rather than copied: handing it an empty stream of this schema makes
+        // it validate the declared frame with no batch to decode, which is the check
+        // `unpack_raw_series` makes before its first poll too.
+        unpack_raw_series(
+            Box::pin(RecordBatchStreamAdapter::new(
+                input.schema(),
+                futures::stream::empty(),
+            )),
+            Arc::clone(&tree_input.schema),
+            label_source.clone(),
+            ordering,
         )?;
-        // The mapping is checked against the declared input schema, so a column this node reads
-        // but the input does not produce is reported when the plan is built.
-        input_mapping(&binding, &input.schema())?;
         // One evaluator runs per input partition, so the node's partition count is the child's.
-        // The child must therefore be a raw-series scan that assigns every series whole to one
-        // partition: `PerSeries` distribution is what makes a per-partition evaluation equivalent
-        // to evaluating the whole series, because no series is split across partitions and no
-        // series appears in two of them.
+        // The child must therefore produce frames that assign every series whole to one partition,
+        // which is what this node asks for when the optimizer enforces its distribution
+        // requirements (see `input_distribution_requirements`): a per-partition evaluation is
+        // equivalent to evaluating the whole series only because no series is split across
+        // partitions and no series appears in two of them.
         //
         // One window holds the state of the series it is reading, and a raw series arrives as one
-        // contiguous run. The scan guarantees that per partition, not across them: a scan that put
-        // half of a series in one partition would leave each tree answering half of it.
+        // contiguous run. The frames guarantee that per partition, not across them: a producer that
+        // put half of a series in one partition would leave each tree answering half of it.
         let input_properties = input.properties();
         let partition_count = input_properties.partitioning.partition_count();
         if partition_count == 0 {
@@ -802,9 +743,13 @@ impl FusedSeriesExec {
         Ok(Self {
             promql,
             tree,
-            binding,
-            input_path,
+            series_schema: tree_input.schema,
+            label_source,
+            ordering,
+            input_path: tree_input.path,
             classification: classification.into(),
+            time_column,
+            tags: tags.into(),
             input,
             output_schema,
             labels,
@@ -842,6 +787,22 @@ impl ExecutionPlan for FusedSeriesExec {
         &self.properties
     }
 
+    /// One `read_key` is one series, so the requirement is that every row of one key lands in one
+    /// partition.
+    ///
+    /// The node holds one window per series at a time and closes it on the chunk carrying
+    /// `is_last` ([`FusedSeriesStream`]), so a series split across two partitions would be answered
+    /// as two halves by two trees, and a series appearing in two of them would have its state
+    /// restarted. [`Distribution::KeyPartitioned`] is exactly the requirement that says "same key,
+    /// same partition"; [`Distribution::SinglePartition`] would say it too, by collapsing the node
+    /// to one partition and giving up the parallelism the input has. A one-partition input already
+    /// satisfies a key requirement, so nothing is inserted for it.
+    fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+        InputDistributionRequirements::new(vec![Distribution::KeyPartitioned(vec![Arc::new(
+            ColumnExpr::new("read_key", 0),
+        ) as _])])
+    }
+
     fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
         vec![&self.input]
     }
@@ -864,14 +825,17 @@ impl ExecutionPlan for FusedSeriesExec {
                 children.len()
             ))
         })?;
-        // Rebuilt rather than patched: the new child's schema is what the input mapping is checked
-        // against, so a child that changed shape is refused instead of read with stale indices.
+        // Rebuilt rather than patched: the new child's schema is what the frame is validated
+        // against, so a child that changed shape is refused instead of decoded with a stale layout.
         Ok(Arc::new(Self::try_new(
             Arc::clone(&self.promql),
-            Arc::clone(&self.tree),
-            Arc::clone(&self.binding),
             child,
+            self.label_source.clone(),
+            self.ordering,
+            Arc::clone(&self.tree),
             self.output_schema.clone(),
+            self.time_column.clone(),
+            self.tags.to_vec(),
         )?))
     }
 
@@ -889,10 +853,19 @@ impl ExecutionPlan for FusedSeriesExec {
             .unwrap_or_else(|e| e.into_inner())
             .attach(&context.runtime_env().memory_pool);
         let input = self.input.execute(partition, Arc::clone(&context))?;
-        let mapping = input_mapping(&self.binding, &self.input.schema())?;
-        // Batches become raw series chunks through the shared frame converter, which is what groups
-        // rows into one chunk per label set and closes each series exactly once.
-        let input = record_batches_to_raw_series(input, mapping)?;
+        // Batches become raw series chunks through the shared frame decoder, which is the inverse of
+        // what the producer packed: one chunk per frame row, each series closed exactly once by the
+        // row carrying `is_last`.
+        let input = unpack_raw_series(
+            input,
+            Arc::clone(&self.series_schema),
+            self.label_source.clone(),
+            self.ordering,
+        )?;
+        // A source bound in key mode carries its identity encoded and no decoded labels, and a chunk
+        // handed to the tree has to name the series it holds.
+        let input =
+            materialize_raw_labels(input).map_err(|e| DataFusionError::External(Box::new(e)))?;
         let state = NativeExecutionState::try_new(
             &self.tree,
             Arc::clone(&context.runtime_env().memory_pool),
@@ -906,7 +879,7 @@ impl ExecutionPlan for FusedSeriesExec {
             executor: Some(NativeExecutor::new(state)),
             input_path: self.input_path.clone(),
             classification: Arc::clone(&self.classification),
-            binding: Arc::clone(&self.binding),
+            tags: Arc::clone(&self.tags),
             output_schema: self.output_schema.clone(),
             labels: Arc::clone(&self.labels),
             ready: VecDeque::new(),
@@ -1005,69 +978,44 @@ fn field_column(samples: &Samples) -> DataFusionResult<ArrayRef> {
     }
 }
 
-/// A canonical, collision-free encoding of a final label set.
+/// The bytes one label set holds, for the memory reservation.
 ///
-/// [`MetricName`] implements `PartialEq + Eq` but not `Hash`, and it is a dependency-owned type,
-/// so the duplicate-label-set state keys on these bytes instead of on the label set itself. Every
-/// component carries a length prefix: label names and values are arbitrary UTF-8, so a separator
-/// byte could occur inside a component and let two different label sets encode to the same bytes.
-/// [`MetricName`] keeps its tags sorted by name, so one label set has exactly one encoding.
-#[derive(Clone, Debug, PartialEq, Eq, Hash)]
-struct LabelsKey(Vec<u8>);
-
-impl LabelsKey {
-    /// Encodes the metric name, marked for presence, and then every label, name before value.
-    fn new(metric: &MetricName) -> Self {
-        let mut bytes = Vec::new();
-        match metric.name() {
-            // The marker keeps an absent metric name distinct from an empty one.
-            Some(name) => {
-                bytes.push(1);
-                push_component(&mut bytes, name);
-            }
-            None => bytes.push(0),
-        }
-        for tag in metric.tags() {
-            push_component(&mut bytes, &tag.name);
-            push_component(&mut bytes, &tag.value);
-        }
-        Self(bytes)
-    }
-
-    /// Number of bytes this key occupies, for the memory reservation.
-    fn len(&self) -> usize {
-        self.0.len()
-    }
+/// It counts each label's name and value with the pair that refers to them, which is the same
+/// convention [`crate::native::execute`]'s own label-set table keeps.
+fn labels_bytes(labels: &LabelSet) -> usize {
+    labels
+        .iter()
+        .map(|(name, value)| size_of::<(Arc<str>, Arc<str>)>() + name.len() + value.len())
+        .sum()
 }
 
-/// Appends one length-prefixed UTF-8 component.
-fn push_component(bytes: &mut Vec<u8>, component: &str) {
-    bytes.extend_from_slice(&component.len().to_le_bytes());
-    bytes.extend_from_slice(component.as_bytes());
-}
-
-/// Renders a label set the way Prometheus prints one, for the duplicate-label-set error.
-fn render_labels(metric: &MetricName) -> String {
-    let mut parts = Vec::new();
-    if let Some(name) = metric.name() {
-        parts.push(format!("__name__={name:?}"));
-    }
-    parts.extend(
-        metric
-            .tags()
-            .iter()
-            .map(|tag| format!("{}={:?}", tag.name, tag.value)),
-    );
+/// Renders the label set Prometheus compares the way it prints one, for the duplicate-label-set
+/// error.
+fn render_labels(labels: &LabelSet) -> String {
+    let parts = labels
+        .iter()
+        .map(|(name, value)| format!("{name}={value:?}"))
+        .collect::<Vec<_>>();
     format!("{{{}}}", parts.join(", "))
 }
 
-/// The grid positions each final label set already occupies, as merged, sorted runs.
+/// The grid positions each final label set already occupies, per field and as merged, sorted runs.
 ///
 /// Shared by every partition's stream, because a label-set collision is a property of the whole
-/// output vector and not of one partition. `PerSeries` distribution keeps each series whole inside
-/// one partition, but two series that carry the same labels after the range function and its maps
-/// drop a label can still land in different partitions. Checking only within a partition would
-/// miss exactly those collisions.
+/// output vector and not of one partition. `KeyPartitioned` distribution keeps each series whole
+/// inside one partition, but two series that carry the same labels after the range function and
+/// its maps drop a label can still land in different partitions. Checking only within a partition
+/// would miss exactly those collisions.
+///
+/// The key is [`crate::grid_series::label_set`]'s normalization of the final labels rather than
+/// [`datatypes::series::MetricName`] itself: a label with an empty value is no label to
+/// Prometheus, so `{host="h", zone=""}` and `{host="h"}` are one label set and have to be compared
+/// as one. Encoding the tags as they are would file them apart and let one point hold two values.
+///
+/// Occupancy is per field and per position, not per chunk: two series whose label sets became equal
+/// while holding values in *different* fields merge cleanly in Prometheus, so only the same field
+/// at the same position collides. That is the rule
+/// [`crate::grid_series::merge_same_labelset`] applies.
 ///
 /// The map belongs to the plan, not to one stream: a partition that ends leaves what it claimed
 /// for the partitions still running, so the entries are only released with the plan itself. The
@@ -1078,8 +1026,9 @@ fn render_labels(metric: &MetricName) -> String {
 /// was unaccounted for.
 #[derive(Default)]
 struct LabelSets {
-    /// The grid positions each final label set already occupies, as merged, sorted runs.
-    seen: HashMap<LabelsKey, Vec<(usize, usize)>>,
+    /// The grid positions each final label set already occupies, one run list per field of the
+    /// series schema.
+    seen: HashMap<LabelSet, Vec<Vec<(usize, usize)>>>,
     /// What `seen` keeps charged on the task's memory pool. `None` until the first execution,
     /// since construction has no `TaskContext` and therefore no pool to register against.
     reservation: Option<MemoryReservation>,
@@ -1099,13 +1048,14 @@ impl LabelSets {
 
     /// Claims the occupied grid positions of one chunk of a final series for its label set.
     ///
-    /// A collision is an error only when both series hold a sample at the same evaluation
-    /// timestamp, which is Prometheus' range-query rule. Series with disjoint points are legal:
-    /// Prometheus merges them into one output series, while this node emits them as separate
-    /// batches and defers that merge to matrix assembly, so every sample survives.
+    /// A collision is an error only when both series hold a sample of *one field* at the same
+    /// evaluation timestamp, which is Prometheus' range-query rule. Series with disjoint points are
+    /// legal, and so are series that fill the same points field by field: Prometheus merges both
+    /// into one output series, while this node emits them as separate batches and defers that merge
+    /// to matrix assembly, so every sample survives.
     fn claim(&mut self, series: &Series) -> DataFusionResult<()> {
         let runs = occupied_runs(series)?;
-        let key = LabelsKey::new(&series.metric);
+        let key = label_set(&series.metric);
         let Self { seen, reservation } = self;
         // Nothing reaches a claim without an `execute` first, and an execution that ran attached the
         // reservation. Claiming without one would leave the map growing uncharged, so it is refused
@@ -1118,18 +1068,45 @@ impl LabelSets {
         match seen.entry(key) {
             Entry::Occupied(mut entry) => {
                 let stored = entry.get_mut();
-                if runs_overlap(stored, &runs) {
-                    return Err(DataFusionError::Execution(format!(
-                        "vector cannot contain metrics with the same labelset: {}",
-                        render_labels(&series.metric)
-                    )));
+                // An entry recorded from a chunk of fewer fields is padded, so every field of this
+                // chunk has a slot of its own: a slot is a field index of the schema, so both sides
+                // count the same field the same way. A chunk of *fewer* fields than the entry
+                // leaves the remaining slots to the chunks that fill them.
+                if stored.len() < runs.len() {
+                    stored.resize_with(runs.len(), Vec::new);
+                }
+                for (field, (stored, new)) in stored.iter().zip(&runs).enumerate() {
+                    if runs_overlap(stored, new) {
+                        return Err(DataFusionError::Execution(format!(
+                            "vector cannot contain metrics with the same labelset: {} (field \
+                             {field})",
+                            render_labels(entry.key())
+                        )));
+                    }
+                }
+                // What the merge below allocates is what this entry grows by: `merge_runs` hands
+                // back a list of `stored + new` runs, so the entry's capacity can only grow by the
+                // difference over what it already holds. Without this the arrays would grow for as
+                // long as the plan lives while the charge stayed at the first claim.
+                let grown = stored
+                    .iter()
+                    .zip(&runs)
+                    .map(|(stored, new)| {
+                        (stored.len() + new.len()).saturating_sub(stored.capacity())
+                    })
+                    .sum::<usize>()
+                    * size_of::<(usize, usize)>();
+                reservation.try_grow(grown)?;
+                for (stored, new) in stored.iter_mut().zip(&runs) {
+                    merge_runs(stored, new);
                 }
-                merge_runs(stored, &runs);
             }
             Entry::Vacant(entry) => {
                 // Only the label-set state is this node's own cross-poll state that grows with
                 // the output, so it is the only thing reserved here.
-                let bytes = entry.key().len() + runs.len() * 2 * size_of::<usize>();
+                let bytes = labels_bytes(entry.key())
+                    + runs.iter().map(|runs| runs.len()).sum::<usize>()
+                        * size_of::<(usize, usize)>();
                 reservation.try_grow(bytes)?;
                 entry.insert(runs);
             }
@@ -1140,12 +1117,14 @@ impl LabelSets {
 
 type SharedLabelSets = Arc<Mutex<LabelSets>>;
 
-/// The grid positions the rows of one chunk hold a value at, as sorted, disjoint runs.
+/// The grid positions the rows of one chunk hold a value at, one run list per field.
 ///
-/// Each run is `(start, end_exclusive)` over absolute grid indices. A row whose every field is
-/// missing is not occupied: Prometheus has no point there, so it cannot collide with another
-/// series.
-fn occupied_runs(series: &Series) -> DataFusionResult<Vec<(usize, usize)>> {
+/// Each run is `(start, end_exclusive)` over absolute grid indices, and the runs of one field are
+/// sorted and disjoint. Occupancy is kept per field because it is what a collision is: two series
+/// whose label sets became equal merge field by field in Prometheus, so only two values of the same
+/// field at one position collide. Marking the whole position as taken as soon as *any* field held a
+/// value would refuse the complementary pair that merges cleanly.
+fn occupied_runs(series: &Series) -> DataFusionResult<Vec<Vec<(usize, usize)>>> {
     // Rows are addressed on the statement's grid, so a row sits at the absolute position two
     // series can collide on. An explicit axis has no such position, and a chunk never carries
     // one: this is the refusal `EvaluatedChunk` already makes when the chunk is built.
@@ -1155,30 +1134,37 @@ fn occupied_runs(series: &Series) -> DataFusionResult<Vec<(usize, usize)>> {
         ));
     }
     let rows = series.time.len();
-    // A series with one dense, null-free float field holds a value on every row it carries, so a
-    // chunk that forms one run occupies exactly the range it reports and no row has to be
-    // inspected. This is the shape the fused evaluator produces for a range function over
-    // ordinary float samples.
-    if rows > 0
-        && series.fields.len() == 1
-        && let Some(DenseSamplesRef::Float(values)) =
-            series.fields.iter().next().and_then(Samples::dense)
-        && values.len() == rows
-        && values.null_count() == 0
-        && let Some(range) = series.time.as_grid_range()
-    {
-        return Ok(vec![(range.start(), range.start() + rows)]);
+    let range = series.time.as_grid_range();
+    let mut runs: Vec<Vec<(usize, usize)>> = vec![Vec::new(); series.fields.len()];
+    // A field that is one dense, null-free float run holds a value on every row the chunk carries,
+    // so it occupies exactly the range the chunk reports and no row of it has to be inspected. This
+    // is the shape the fused evaluator produces for a range function over ordinary float samples.
+    // Every other field is walked below, and a field filled here is skipped there.
+    let mut walked: Vec<(usize, &Samples)> = Vec::new();
+    for (index, samples) in series.fields.iter().enumerate() {
+        match (range, samples.dense()) {
+            (Some(range), Some(DenseSamplesRef::Float(values)))
+                if rows > 0 && values.len() == rows && values.null_count() == 0 =>
+            {
+                runs[index] = vec![(range.start(), range.start() + rows)];
+            }
+            _ => walked.push((index, samples)),
+        }
+    }
+    if walked.is_empty() {
+        return Ok(runs);
     }
     // `positions` walks a chunk that forms one run and one whose rows form several the same way,
     // in position order, so the holes between two runs are never read as rows.
-    let mut runs: Vec<(usize, usize)> = Vec::new();
     for (position, row) in series.time.positions() {
-        if !series.fields.iter().any(|samples| has_value(samples, row)) {
-            continue;
-        }
-        match runs.last_mut() {
-            Some(run) if run.1 == position => run.1 += 1,
-            _ => runs.push((position, position + 1)),
+        for (index, samples) in &walked {
+            if !has_value(samples, row) {
+                continue;
+            }
+            match runs[*index].last_mut() {
+                Some(run) if run.1 == position => run.1 += 1,
+                _ => runs[*index].push((position, position + 1)),
+            }
         }
     }
     Ok(runs)
@@ -1248,9 +1234,9 @@ fn merge_runs(stored: &mut Vec<(usize, usize)>, new: &[(usize, usize)]) {
 
 /// The batches one execution of [`FusedSeriesExec`] yields.
 pub struct FusedSeriesStream {
-    /// The input batches, read as raw series chunks, or `None` once the stream ended in an error.
+    /// The input frames, read as raw series chunks, or `None` once the stream ended in an error.
     ///
-    /// The converter behind it holds its current batch, the chunk it is building and its own
+    /// The decoder behind it holds its current batch, the chunks it decoded from it and its own
     /// source, so a terminated stream that kept it alive would hold upstream resources until the
     /// consumer drops the stream.
     input: Option<SendableRawSeriesStream>,
@@ -1261,7 +1247,7 @@ pub struct FusedSeriesStream {
     /// Where each output column comes from.
     classification: Arc<[OutputColumn]>,
     /// The tag column names, indexed by `OutputColumn::Tag`.
-    binding: Arc<FusedSeriesBinding>,
+    tags: Arc<[String]>,
     /// The schema of every batch this stream yields.
     output_schema: SchemaRef,
     /// The label-set occupancy shared with the other partitions' streams.
@@ -1304,7 +1290,7 @@ impl FusedSeriesStream {
     ///
     /// An ordinary drop of the stream releases all of this as well, so this is not about the last
     /// release: it is about a consumer that keeps the terminated stream, which would otherwise
-    /// hold the evaluator, the converter's held batch and chunk, and the plan upstream of it.
+    /// hold the evaluator, the decoder's held batch and chunks, and the plan upstream of it.
     fn fail(&mut self) {
         self.done = true;
         self.executor = None;
@@ -1327,7 +1313,7 @@ impl FusedSeriesStream {
                 .claim(&chunk.series)?;
             let batch = series_to_record_batch(
                 &self.classification,
-                &self.binding.tags,
+                &self.tags,
                 &self.output_schema,
                 chunk,
             )?;
diff --git a/src/promql/src/extension_plan/fused_series/tests.rs b/src/promql/src/extension_plan/fused_series/tests.rs
index f0ea868a3ee..c5e2b9e93f6 100644
--- a/src/promql/src/extension_plan/fused_series/tests.rs
+++ b/src/promql/src/extension_plan/fused_series/tests.rs
@@ -19,6 +19,12 @@
 //! `rate(x[5m])`: the compiler itself lives in the `query` crate, which depends on this one, so a
 //! test in this crate cannot call it.
 //!
+//! The input is what a scan hands over: native raw series frames, one batch per chunk here, packed
+//! by the same code the scan uses ([`pack_raw_series_chunk`]). A frame carries the series' encoded
+//! key, the chunk's finality, its timestamps and its fields, plus the complete labels, so a test
+//! says which series a chunk belongs to and whether it closes it instead of relying on where a row
+//! boundary happens to fall.
+//!
 //! Kernel numerics are not the subject: `rate` is checked against a counter that advances one
 //! second per second, so the value it has to produce is exactly 1 (or 2) at every grid point and
 //! any position or finalization mistake shows up as a wrong, missing or duplicated row.
@@ -30,9 +36,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::task::{Context, Poll};
 
-use datafusion::arrow::array::{
-    Array, ArrayRef, Float64Array, StringArray, TimestampMillisecondArray,
-};
+use datafusion::arrow::array::{Array, Float64Array, StringArray, TimestampMillisecondArray};
 use datafusion::arrow::buffer::{Buffer, ScalarBuffer};
 use datafusion::arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
 use datafusion::arrow::record_batch::RecordBatch;
@@ -47,24 +51,30 @@ use datafusion::execution::runtime_env::RuntimeEnvBuilder;
 use datafusion::logical_expr::{EmptyRelation, LogicalPlan, UserDefinedLogicalNodeCore};
 use datafusion::physical_expr::{EquivalenceProperties, PhysicalExpr};
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
+use datafusion::physical_plan::expressions::Column as ColumnExpr;
 use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricValue, MetricsSet};
 use datafusion::physical_plan::{
-    DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties, RecordBatchStream,
-    SendableRecordBatchStream, collect, displayable,
+    DisplayAs, DisplayFormatType, Distribution, ExecutionPlan, Partitioning, PlanProperties,
+    RecordBatchStream, SendableRecordBatchStream, collect, displayable,
 };
 use datafusion::prelude::SessionContext;
+use datatypes::series::frame::{RawLabelMode, pack_raw_series_chunk, raw_series_frame_schema};
 use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
 use datatypes::series::samples::{FieldSpec, Samples, Series, SeriesFields, SeriesSchema};
 use datatypes::series::time::{EvalGrid, GridSpan, GridSpec, TimeAxis};
 use datatypes::series::{Label, MetricName, SeriesOrdering};
 use futures::{Stream, StreamExt, TryStreamExt};
+use store_api::region_engine::RawLabelSource;
 
-use super::{FusedSeries, FusedSeriesBinding, OutputColumn, POLL_BUDGET, series_to_record_batch};
+use super::{
+    FusedSeries, FusedSeriesBinding, LabelSets, OutputColumn, POLL_BUDGET, label_set,
+    series_to_record_batch,
+};
 use crate::evaluated::EvaluatedChunk;
 use crate::native::execute::{NativeExecutionState, NativeExecutor};
 use crate::native::plan::{
-    AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundWindow, Grouping, NativeExpr,
-    NativeNode, NativeTree, NativeType,
+    AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundWindow, Grouping,
+    NativeExpr, NativeNode, NativeTree, NativeType,
 };
 use crate::range_eval::{RangeFunction, RangeWindow};
 
@@ -79,7 +89,7 @@ const GRID_LEN: u64 = 5;
 /// One raw sample per minute, from 0 to 9 minutes: enough to fill every window on the grid.
 const SAMPLE_STEP: i64 = 60_000;
 const SAMPLES: i64 = 10;
-/// How many one-sample batches the quiet series below arrives in.
+/// How many one-sample chunks the quiet series below arrives in.
 ///
 /// Two bounds hold at once: the series reaches the end of the last grid point's window (540s, so
 /// indices `0..=540`), which is what lets it rate at 1 over the whole grid, and it arrives in more
@@ -94,9 +104,12 @@ const QUIET_STEP: i64 = 1_000;
 
 const TIME: &str = "time";
 const VALUE: &str = "value";
+const OTHER: &str = "other";
 const HOST: &str = "host";
 const ZONE: &str = "zone";
 const NAME: &str = "__name__";
+/// A metadata label, which the window drops along with the metric name.
+const TYPE: &str = "__type__";
 const METRIC: &str = "http_requests_total";
 
 fn grid() -> Arc<EvalGrid> {
@@ -113,16 +126,27 @@ fn raw_schema() -> Arc<SeriesSchema> {
     })
 }
 
-/// The tree `rate(x[5m])` compiles to: raw samples folded by one window onto the grid.
-fn rate_tree() -> NativeTree {
+/// A raw schema of two float fields, so one series can fill one of them and another the other.
+fn two_field_schema() -> Arc<SeriesSchema> {
+    Arc::new(SeriesSchema {
+        fields: Arc::from([
+            FieldSpec { name: VALUE.into() },
+            FieldSpec { name: OTHER.into() },
+        ]),
+        eval_grid: None,
+    })
+}
+
+/// The tree `rate(x[5m])` compiles to over `schema`: raw samples folded by one window onto the grid.
+fn rate_tree_of(schema: Arc<SeriesSchema>) -> NativeTree {
     let grid = grid();
     // One schema shared by the window's binding and its output, which is what `NativeTree`'s
     // validation compares by identity.
     let evaluated = NativeType::Samples(Arc::new(SeriesSchema {
-        fields: raw_schema().fields.clone(),
+        fields: schema.fields.clone(),
         eval_grid: Some(Arc::clone(&grid)),
     }));
-    let raw = NativeType::Samples(raw_schema());
+    let raw = NativeType::Samples(schema);
     NativeTree::try_new(NativeNode {
         expr: NativeExpr::Window {
             input: Box::new(NativeNode {
@@ -149,6 +173,11 @@ fn rate_tree() -> NativeTree {
     .unwrap()
 }
 
+/// The tree `rate(x[5m])` compiles to over one field.
+fn rate_tree() -> NativeTree {
+    rate_tree_of(raw_schema())
+}
+
 /// The `rate(x[5m])` tree with a `sum by (job)` over it, which the runtime does not execute.
 fn aggregate_tree() -> NativeTree {
     let rate = rate_tree();
@@ -177,10 +206,45 @@ fn aggregate_tree() -> NativeTree {
     .expect("the bound tree is legal")
 }
 
-/// What the tree's input was compiled against, read off the tree itself.
-fn binding(tree: &NativeTree) -> FusedSeriesBinding {
+/// The tree an instant selection compiles to: the last raw sample at or before each grid point.
+///
+/// Unlike a window or a map, a selection rewrites no label: the series it answers carries the labels
+/// its input did.
+fn instant_tree() -> NativeTree {
+    let grid = grid();
+    let evaluated = NativeType::Samples(Arc::new(SeriesSchema {
+        fields: raw_schema().fields.clone(),
+        eval_grid: Some(Arc::clone(&grid)),
+    }));
+    let raw = NativeType::Samples(raw_schema());
+    NativeTree::try_new(NativeNode {
+        expr: NativeExpr::Instant {
+            input: Box::new(NativeNode {
+                expr: NativeExpr::Input(BoundInput {
+                    output: raw.clone(),
+                    ordering: SeriesOrdering::KeyThenTime,
+                }),
+                output: raw,
+            }),
+            op: BoundInstant {
+                lookback_ms: RANGE_MS,
+                offset_ms: 0,
+                output_grid: grid,
+            },
+        },
+        output: evaluated,
+    })
+    .expect("the bound tree is legal")
+}
+
+/// What the tree's input was compiled against, for the input schema and output tags given.
+fn binding_of(
+    tree: &NativeTree,
+    series_schema: Arc<SeriesSchema>,
+    tags: Vec<String>,
+) -> FusedSeriesBinding {
     FusedSeriesBinding {
-        series_schema: raw_schema(),
+        series_schema,
         grid: tree
             .root()
             .output
@@ -191,21 +255,23 @@ fn binding(tree: &NativeTree) -> FusedSeriesBinding {
             .expect("this tree's output is evaluated on a grid"),
         metric_name: Some(Arc::from(METRIC)),
         time_index: TIME.to_string(),
-        tags: vec![HOST.to_string(), NAME.to_string()],
+        tags,
     }
 }
 
-/// The batches the input plan produces: the time, the field and the tag column.
-fn input_schema() -> SchemaRef {
-    Arc::new(Schema::new(vec![
-        Field::new(
-            TIME,
-            DataType::Timestamp(TimeUnit::Millisecond, None),
-            false,
-        ),
-        Field::new(VALUE, DataType::Float64, true),
-        Field::new(HOST, DataType::Utf8, true),
-    ]))
+/// What the tree's input was compiled against, read off the tree itself.
+fn binding(tree: &NativeTree) -> FusedSeriesBinding {
+    binding_of(tree, raw_schema(), vec![HOST.to_string(), NAME.to_string()])
+}
+
+/// The schema of a frame carrying series of `schema` with their labels explicit.
+fn frame_schema_of(schema: &SeriesSchema) -> SchemaRef {
+    raw_series_frame_schema(schema, RawLabelMode::Explicit).expect("explicit labels are a frame")
+}
+
+/// The schema of a frame carrying these series, which is what the node declares as its input.
+fn frame_schema() -> SchemaRef {
+    frame_schema_of(&raw_schema())
 }
 
 /// The columns this node produces: the field, the tags and the grid timestamps.
@@ -222,10 +288,10 @@ fn output_schema() -> SchemaRef {
     ]))
 }
 
-fn input_plan() -> LogicalPlan {
+fn input_plan(schema: SchemaRef) -> LogicalPlan {
     LogicalPlan::EmptyRelation(EmptyRelation {
         produce_one_row: false,
-        schema: input_schema().to_dfschema_ref().unwrap(),
+        schema: schema.to_dfschema_ref().unwrap(),
     })
 }
 
@@ -236,17 +302,26 @@ fn memory_exec_with(schema: SchemaRef, batches: Vec<RecordBatch>) -> Arc<dyn Exe
     )))
 }
 
-/// A physical input producing `batches` in one partition, on the input schema of these tests.
+/// A physical input producing `batches` in one partition, on the frame schema of these tests.
 fn memory_exec(batches: Vec<RecordBatch>) -> Arc<dyn ExecutionPlan> {
-    memory_exec_with(input_schema(), batches)
+    memory_exec_with(frame_schema(), batches)
+}
+
+/// A physical input producing each of `partitions`' batches in a partition of its own, with
+/// `schema` as its schema.
+fn memory_exec_partitioned_with(
+    schema: SchemaRef,
+    partitions: Vec<Vec<RecordBatch>>,
+) -> Arc<dyn ExecutionPlan> {
+    Arc::new(DataSourceExec::new(Arc::new(
+        MemorySourceConfig::try_new(&partitions, schema, None)
+            .expect("every partition holds batches of the given frame schema"),
+    )))
 }
 
 /// A physical input producing each of `partitions`' batches in a partition of its own.
 fn memory_exec_partitioned(partitions: Vec<Vec<RecordBatch>>) -> Arc<dyn ExecutionPlan> {
-    Arc::new(DataSourceExec::new(Arc::new(
-        MemorySourceConfig::try_new(&partitions, input_schema(), None)
-            .expect("every partition holds batches of the input schema"),
-    )))
+    memory_exec_partitioned_with(frame_schema(), partitions)
 }
 
 /// A task context whose memory pool is `pool`, so what the node charges can be read afterwards.
@@ -434,26 +509,126 @@ fn fused(tree: NativeTree) -> FusedSeries {
         Arc::from(PROMQL),
         tree,
         binding,
-        input_plan(),
+        input_plan(frame_schema()),
         output_schema().to_dfschema_ref().unwrap(),
     )
     .expect("the binding matches the tree")
 }
 
-/// One batch of one series' `(timestamp, value)` rows.
-fn series_batch(host: &str, rows: &[(i64, f64)]) -> RecordBatch {
-    let times = rows.iter().map(|(time, _)| *time).collect::<Vec<_>>();
-    let values = rows.iter().map(|(_, value)| *value).collect::<Vec<_>>();
-    let hosts = rows.iter().map(|_| host).collect::<Vec<_>>();
-    RecordBatch::try_new(
-        input_schema(),
-        vec![
-            Arc::new(TimestampMillisecondArray::from(times)),
-            Arc::new(Float64Array::from(values)),
-            Arc::new(StringArray::from(hosts)),
-        ],
+/// The physical node of `node` over `input`, reading the complete labels every frame carries.
+fn plan_of(
+    node: FusedSeries,
+    input: Arc<dyn ExecutionPlan>,
+) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+    node.to_execution_plan(input, RawLabelSource::Explicit, SeriesOrdering::Contiguous)
+}
+
+/// The physical node of `tree` over `input`.
+fn plan(
+    tree: NativeTree,
+    input: Arc<dyn ExecutionPlan>,
+) -> DataFusionResult<Arc<dyn ExecutionPlan>> {
+    plan_of(fused(tree), input)
+}
+
+/// The timestamps of `rows`.
+fn times_of(rows: &[(i64, f64)]) -> Vec<i64> {
+    rows.iter().map(|(time, _)| *time).collect()
+}
+
+/// The values of `rows`, one sample per timestamp.
+fn values_of(rows: &[(i64, f64)]) -> Vec<Option<f64>> {
+    rows.iter().map(|(_, value)| Some(*value)).collect()
+}
+
+/// One missing sample per timestamp of `rows`, which is a field the series holds no value in.
+fn no_values(rows: &[(i64, f64)]) -> Vec<Option<f64>> {
+    rows.iter().map(|_| None).collect()
+}
+
+/// One chunk of `metric`, keyed `key`, holding `times` and one sample per field of `schema`.
+fn chunk_of(
+    schema: &SeriesSchema,
+    key: &str,
+    metric: MetricName,
+    times: &[i64],
+    values: Vec<Vec<Option<f64>>>,
+    last: bool,
+) -> RawSeriesChunk {
+    assert_eq!(
+        schema.fields.len(),
+        values.len(),
+        "one value list per field of the schema"
+    );
+    assert!(
+        values.iter().all(|values| values.len() == times.len()),
+        "one sample per timestamp per field"
+    );
+    let fields = values
+        .into_iter()
+        .map(|values| Samples::from_float(Float64Array::from(values)))
+        .collect::<Vec<_>>();
+    RawSeriesChunk::try_new(
+        schema,
+        Buffer::from_slice_ref(key.as_bytes()),
+        Some(metric),
+        RawTimestamps::try_new(ScalarBuffer::from(times.to_vec())).expect("ascending timestamps"),
+        SeriesFields::try_from_vec(fields).expect("one sample per timestamp per field"),
+        last,
     )
-    .unwrap()
+    .expect("one series' samples are a legal raw chunk")
+}
+
+/// One chunk of the series `{__name__=METRIC, host=<host>}`: `rows` as its samples, and the chunk
+/// that closes the series when `last`. The host names the series, so it doubles as the key.
+fn series_chunk(host: &str, rows: &[(i64, f64)], last: bool) -> RawSeriesChunk {
+    chunk_of(
+        &raw_schema(),
+        host,
+        metric(&[(HOST, host)], Some(METRIC)),
+        &times_of(rows),
+        vec![values_of(rows)],
+        last,
+    )
+}
+
+/// The metric these tests read, with `labels` on it and the name given.
+fn metric(labels: &[(&str, &str)], name: Option<&str>) -> MetricName {
+    MetricName::try_new(
+        name.map(Arc::from),
+        labels
+            .iter()
+            .map(|(name, value)| Label {
+                name: Arc::from(*name),
+                value: Arc::from(*value),
+            })
+            .collect(),
+    )
+    .expect("the labels of these tests are legal")
+}
+
+/// One frame batch carrying `chunk` of `schema`, packed by the scan's own encoder.
+fn frame_of(schema: &SeriesSchema, chunk: &RawSeriesChunk) -> RecordBatch {
+    let mut batches = pack_raw_series_chunk(schema, RawLabelMode::Explicit, chunk)
+        .expect("a chunk of this schema packs into a frame");
+    assert_eq!(1, batches.len(), "one small chunk is one frame batch");
+    batches.remove(0)
+}
+
+/// One frame batch carrying `chunk` of the one-field schema of these tests.
+fn frame(chunk: &RawSeriesChunk) -> RecordBatch {
+    frame_of(&raw_schema(), chunk)
+}
+
+/// The frames carrying `chunks` of `schema`, one batch each, so a test can say how much input one
+/// poll reads.
+fn frames_of(schema: &SeriesSchema, chunks: &[RawSeriesChunk]) -> Vec<RecordBatch> {
+    chunks.iter().map(|chunk| frame_of(schema, chunk)).collect()
+}
+
+/// The frames carrying `chunks` of the one-field schema of these tests, one batch each.
+fn frames(chunks: &[RawSeriesChunk]) -> Vec<RecordBatch> {
+    frames_of(&raw_schema(), chunks)
 }
 
 /// One series' samples: one per minute from 0, `scale` seconds per second.
@@ -466,27 +641,27 @@ fn series_samples(scale: f64) -> Vec<(i64, f64)> {
         .collect()
 }
 
-/// Two series, each split over two batches. A series' batches stay adjacent, which is the input
-/// contract: a label set's rows are contiguous across the whole stream.
-fn two_series_batches() -> Vec<RecordBatch> {
+/// Two series, each split over two chunks. A series' chunks stay adjacent, and the second of each
+/// pair closes it, which is the input contract: a series' chunks arrive together.
+fn two_series_frames() -> Vec<RecordBatch> {
     let first = series_samples(1.0);
     let second = series_samples(2.0);
-    vec![
-        series_batch("web-1", &first[..3]),
-        series_batch("web-1", &first[3..]),
-        series_batch("web-2", &second[..3]),
-        series_batch("web-2", &second[3..]),
-    ]
+    frames(&[
+        series_chunk("web-1", &first[..3], false),
+        series_chunk("web-1", &first[3..], true),
+        series_chunk("web-2", &second[..3], false),
+        series_chunk("web-2", &second[3..], true),
+    ])
 }
 
-/// One series split over three batches, so its chunks have to be stitched.
-fn one_series_batches() -> Vec<RecordBatch> {
+/// One series split over three chunks, so its chunks have to be stitched.
+fn one_series_frames() -> Vec<RecordBatch> {
     let rows = series_samples(1.0);
-    vec![
-        series_batch("web-1", &rows[..4]),
-        series_batch("web-1", &rows[4..7]),
-        series_batch("web-1", &rows[7..]),
-    ]
+    frames(&[
+        series_chunk("web-1", &rows[..4], false),
+        series_chunk("web-1", &rows[4..7], false),
+        series_chunk("web-1", &rows[7..], true),
+    ])
 }
 
 /// The rows of `batches`: timestamp, value, host and metric name.
@@ -531,70 +706,6 @@ fn rows(batches: &[RecordBatch]) -> Vec<Row> {
     rows
 }
 
-/// The input schema of the collision below: the time, the field, `host` and `zone`.
-fn zone_schema() -> SchemaRef {
-    Arc::new(Schema::new(vec![
-        Field::new(
-            TIME,
-            DataType::Timestamp(TimeUnit::Millisecond, None),
-            false,
-        ),
-        Field::new(VALUE, DataType::Float64, true),
-        Field::new(HOST, DataType::Utf8, true),
-        Field::new(ZONE, DataType::Utf8, true),
-    ]))
-}
-
-/// The two rows of the collision: one series, once with an empty `zone` and once with no `zone`.
-fn zone_batches() -> Vec<RecordBatch> {
-    let schema = zone_schema();
-    let samples = series_samples(1.0);
-    let times: Vec<_> = samples.iter().map(|(time, _)| *time).collect();
-    let values: Vec<_> = samples.iter().map(|(_, value)| *value).collect();
-    let hosts: Vec<_> = times.iter().map(|_| "web-1").collect();
-    // The two rows differ only in the zone column: empty against null.
-    let zones: [ArrayRef; 2] = [
-        Arc::new(StringArray::from(vec![""; times.len()])),
-        Arc::new(StringArray::from(vec![None::<&str>; times.len()])),
-    ];
-    zones
-        .into_iter()
-        .map(|zone| {
-            RecordBatch::try_new(
-                Arc::clone(&schema),
-                vec![
-                    Arc::new(TimestampMillisecondArray::from(times.clone())),
-                    Arc::new(Float64Array::from(values.clone())),
-                    Arc::new(StringArray::from(hosts.clone())),
-                    zone,
-                ],
-            )
-            .expect("one batch of the two rows")
-        })
-        .collect()
-}
-
-/// `rate(x[5m])` over an input that carries `zone` as well as `host`, with both in the output.
-fn zone_node() -> FusedSeries {
-    let schema = zone_schema();
-    let tree = rate_tree();
-    let binding = FusedSeriesBinding {
-        tags: vec![HOST.to_string(), ZONE.to_string(), NAME.to_string()],
-        ..binding(&tree)
-    };
-    FusedSeries::try_new(
-        Arc::from(PROMQL),
-        tree,
-        binding,
-        LogicalPlan::EmptyRelation(EmptyRelation {
-            produce_one_row: false,
-            schema: Arc::clone(&schema).to_dfschema_ref().unwrap(),
-        }),
-        zone_output_schema().to_dfschema_ref().unwrap(),
-    )
-    .expect("the binding matches the tree")
-}
-
 /// The output of [`zone_node`]: the field, `host`, `zone`, the dropped name and the grid times.
 fn zone_output_schema() -> SchemaRef {
     Arc::new(Schema::new(vec![
@@ -610,24 +721,89 @@ fn zone_output_schema() -> SchemaRef {
     ]))
 }
 
+/// `rate(x[5m])` over frames that carry `zone` as well as `host`, with both in the output.
+fn zone_node() -> FusedSeries {
+    let tree = rate_tree();
+    let binding = binding_of(
+        &tree,
+        raw_schema(),
+        vec![HOST.to_string(), ZONE.to_string(), NAME.to_string()],
+    );
+    FusedSeries::try_new(
+        Arc::from(PROMQL),
+        tree,
+        binding,
+        input_plan(frame_schema()),
+        zone_output_schema().to_dfschema_ref().unwrap(),
+    )
+    .expect("the binding matches the tree")
+}
+
+/// The instant-selection node over frames that carry `zone` as well as `host`.
+///
+/// Its output labels are the input's, so an empty-valued label reaches this node's own state as the
+/// frame wrote it.
+fn instant_node() -> FusedSeries {
+    let tree = instant_tree();
+    let binding = binding_of(
+        &tree,
+        raw_schema(),
+        vec![HOST.to_string(), ZONE.to_string(), NAME.to_string()],
+    );
+    FusedSeries::try_new(
+        Arc::from(PROMQL),
+        tree,
+        binding,
+        input_plan(frame_schema()),
+        zone_output_schema().to_dfschema_ref().unwrap(),
+    )
+    .expect("the binding matches the tree")
+}
+
+/// The two chunks of the collision: one series once with an empty `zone` and once with no `zone`.
+///
+/// They are one identity to Prometheus — an empty label value is no label — but two different raw
+/// series: the scan encodes the empty value and the absent one differently, so they reach the
+/// runtime as two series that close on the same label set.
+fn zone_chunks() -> Vec<RawSeriesChunk> {
+    let rows = series_samples(1.0);
+    vec![
+        chunk_of(
+            &raw_schema(),
+            "zone-empty",
+            metric(&[(HOST, "web-1"), (ZONE, "")], Some(METRIC)),
+            &times_of(&rows),
+            vec![values_of(&rows)],
+            true,
+        ),
+        chunk_of(
+            &raw_schema(),
+            "zone-absent",
+            metric(&[(HOST, "web-1")], Some(METRIC)),
+            &times_of(&rows),
+            vec![values_of(&rows)],
+            true,
+        ),
+    ]
+}
+
 /// The canonical label set these tests collide on, as the runtime renders it in its refusal.
 const CLOSED_LABEL_SET: &str =
     "the node at root already closed the label set [(\"host\", \"web-1\")]";
 
-/// Two rows that are already one Prometheus identity, carried under two different raw keys, are
-/// refused by the runtime that would have to answer both.
+/// Two raw series that are already one Prometheus identity are refused by the runtime that would
+/// have to answer both.
 ///
 /// `{host="web-1", zone=""}` and `{host="web-1"}` are one identity to Prometheus: an empty label
-/// value is no label, so the two rows are the same series and `rate` creates no collision here.
-/// What the two *raw* keys differ in is the empty `zone` against the null one, which the scan
-/// encodes differently and the converter therefore opens two input series for. `rate` drops the
+/// value is no label, so the two chunks are the same series and `rate` creates no collision here.
+/// What the two *raw* keys differ in is the empty `zone` against the absent one, which the scan
+/// encodes differently and the decoder therefore opens two input series for. `rate` drops the
 /// metric name and the empty value, so the second raw series arrives on the canonical label set the
 /// runtime has already closed; the refusal names that label set rather than answering it, because
 /// answering would report one series twice.
 #[tokio::test]
 async fn differing_raw_keys_that_close_on_one_label_set_are_refused() {
-    let exec = zone_node()
-        .to_execution_plan(memory_exec_with(zone_schema(), zone_batches()))
+    let exec = plan_of(zone_node(), memory_exec(frames(&zone_chunks())))
         .expect("the runtime executes a rate tree");
     let error = collect(exec, SessionContext::default().task_ctx())
         .await
@@ -640,21 +816,20 @@ async fn differing_raw_keys_that_close_on_one_label_set_are_refused() {
 
 /// A runtime error after buffered rows is reported, and the terminated stream releases its input.
 ///
-/// The converter holds the batch it is reading, the chunk it is building and its own source, so a
+/// The decoder holds the batch it is reading, the chunks it decoded from it and its own source, so a
 /// stream that kept its input after the error would hold all three until the consumer drops it. The
 /// collision above is the error: it happens on the second series, after the first one's rows have
 /// been handed out.
 #[tokio::test]
 async fn a_runtime_error_after_rows_releases_the_input() {
     let dropped = Arc::new(AtomicBool::new(false));
-    let schema = zone_schema();
-    let batches = zone_batches();
+    let schema = frame_schema();
+    let batches = frames(&zone_chunks());
     let source = staged(schema.clone(), {
         let dropped = Arc::clone(&dropped);
         move || Box::pin(StagedStream::new(schema.clone(), batches.clone()).dropped(&dropped))
     });
-    let mut stream = zone_node()
-        .to_execution_plan(source)
+    let mut stream = plan_of(zone_node(), source)
         .expect("the runtime executes a rate tree")
         .execute(0, SessionContext::default().task_ctx())
         .expect("the node executes");
@@ -690,9 +865,9 @@ async fn a_runtime_error_after_rows_releases_the_input() {
 ///
 /// This is the collision the canonical label set exists for, at the layer that sees it:
 /// `http_requests_total{host="web-1"}` and `http_errors_total{host="web-1"}` reach the window as two
-/// identities and leave it as `{host="web-1"}`. [`FusedSeries`] binds one constant metric name for
-/// every series it reads, so this collision cannot be staged through it; the runtime is driven
-/// directly, with the names on the raw chunks.
+/// identities and leave it as `{host="web-1"}`. [`FusedSeries`] reads one metric per scan, so this
+/// collision cannot be staged through it; the runtime is driven directly, with the names on the raw
+/// chunks.
 #[test]
 fn two_differently_named_series_collide_once_rate_drops_the_name() {
     let tree = rate_tree();
@@ -707,14 +882,31 @@ fn two_differently_named_series_collide_once_rate_drops_the_name() {
     );
     let samples = series_samples(1.0);
     execution
-        .push_raw(raw_chunk(METRIC, &samples))
+        .push_raw(chunk_of(
+            &raw_schema(),
+            METRIC,
+            metric(&[(HOST, "web-1")], Some(METRIC)),
+            &times_of(&samples),
+            vec![values_of(&samples)],
+            true,
+        ))
         .expect("the first series is a legal raw chunk");
+    assert!(
+        execution
+            .next_chunk()
+            .expect("the first series runs")
+            .is_some(),
+        "the first series is answered"
+    );
     execution
-        .next_chunk()
-        .expect("the first series is answered")
-        .expect("rate decides every grid point of the first series");
-    execution
-        .push_raw(raw_chunk("http_errors_total", &samples))
+        .push_raw(chunk_of(
+            &raw_schema(),
+            "http_errors_total",
+            metric(&[(HOST, "web-1")], Some("http_errors_total")),
+            &times_of(&samples),
+            vec![values_of(&samples)],
+            true,
+        ))
         .expect("the second series is a legal raw chunk");
     let error = execution
         .next_chunk()
@@ -725,30 +917,6 @@ fn two_differently_named_series_collide_once_rate_drops_the_name() {
     );
 }
 
-/// One raw chunk of the series `name{host="web-1"}`: one sample per minute, closing its series.
-fn raw_chunk(name: &str, samples: &[(i64, f64)]) -> RawSeriesChunk {
-    let times = samples.iter().map(|(time, _)| *time).collect::<Vec<_>>();
-    let values = samples.iter().map(|(_, value)| *value).collect::<Vec<_>>();
-    RawSeriesChunk::try_new(
-        &raw_schema(),
-        Buffer::from_slice_ref(name.as_bytes()),
-        Some(
-            MetricName::try_new(
-                Some(Arc::from(name)),
-                vec![Label {
-                    name: Arc::from(HOST),
-                    value: Arc::from("web-1"),
-                }],
-            )
-            .expect("the labels of this test are legal"),
-        ),
-        RawTimestamps::try_new(ScalarBuffer::from(times)).expect("ascending timestamps"),
-        SeriesFields::one(Samples::from_float(Float64Array::from(values))),
-        true,
-    )
-    .expect("one series' samples are a legal raw chunk")
-}
-
 /// The grid timestamps of the statement, in order.
 fn grid_timestamps() -> Vec<i64> {
     (0..GRID_LEN)
@@ -775,15 +943,23 @@ impl CountingWake {
     }
 }
 
-/// One series as many chunks that settle nothing: one sample per batch, all before the grid starts.
+/// One series as many chunks that settle nothing: one sample per chunk, all before the grid starts.
 ///
 /// The counter advances one second per second, so `rate` answers 1 at every grid point once the
 /// series is read to its end; until a sample reaches the grid's first point, a push decides
 /// nothing and settles no chunk.
-fn quiet_series_batches() -> Vec<RecordBatch> {
-    (0..QUIET_SAMPLES)
-        .map(|index| series_batch("web-1", &[(index * QUIET_STEP, index as f64)]))
-        .collect()
+fn quiet_series_frames() -> Vec<RecordBatch> {
+    frames(
+        &(0..QUIET_SAMPLES)
+            .map(|index| {
+                series_chunk(
+                    "web-1",
+                    &[(index * QUIET_STEP, index as f64)],
+                    index == QUIET_SAMPLES - 1,
+                )
+            })
+            .collect::<Vec<_>>(),
+    )
 }
 
 /// Asserts that `rows` are `rates`' series, each over the whole grid in order.
@@ -837,9 +1013,8 @@ fn fmt_for_explain_prints_the_promql() {
 /// The physical node prints the tree it was built from, rendered by the tree's own display.
 #[test]
 fn fmt_as_prints_the_native_tree() {
-    let exec = fused(rate_tree())
-        .to_execution_plan(memory_exec(Vec::new()))
-        .expect("the runtime executes a rate tree");
+    let exec =
+        plan(rate_tree(), memory_exec(Vec::new())).expect("the runtime executes a rate tree");
     let text = displayable(exec.as_ref()).indent(false).to_string();
 
     assert!(
@@ -860,8 +1035,7 @@ fn fmt_as_prints_the_native_tree() {
 /// `rate(x[5m])` runs end to end over two raw series: values, tags and the dropped metric name.
 #[tokio::test]
 async fn runs_rate_over_two_series() {
-    let exec = fused(rate_tree())
-        .to_execution_plan(memory_exec(two_series_batches()))
+    let exec = plan(rate_tree(), memory_exec(two_series_frames()))
         .expect("the runtime executes a rate tree");
     let batches = collect(exec, SessionContext::default().task_ctx())
         .await
@@ -899,8 +1073,7 @@ async fn runs_rate_over_two_series() {
 /// A series that spans several input chunks keeps its absolute grid positions and closes once.
 #[tokio::test]
 async fn a_series_spanning_chunks_keeps_absolute_positions_and_closes_once() {
-    let exec = fused(rate_tree())
-        .to_execution_plan(memory_exec(one_series_batches()))
+    let exec = plan(rate_tree(), memory_exec(one_series_frames()))
         .expect("the runtime executes a rate tree");
     let batches = collect(Arc::clone(&exec), SessionContext::default().task_ctx())
         .await
@@ -928,8 +1101,7 @@ async fn a_series_spanning_chunks_keeps_absolute_positions_and_closes_once() {
 #[tokio::test]
 async fn a_tree_the_runtime_cannot_run_is_rejected_when_the_exec_is_built() {
     let node = fused(aggregate_tree());
-    let error = node
-        .to_execution_plan(memory_exec(Vec::new()))
+    let error = plan_of(node, memory_exec(Vec::new()))
         .expect_err("the runtime does not execute aggregate nodes");
     assert!(
         matches!(error, DataFusionError::NotImplemented(_)),
@@ -952,7 +1124,7 @@ fn a_tree_bound_to_another_grid_is_rejected() {
         Arc::from(PROMQL),
         tree,
         binding,
-        input_plan(),
+        input_plan(frame_schema()),
         output_schema().to_dfschema_ref().unwrap(),
     )
     .expect_err("the tree is not evaluated on the declared grid");
@@ -983,7 +1155,7 @@ fn a_separately_allocated_grid_with_an_equal_spec_is_accepted() {
         Arc::from(PROMQL),
         tree,
         binding,
-        input_plan(),
+        input_plan(frame_schema()),
         output_schema().to_dfschema_ref().unwrap(),
     )
     .expect("an equal grid spec is the same grid");
@@ -992,9 +1164,8 @@ fn a_separately_allocated_grid_with_an_equal_spec_is_accepted() {
 /// An input that ends without a series still runs the tree, and yields no rows.
 #[tokio::test]
 async fn an_empty_input_runs_the_tree_and_yields_no_rows() {
-    let exec = fused(rate_tree())
-        .to_execution_plan(memory_exec(Vec::new()))
-        .expect("the runtime executes a rate tree");
+    let exec =
+        plan(rate_tree(), memory_exec(Vec::new())).expect("the runtime executes a rate tree");
     let batches = collect(Arc::clone(&exec), SessionContext::default().task_ctx())
         .await
         .expect("a tree with no series in it still closes");
@@ -1011,8 +1182,8 @@ async fn an_empty_input_runs_the_tree_and_yields_no_rows() {
 #[tokio::test]
 async fn an_upstream_error_after_rows_is_reported() {
     let dropped = Arc::new(AtomicBool::new(false));
-    let schema = input_schema();
-    let batches = one_series_batches();
+    let schema = frame_schema();
+    let batches = one_series_frames();
     let source = staged(schema.clone(), {
         let dropped = Arc::clone(&dropped);
         move || {
@@ -1023,8 +1194,7 @@ async fn an_upstream_error_after_rows_is_reported() {
             )
         }
     });
-    let mut stream = fused(rate_tree())
-        .to_execution_plan(source)
+    let mut stream = plan(rate_tree(), source)
         .expect("the runtime executes a rate tree")
         .execute(0, SessionContext::default().task_ctx())
         .expect("the node executes");
@@ -1035,15 +1205,12 @@ async fn an_upstream_error_after_rows_is_reported() {
         match stream.next().await {
             Some(Ok(batch)) => answered.push(batch),
             Some(Err(error)) => break error,
-            None => panic!(
-                "the input's error was swallowed after {:?}",
-                rows(&answered)
-            ),
+            None => panic!("the source failure was swallowed: {:?}", rows(&answered)),
         }
     };
     assert!(
         error.to_string().contains("the source failed"),
-        "reported as another error: {error}"
+        "reported another error: {error}"
     );
     // Still alive here: the release is the stream's, not the drop of a temporary.
     assert!(
@@ -1057,8 +1224,7 @@ async fn an_upstream_error_after_rows_is_reported() {
 async fn a_pending_boundary_loses_no_rows() {
     let expected = rows(
         &collect(
-            fused(rate_tree())
-                .to_execution_plan(memory_exec(one_series_batches()))
+            plan(rate_tree(), memory_exec(one_series_frames()))
                 .expect("the runtime executes a rate tree"),
             SessionContext::default().task_ctx(),
         )
@@ -1067,17 +1233,16 @@ async fn a_pending_boundary_loses_no_rows() {
     );
     assert!(!expected.is_empty(), "the fixture decides some grid points");
 
-    let schema = input_schema();
-    let batches = one_series_batches();
+    let schema = frame_schema();
+    let batches = one_series_frames();
     let source = staged(schema.clone(), move || {
         Box::pin(
             StagedStream::new(schema.clone(), batches.clone())
-                // One batch is in the converter's hands when the stream reports pending.
+                // One batch is in the decoder's hands when the stream reports pending.
                 .pending_after(1),
         )
     });
-    let mut stream = fused(rate_tree())
-        .to_execution_plan(source)
+    let mut stream = plan(rate_tree(), source)
         .expect("the runtime executes a rate tree")
         .execute(0, SessionContext::default().task_ctx())
         .expect("the node executes");
@@ -1098,125 +1263,78 @@ async fn a_pending_boundary_loses_no_rows() {
     assert_eq!(expected, rows(&batches));
 }
 
-/// A tag the input has no column for is refused when the plan is built, naming the column.
+/// A frame whose labels are encoded in its key is refused when the plan is built.
+///
+/// The tags this node declares are read off each chunk's labels, and a key-mode frame carries none:
+/// decoding it with a binding that promises complete labels would produce chunks that name no
+/// series. The frame's own declared layout says which of the two it is, so the mismatch is caught
+/// before the first batch, not when the missing labels reach the output.
 #[test]
-fn a_missing_tag_column_is_rejected_when_the_plan_is_built() {
-    let tree = rate_tree();
-    let binding = FusedSeriesBinding {
-        tags: vec![HOST.to_string(), ZONE.to_string(), NAME.to_string()],
-        ..binding(&tree)
-    };
-    let error = FusedSeries::try_new(
-        Arc::from(PROMQL),
-        tree,
-        binding,
-        input_plan(),
-        zone_output_schema().to_dfschema_ref().unwrap(),
-    )
-    .expect("the logical node does not see the input's columns")
-    .to_execution_plan(memory_exec(Vec::new()))
-    .expect_err("the input has no zone column");
+fn a_frame_whose_labels_are_encoded_is_rejected_when_the_plan_is_built() {
+    let keyed =
+        raw_series_frame_schema(&raw_schema(), RawLabelMode::Key).expect("key labels are a frame");
+    let error = plan(rate_tree(), memory_exec_with(keyed, Vec::new()))
+        .expect_err("the frames carry no labels to read tags from");
     assert!(
-        error.to_string().contains("no column \"zone\""),
+        error.to_string().contains("raw frame"),
         "refused for another reason: {error}"
     );
 }
 
-/// A tag column whose type the converter does not read as a label is refused when the plan is
-/// built, naming the column.
+/// A frame that carries an evaluation grid is refused when the plan is built.
+///
+/// A frame holds raw series: its rows are samples, not grid positions, and the grid every output
+/// row is addressed on is the tree's own. A schema that declares a grid is one this node would
+/// read positions out of, so it is refused rather than decoded.
 #[test]
-fn a_wrong_typed_tag_column_is_rejected_when_the_plan_is_built() {
-    // `zone` is numeric here, which no label of a raw series can be read from.
-    let schema = Arc::new(Schema::new(vec![
-        Field::new(
-            TIME,
-            DataType::Timestamp(TimeUnit::Millisecond, None),
-            false,
-        ),
-        Field::new(VALUE, DataType::Float64, true),
-        Field::new(HOST, DataType::Utf8, true),
-        Field::new(ZONE, DataType::Float64, true),
-    ]));
-    let tree = rate_tree();
-    let binding = FusedSeriesBinding {
-        tags: vec![HOST.to_string(), ZONE.to_string(), NAME.to_string()],
-        ..binding(&tree)
-    };
-    let error = FusedSeries::try_new(
-        Arc::from(PROMQL),
-        tree,
-        binding,
-        LogicalPlan::EmptyRelation(EmptyRelation {
-            produce_one_row: false,
-            schema: Arc::clone(&schema).to_dfschema_ref().unwrap(),
-        }),
-        zone_output_schema().to_dfschema_ref().unwrap(),
-    )
-    .expect("the logical node does not see the input's column types")
-    .to_execution_plan(memory_exec_with(schema, Vec::new()))
-    .expect_err("a numeric column cannot carry a label");
+fn a_frame_with_an_evaluation_grid_is_rejected_when_the_plan_is_built() {
+    let mut metadata = frame_schema().metadata().clone();
+    metadata.insert("greptime.series.grid".to_string(), "v1".to_string());
+    let gridded = Arc::new(Schema::new_with_metadata(
+        frame_schema().fields().clone(),
+        metadata,
+    ));
+    let error = plan(rate_tree(), memory_exec_with(gridded, Vec::new()))
+        .expect_err("the frame declares a grid");
     assert!(
-        error.to_string().contains("tag \"zone\""),
-        "refused without naming the column: {error}"
+        error.to_string().contains("raw frame"),
+        "refused for another reason: {error}"
     );
 }
 
-/// An input column the binding does not map is refused when the plan is built, naming the column.
+/// A frame of another series schema is refused when the plan is built.
 ///
-/// `zone` is part of these series' identity to the scan but not to the binding, so reading the
-/// input would drop it: rows a Prometheus query keeps apart would be answered as one series with
-/// the wrong labels.
+/// The tree's input node was compiled against one schema, and a frame for a different one names
+/// fields the tree does not read and omits ones it does: decoding it would hand the tree a column
+/// it never bound. Nothing about the two schemas says which one was meant, so the frame is refused
+/// instead of half-read.
 #[test]
-fn an_input_column_the_binding_does_not_map_is_rejected() {
-    let tree = rate_tree();
-    let binding = binding(&tree);
-    let error = FusedSeries::try_new(
-        Arc::from(PROMQL),
-        tree,
-        binding,
-        LogicalPlan::EmptyRelation(EmptyRelation {
-            produce_one_row: false,
-            schema: zone_schema().to_dfschema_ref().unwrap(),
-        }),
-        output_schema().to_dfschema_ref().unwrap(),
-    )
-    .expect("the logical node does not see the input's columns")
-    .to_execution_plan(memory_exec_with(zone_schema(), Vec::new()))
-    .expect_err("the binding does not account for the zone column");
+fn a_frame_of_another_series_schema_is_rejected_when_the_plan_is_built() {
+    let other = frame_schema_of(&two_field_schema());
+    let error = plan(rate_tree(), memory_exec_with(other, Vec::new()))
+        .expect_err("the tree's input reads one field");
     assert!(
-        error.to_string().contains("column \"zone\""),
-        "refused without naming the column: {error}"
+        error.to_string().contains("raw frame"),
+        "refused for another reason: {error}"
     );
 }
 
-/// An input with two columns of one name is refused when the plan is built.
+/// An input with no partitions is refused when the plan is built.
 ///
-/// The coverage check reads names, so both `host` columns pass it as a role the binding maps, while
-/// the mapping itself reads `index_of`, which answers with one index: the other column would be
-/// part of these rows' identity that a raw series never carries, and rows a Prometheus query keeps
-/// apart would be answered as one series with the wrong labels -- and no error. Nothing about the
-/// two columns says which one was meant, so the schema is refused rather than one of them chosen.
+/// The node runs one evaluator per input partition and adds none of its own, so an input with no
+/// partition is one that could answer nothing: the plan is refused rather than built as a node that
+/// would report an empty result for data it never read.
 #[test]
-fn an_input_with_two_columns_of_one_name_is_rejected() {
-    let duplicated = Arc::new(Schema::new(vec![
-        Field::new(
-            TIME,
-            DataType::Timestamp(TimeUnit::Millisecond, None),
-            false,
-        ),
-        Field::new(VALUE, DataType::Float64, true),
-        Field::new(HOST, DataType::Utf8, true),
-        // The second `host` column, which the name-based coverage check accepts just as well.
-        Field::new(HOST, DataType::Utf8, true),
-    ]));
-    let error = fused(rate_tree())
-        .to_execution_plan(memory_exec_with(duplicated, Vec::new()))
-        .expect_err("the binding maps one of the two host columns");
+fn an_input_with_no_partitions_is_rejected_when_the_plan_is_built() {
+    let schema = frame_schema();
+    let empty = Arc::new(DataSourceExec::new(Arc::new(
+        MemorySourceConfig::try_new(&[], Arc::clone(&schema), None)
+            .expect("no partitions is a legal source"),
+    )));
+    let error = plan(rate_tree(), empty).expect_err("there is no partition to run");
     assert!(
-        error
-            .to_string()
-            .contains("more than one column named \"host\""),
-        "refused without naming the duplicate: {error}"
+        error.to_string().contains("no partitions"),
+        "refused for another reason: {error}"
     );
 }
 
@@ -1229,8 +1347,8 @@ fn an_input_with_two_columns_of_one_name_is_rejected() {
 /// the second one covers the grid, so a poll that resumes properly still answers both.
 #[tokio::test]
 async fn one_poll_reads_at_most_the_budget_and_loses_nothing() {
-    let mut batches = quiet_series_batches();
-    batches.push(series_batch("web-2", &series_samples(2.0)));
+    let mut batches = quiet_series_frames();
+    batches.push(frame(&series_chunk("web-2", &series_samples(2.0), true)));
     let total = batches.len();
     assert!(
         total > POLL_BUDGET,
@@ -1239,13 +1357,12 @@ async fn one_poll_reads_at_most_the_budget_and_loses_nothing() {
 
     let pulled = Arc::new(AtomicUsize::new(0));
     let woken = Arc::new(CountingWake::default());
-    let schema = input_schema();
+    let schema = frame_schema();
     let source = staged(schema.clone(), {
         let pulled = Arc::clone(&pulled);
         move || Box::pin(StagedStream::new(schema.clone(), batches.clone()).counted(&pulled))
     });
-    let mut stream = fused(rate_tree())
-        .to_execution_plan(source)
+    let mut stream = plan(rate_tree(), source)
         .expect("the runtime executes a rate tree")
         .execute(0, SessionContext::default().task_ctx())
         .expect("the node executes");
@@ -1259,11 +1376,11 @@ async fn one_poll_reads_at_most_the_budget_and_loses_nothing() {
     );
     let read = pulled.load(Ordering::SeqCst);
     assert!(read > 0, "the poll did read its input");
-    // One step polls the converter once, and the converter reads at most two batches per poll: the
-    // key it is holding and the first row of the next, which is what tells it the first key ended.
-    // The bound is therefore two batches per step, and the budget is what it is measured in.
+    // One step polls the decoder once, and the decoder reads at most one frame per poll: every
+    // frame here carries one chunk. The bound is therefore one frame per step, and the budget is
+    // what it is measured in.
     assert!(
-        read <= 2 * POLL_BUDGET + 1,
+        read <= POLL_BUDGET + 1,
         "one poll read {read} of {total} batches, past its budget of {POLL_BUDGET} steps"
     );
     assert!(
@@ -1285,23 +1402,28 @@ async fn one_poll_reads_at_most_the_budget_and_loses_nothing() {
 
 /// A batch that is ready is handed over before any batch behind it is read.
 ///
-/// Each series below is one batch, and a series closes on the first batch of the next one, so the
-/// first answer needs two batches and nothing more. A poll that read on would answer the same batch
-/// only after the whole input, which is exactly what the budget above exists to prevent.
+/// Each series below is one chunk, and a chunk that carries `is_last` closes its series, so the
+/// first answer needs only the first chunk. A poll that read on would answer the same batch only
+/// after the whole input, which is exactly what the budget above exists to prevent.
 #[tokio::test]
 async fn a_ready_batch_is_returned_before_later_input_is_read() {
     let batches = (0..20)
-        .map(|index| series_batch(&format!("web-{index}"), &series_samples(1.0)))
+        .map(|index| {
+            frame(&series_chunk(
+                &format!("web-{index}"),
+                &series_samples(1.0),
+                true,
+            ))
+        })
         .collect::<Vec<_>>();
     let total = batches.len();
     let pulled = Arc::new(AtomicUsize::new(0));
-    let schema = input_schema();
+    let schema = frame_schema();
     let source = staged(schema.clone(), {
         let pulled = Arc::clone(&pulled);
         move || Box::pin(StagedStream::new(schema.clone(), batches.clone()).counted(&pulled))
     });
-    let mut stream = fused(rate_tree())
-        .to_execution_plan(source)
+    let mut stream = plan(rate_tree(), source)
         .expect("the runtime executes a rate tree")
         .execute(0, SessionContext::default().task_ctx())
         .expect("the node executes");
@@ -1313,13 +1435,13 @@ async fn a_ready_batch_is_returned_before_later_input_is_read() {
         .expect("the first series is answered without an error");
     let answered = rows(&[first]);
     assert_series_rates(&answered, &[(1.0, "web-0")]);
-    // Two batches: the first series' own, and the one whose first key closes it.
+    // One batch: the series' own chunk closes it, so nothing behind it has to be read.
     assert_eq!(
-        2,
+        1,
         pulled.load(Ordering::SeqCst),
         "the answer waited for input behind the series it reports"
     );
-    assert!(total > 2, "the fixture has to outlast the first answer");
+    assert!(total > 1, "the fixture has to outlast the first answer");
 }
 
 /// A chunk whose rows form several runs is converted row by row from the rows it carries.
@@ -1404,15 +1526,17 @@ fn error_chain(error: &DataFusionError) -> String {
 /// Two partitions, one series each, and both series leave the window on the same label set.
 ///
 /// `rate` drops `__name__`, so `http_requests_total{host="web-1"}` reads as `{host="web-1"}` in
-/// the output. `PerSeries` distribution keeps each of the two whole inside its own partition, so a
-/// check that lived inside one execution would see one series each and report nothing: only state
-/// shared by the partitions sees that the output vector holds the label set twice.
+/// the output. The key column keeps each of the two whole inside its own partition, so a check that
+/// lived inside one execution would see one series each and report nothing: only state shared by
+/// the partitions sees that the output vector holds the label set twice.
 #[tokio::test]
 async fn a_duplicate_label_set_across_partitions_is_refused() {
-    let series = vec![series_batch("web-1", &series_samples(1.0))];
-    let exec = fused(rate_tree())
-        .to_execution_plan(memory_exec_partitioned(vec![series.clone(), series]))
-        .expect("the runtime executes a rate tree");
+    let series = frame(&series_chunk("web-1", &series_samples(1.0), true));
+    let exec = plan(
+        rate_tree(),
+        memory_exec_partitioned(vec![vec![series.clone()], vec![series]]),
+    )
+    .expect("the runtime executes a rate tree");
     assert_eq!(
         2,
         exec.properties().output_partitioning().partition_count(),
@@ -1441,20 +1565,21 @@ async fn a_duplicate_label_set_across_partitions_is_refused() {
 /// merge to matrix assembly, so the check has to allow what it would otherwise refuse.
 #[tokio::test]
 async fn a_duplicate_label_set_on_disjoint_grid_points_is_emitted() {
-    let first = vec![series_batch(
+    let first = vec![frame(&series_chunk(
         "web-1",
         &[(60_000, 60.0), (GRID_START, GRID_START as f64 / 1_000.0)],
-    )];
+        true,
+    ))];
     let last = GRID_START + (GRID_LEN as i64 - 1) * GRID_STEP;
-    let second = vec![series_batch(
+    let second = vec![frame(&series_chunk(
         "web-1",
         &[
             (GRID_START, GRID_START as f64 / 1_000.0),
             (last, last as f64 / 1_000.0),
         ],
-    )];
-    let exec = fused(rate_tree())
-        .to_execution_plan(memory_exec_partitioned(vec![first, second]))
+        true,
+    ))];
+    let exec = plan(rate_tree(), memory_exec_partitioned(vec![first, second]))
         .expect("the runtime executes a rate tree");
 
     let batches = collect(exec, SessionContext::default().task_ctx())
@@ -1475,33 +1600,35 @@ async fn a_duplicate_label_set_on_disjoint_grid_points_is_emitted() {
     }
 }
 
-/// What a label set holds stays charged after the stream that first claimed it ends.
+/// What a label set holds stays charged after the stream that first claimed it ends, and what a
+/// later partition adds to it is charged as well.
 ///
 /// The map belongs to the plan: it outlives every partition and is released with the plan. Charging
 /// each stream for what it inserted returned the bytes of a label set when *that* stream ended,
 /// while the entry the next partition is checked against stayed in the map, so everything the map
 /// held after the first partition was invisible to the pool. What is charged here is therefore read
-/// between the partitions: it has to survive the first stream and grow no further when the second
-/// partition lands on the label set the first one claimed.
+/// between the partitions: it has to survive the first stream, and the runs the second partition
+/// merges into the entry have to cost what they hold.
 #[tokio::test]
 async fn a_label_set_stays_charged_after_its_first_stream_ends() {
     // The first series holds a value only at the first grid point and the second only at the last,
     // so the two share a final label set without sharing an evaluation timestamp: the second
     // partition is legal and has to be answered, and it has to find the first one's entry there.
-    let first = vec![series_batch(
+    let first = vec![frame(&series_chunk(
         "web-0",
         &[(60_000, 60.0), (GRID_START, GRID_START as f64 / 1_000.0)],
-    )];
+        true,
+    ))];
     let last = GRID_START + (GRID_LEN as i64 - 1) * GRID_STEP;
-    let second = vec![series_batch(
+    let second = vec![frame(&series_chunk(
         "web-0",
         &[
             (GRID_START, GRID_START as f64 / 1_000.0),
             (last, last as f64 / 1_000.0),
         ],
-    )];
-    let exec = fused(rate_tree())
-        .to_execution_plan(memory_exec_partitioned(vec![first, second]))
+        true,
+    ))];
+    let exec = plan(rate_tree(), memory_exec_partitioned(vec![first, second]))
         .expect("the runtime executes a rate tree");
     assert_eq!(
         2,
@@ -1528,7 +1655,7 @@ async fn a_label_set_stays_charged_after_its_first_stream_ends() {
     );
 
     // The second partition lands on that same label set on points the first one does not hold: it
-    // is answered, and what it adds to the entry is already paid for.
+    // is answered, and the run it merges into the entry is what its charge grows by.
     let mut stream = exec
         .execute(1, Arc::clone(&context))
         .expect("the node executes");
@@ -1538,10 +1665,11 @@ async fn a_label_set_stays_charged_after_its_first_stream_ends() {
     }
     assert_eq!(1, answered.len(), "disjoint points are legal");
     drop(stream);
-    assert_eq!(
-        charged,
-        pool.reserved(),
-        "a label set the map already holds is not charged a second time"
+    assert!(
+        pool.reserved() > charged,
+        "the second partition's run grew the entry, but nothing was charged for it: {} is not \
+         above {charged}",
+        pool.reserved()
     );
 
     // The entries live as long as the plan, so dropping it is what gives their bytes back.
@@ -1560,14 +1688,14 @@ async fn every_partition_evaluates_its_own_series() {
         .into_iter()
         .enumerate()
         .map(|(index, scale)| {
-            vec![series_batch(
+            vec![frame(&series_chunk(
                 &format!("web-{index}"),
                 &series_samples(scale),
-            )]
+                true,
+            ))]
         })
         .collect::<Vec<_>>();
-    let exec = fused(rate_tree())
-        .to_execution_plan(memory_exec_partitioned(partitions))
+    let exec = plan(rate_tree(), memory_exec_partitioned(partitions))
         .expect("the runtime executes a rate tree");
     assert_eq!(
         3,
@@ -1599,6 +1727,30 @@ async fn every_partition_evaluates_its_own_series() {
     }
 }
 
+/// The node asks the optimizer to keep each series in one partition.
+///
+/// The node runs one window per series and closes it on the chunk carrying `is_last`, so a series
+/// the optimizer split across partitions would be answered as two halves by two trees. What it
+/// declares is a key requirement on the frame's key column, which is the column a producer writes
+/// one series' encoded identity into: the same key then has to reach the same partition, while the
+/// parallelism of an input that already owns whole series per partition is left alone.
+#[test]
+fn the_input_is_required_to_be_key_partitioned() {
+    let exec = plan(rate_tree(), memory_exec_partitioned(vec![vec![], vec![]]))
+        .expect("the runtime executes a rate tree");
+    let distributions = exec.input_distribution_requirements().into_per_child();
+    assert_eq!(1, distributions.len(), "one requirement, one child");
+    let Distribution::KeyPartitioned(keys) = &distributions[0] else {
+        panic!("a single partition would give up the input's parallelism: {distributions:?}");
+    };
+    let key = keys
+        .first()
+        .and_then(|key| (key.as_ref() as &dyn std::any::Any).downcast_ref::<ColumnExpr>())
+        .expect("the key requirement names a column of the frame");
+    assert_eq!("read_key", key.name(), "one read_key is one series");
+    assert_eq!(0, key.index(), "read_key is the frame's first column");
+}
+
 /// A partition index the input does not have is refused rather than answered.
 ///
 /// The node runs one evaluator per input partition and adds none of its own, so it forwards the
@@ -1606,13 +1758,11 @@ async fn every_partition_evaluates_its_own_series() {
 /// refusal, not a series read from another partition.
 #[tokio::test]
 async fn a_partition_past_the_inputs_count_is_refused() {
-    let schema = input_schema();
+    let schema = frame_schema();
     let source = staged(schema.clone(), move || {
         Box::pin(StagedStream::new(schema.clone(), Vec::new()))
     });
-    let exec = fused(rate_tree())
-        .to_execution_plan(source)
-        .expect("the runtime executes a rate tree");
+    let exec = plan(rate_tree(), source).expect("the runtime executes a rate tree");
     assert_eq!(
         1,
         exec.properties().output_partitioning().partition_count(),
@@ -1629,3 +1779,242 @@ async fn a_partition_past_the_inputs_count_is_refused() {
         "refused for another reason: {error}"
     );
 }
+
+/// The `rate(x[5m])` tree over two fields, with both fields and the tags in the output.
+fn two_field_node() -> FusedSeries {
+    let tree = rate_tree_of(two_field_schema());
+    let binding = binding_of(
+        &tree,
+        two_field_schema(),
+        vec![HOST.to_string(), NAME.to_string()],
+    );
+    FusedSeries::try_new(
+        Arc::from(PROMQL),
+        tree,
+        binding,
+        input_plan(frame_schema_of(&two_field_schema())),
+        two_field_output_schema().to_dfschema_ref().unwrap(),
+    )
+    .expect("the binding matches the tree")
+}
+
+/// The output of [`two_field_node`]: both fields, the tags and the grid times.
+fn two_field_output_schema() -> SchemaRef {
+    Arc::new(Schema::new(vec![
+        Field::new(VALUE, DataType::Float64, true),
+        Field::new(OTHER, DataType::Float64, true),
+        Field::new(HOST, DataType::Utf8, true),
+        Field::new(NAME, DataType::Utf8, true),
+        Field::new(
+            TIME,
+            DataType::Timestamp(TimeUnit::Millisecond, None),
+            false,
+        ),
+    ]))
+}
+
+/// One chunk of `{__name__=METRIC, host=<host>, __type__=<kind>}` whose `field` holds `rows` and
+/// whose other field is missing everywhere.
+fn one_field_chunk(field: usize, host: &str, kind: &str, last: bool) -> RawSeriesChunk {
+    let rows = series_samples(1.0);
+    let (value, other) = if field == 0 {
+        (values_of(&rows), no_values(&rows))
+    } else {
+        (no_values(&rows), values_of(&rows))
+    };
+    chunk_of(
+        &two_field_schema(),
+        &format!("{host}-{kind}"),
+        metric(&[(HOST, host), (TYPE, kind)], Some(METRIC)),
+        &times_of(&rows),
+        vec![value, other],
+        last,
+    )
+}
+
+/// Two series that close on one label set may still fill it, one field each.
+///
+/// `rate` drops `__name__` and `__type__`, so the two series below leave the window as
+/// `{host="web-1"}` and read as one label set. They hold values in *different* fields, which is the
+/// pair [`crate::grid_series::merge_same_labelset`] merges: one value of one field at each position,
+/// never two. Occupancy is therefore per field, and refusing this pair would refuse a series
+/// Prometheus answers.
+#[tokio::test]
+async fn a_duplicate_label_set_of_complementary_fields_is_emitted() {
+    let exec = plan_of(
+        two_field_node(),
+        memory_exec_partitioned_with(
+            frame_schema_of(&two_field_schema()),
+            vec![
+                frames_of(
+                    &two_field_schema(),
+                    &[one_field_chunk(0, "web-1", "counter", true)],
+                ),
+                frames_of(
+                    &two_field_schema(),
+                    &[one_field_chunk(1, "web-1", "gauge", true)],
+                ),
+            ],
+        ),
+    )
+    .expect("the runtime executes a rate tree");
+    assert_eq!(
+        2,
+        exec.properties().output_partitioning().partition_count(),
+        "this test needs one series in each of two partitions"
+    );
+
+    let batches = collect(exec, SessionContext::default().task_ctx())
+        .await
+        .expect("complementary fields are legal");
+    assert_eq!(2, batches.len(), "each partition emits its own series");
+    // Each batch answers one field of the shared label set and leaves the other missing: the two
+    // together are the merged series, with no value of one field holding a position twice.
+    let answered = batches
+        .iter()
+        .map(|batch| {
+            let host = batch
+                .column_by_name(HOST)
+                .and_then(|host| host.as_any().downcast_ref::<StringArray>())
+                .expect("every batch has its tag column");
+            assert!(
+                (0..host.len()).all(|row| host.value(row) == "web-1"),
+                "both batches answer the one label set"
+            );
+            let present = [VALUE, OTHER]
+                .into_iter()
+                .filter(|field| {
+                    let column = batch
+                        .column_by_name(field)
+                        .unwrap_or_else(|| panic!("every batch has a {field} column"))
+                        .as_any()
+                        .downcast_ref::<Float64Array>()
+                        .expect("the field column is Float64");
+                    column.null_count() < column.len()
+                })
+                .collect::<Vec<_>>();
+            assert_eq!(
+                1,
+                present.len(),
+                "each series fills exactly one field: {present:?}"
+            );
+            present[0]
+        })
+        .collect::<Vec<_>>();
+    let mut answered = answered;
+    answered.sort_unstable();
+    assert_eq!(vec![OTHER, VALUE], answered, "the two fields are filled");
+}
+
+/// Two series that close on one label set may not hold the same field at the same position.
+///
+/// This is the pair no merge resolves: at one grid position one field of the label set would carry
+/// two values. It is refused even though the two series name different `__type__` labels, because
+/// the window drops that label and what reaches the output is one label set twice.
+#[tokio::test]
+async fn a_duplicate_label_set_of_one_field_is_refused() {
+    let exec = plan_of(
+        two_field_node(),
+        memory_exec_partitioned_with(
+            frame_schema_of(&two_field_schema()),
+            vec![
+                frames_of(
+                    &two_field_schema(),
+                    &[one_field_chunk(0, "web-1", "counter", true)],
+                ),
+                frames_of(
+                    &two_field_schema(),
+                    &[one_field_chunk(0, "web-1", "gauge", true)],
+                ),
+            ],
+        ),
+    )
+    .expect("the runtime executes a rate tree");
+
+    let error = collect(exec, SessionContext::default().task_ctx())
+        .await
+        .expect_err("both series hold a value of one field at one position");
+    let message = error_chain(&error);
+    assert!(
+        message.contains("vector cannot contain metrics with the same labelset"),
+        "refused for another reason: {message}"
+    );
+    assert!(
+        message.contains("host=\"web-1\""),
+        "the label set must be readable: {message}"
+    );
+}
+
+/// Two partitions whose label sets differ only in an empty value hold one label set.
+///
+/// `{host="web-1", zone=""}` and `{host="web-1"}` are one series to Prometheus: an empty label value
+/// is no label. An instant selection rewrites nothing, so both label sets reach this node's own
+/// state as their frames wrote them, and a state that compared the tags as they arrived would file
+/// them apart. Neither partition sees the other's series, so nothing else in the plan can refuse
+/// the pair: this node's state is what compares them, and what it has to compare is the label set
+/// Prometheus compares.
+#[tokio::test]
+async fn partitions_whose_label_sets_differ_only_in_an_empty_value_are_refused() {
+    let [empty, absent] = <[_; 2]>::try_from(zone_chunks()).expect("the fixture has two chunks");
+    let exec = plan_of(
+        instant_node(),
+        memory_exec_partitioned(vec![frames(&[empty]), frames(&[absent])]),
+    )
+    .expect("the runtime executes an instant selection");
+    assert_eq!(
+        2,
+        exec.properties().output_partitioning().partition_count(),
+        "this test needs one series in each of two partitions"
+    );
+
+    let error = collect(exec, SessionContext::default().task_ctx())
+        .await
+        .expect_err("both partitions hold one value of one field at the same position");
+    let message = error_chain(&error);
+    assert!(
+        message.contains("vector cannot contain metrics with the same labelset"),
+        "refused for another reason: {message}"
+    );
+    assert!(
+        message.contains("host=\"web-1\""),
+        "the label set must be readable: {message}"
+    );
+}
+
+/// The claim compares the label set Prometheus compares, so an empty value is no label.
+///
+/// Driven directly at the layer that keeps the state: a window and a map both drop an empty-valued
+/// label before a chunk reaches this node, so no tree this crate can build produces the pair below.
+/// The state has to key on the same normalization they apply all the same, or the equality it
+/// enforces would be one Prometheus does not have.
+#[test]
+fn a_claim_compares_label_sets_without_empty_values() {
+    let pool: Arc<dyn MemoryPool> = Arc::new(UnboundedMemoryPool::default());
+    let mut labels = LabelSets::default();
+    labels.attach(&pool);
+    let axis = TimeAxis::from_grid_spans(Arc::clone(&grid()), vec![GridSpan::new(0, 5)])
+        .expect("one run over the whole grid");
+    let series_of = |metric: MetricName| Series {
+        metric,
+        time: axis.clone(),
+        fields: SeriesFields::one(Samples::from_float(Float64Array::from(vec![1.0; 5]))),
+    };
+
+    labels
+        .claim(&series_of(metric(&[(HOST, "h"), (ZONE, "")], Some(METRIC))))
+        .expect("the first label set is claimed");
+    let error = labels
+        .claim(&series_of(metric(&[(HOST, "h")], Some(METRIC))))
+        .expect_err("an empty label value is no label");
+    assert!(
+        error
+            .to_string()
+            .contains("vector cannot contain metrics with the same labelset"),
+        "refused for another reason: {error}"
+    );
+    assert_eq!(
+        vec![(Arc::from(HOST), Arc::from("h"))],
+        label_set(&metric(&[(HOST, "h"), (ZONE, "")], None)),
+        "the empty value is not part of the label set"
+    );
+}