chore: merge narthex into feat/narthex-label-node

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

# Conflicts:
#	src/promql/src/native/execute/tests.rs
#	src/query/src/promql/native_compile/tests.rs

src/promql/src/aggregate.rs

@@ -27,11 +27,14 @@ pub use state::{
 use std::cmp::Reverse;
 use std::collections::hash_map::Entry;
 use std::collections::{BTreeMap, BinaryHeap, HashMap};
+use std::mem::size_of;
 use std::sync::Arc;
 
 use datafusion::error::{DataFusionError, Result};
+use datatypes::arrow::array::Float64Array;
 use datatypes::series::MetricName;
-use datatypes::series::samples::{SampleRef, Series, SeriesSchema};
+use datatypes::series::samples::{SampleRef, Samples, Series, SeriesSchema};
+use datatypes::series::time::{GridSpan, TimeAxis};
 
 use crate::go_format::format_float_f;
 use crate::grid_series::{GridSeriesBuilder, external};
@@ -97,6 +100,7 @@ pub struct FloatAggregator {
     labels: Vec<Arc<str>>,
     group_index: HashMap<LabelKey, usize>,
     groups: Vec<Group>,
+    retained: usize,
 }
 
 impl FloatAggregator {
@@ -104,25 +108,78 @@ impl FloatAggregator {
         schema: &Arc<SeriesSchema>,
         op: AggregateOp,
         grouping: Grouping<'_>,
+    ) -> Result<Self> {
+        Self::try_new_budgeted(schema, op, grouping, |_| Ok(()))
+    }
+
+    pub(crate) fn try_new_budgeted(
+        schema: &Arc<SeriesSchema>,
+        op: AggregateOp,
+        grouping: Grouping<'_>,
+        mut budget: impl FnMut(usize) -> Result<()>,
     ) -> Result<Self> {
         validate_input(schema, &[], grouping)?;
         let (by, labels) = match grouping {
             Grouping::By(labels) => (true, labels),
             Grouping::Without(labels) => (false, labels),
         };
+        let label_storage =
+            grouping_bytes(grouping).saturating_add(labels_capacity_bytes(grouping));
+        budget(label_storage)?;
+        let mut owned_labels: Vec<Arc<str>> = Vec::new();
+        owned_labels
+            .try_reserve_exact(labels.len())
+            .map_err(|error| {
+                DataFusionError::Execution(format!(
+                    "cannot reserve aggregate grouping labels: {error}"
+                ))
+            })?;
+        owned_labels.extend(labels.iter().map(|label| Arc::from(*label)));
+        let retained = owned_labels
+            .capacity()
+            .saturating_mul(size_of::<Arc<str>>())
+            .saturating_add(owned_labels.iter().fold(0usize, |bytes, label| {
+                bytes.saturating_add(label.len() + 2 * size_of::<usize>())
+            }));
         Ok(Self {
             schema: schema.clone(),
             op,
             by,
-            labels: labels.iter().map(|label| Arc::from(*label)).collect(),
+            labels: owned_labels,
             group_index: HashMap::new(),
             groups: Vec::new(),
+            retained,
         })
     }
 
     /// Folds `series` into its group. A series that fails validation leaves the state unchanged.
+    ///
+    /// A budget refusal is not a validation failure: the fold may have opened a group or filled
+    /// blocks before the allocation it could not admit, so a failed [`Self::update_budgeted`]
+    /// leaves the state half folded and its retained figure stale. The runtime stops the execution
+    /// and drops the aggregator on that error, which is why no rollback is kept; a caller holding
+    /// one must drop it rather than fold into it again.
     pub fn update(&mut self, series: &Series) -> Result<()> {
+        self.update_budgeted(series, |_| Ok(()))
+    }
+
+    pub(crate) fn update_budgeted(
+        &mut self,
+        series: &Series,
+        mut budget: impl FnMut(usize) -> Result<()>,
+    ) -> Result<()> {
         validate_float_series(&self.schema, series)?;
+        let existing_bytes = self.state_bytes();
+        let grouping_workspace = grouping_workspace_bytes(&series.metric);
+        let mut workspace_bytes =
+            grouping_workspace.saturating_add(pending_bytes(self.schema.fields.len()));
+        budget(existing_bytes.saturating_add(workspace_bytes))?;
+        workspace_bytes = workspace_bytes.saturating_add(
+            self.labels
+                .len()
+                .saturating_mul(std::mem::size_of::<&str>()),
+        );
+        budget(existing_bytes.saturating_add(workspace_bytes))?;
         let labels: Vec<&str> = self.labels.iter().map(|label| &**label).collect();
         let grouping = if self.by {
             Grouping::By(&labels)
@@ -130,39 +187,210 @@ impl FloatAggregator {
             Grouping::Without(&labels)
         };
         let metric = grouping.apply(&series.metric)?;
-        let index = match self.group_index.entry(label_key(&metric)) {
+        let key_bytes = metric
+            .tags()
+            .len()
+            .saturating_add(1)
+            .saturating_mul(std::mem::size_of::<(Arc<str>, Arc<str>)>());
+        workspace_bytes = workspace_bytes
+            .saturating_add(grouping_workspace_bytes(&metric))
+            .saturating_add(key_bytes);
+        budget(existing_bytes.saturating_add(workspace_bytes))?;
+        let key = label_key(&metric);
+        let key_retained = key
+            .capacity()
+            .saturating_mul(size_of::<(Arc<str>, Arc<str>)>());
+        let new_group = !self.group_index.contains_key(&key);
+        let group_capacity = self.groups.capacity();
+        let map_capacity = self.group_index.capacity();
+        let old_map_backing = hash_map_backing_bytes(map_capacity);
+        if new_group {
+            let next_group_capacity = if self.groups.len() == group_capacity {
+                group_capacity.saturating_mul(2).max(4)
+            } else {
+                group_capacity
+            };
+            let group_realloc_bytes = if next_group_capacity > group_capacity {
+                next_group_capacity.saturating_mul(size_of::<Group>())
+            } else {
+                0
+            };
+            budget(
+                existing_bytes
+                    .saturating_add(metric_bytes(&metric))
+                    .saturating_add(key_bytes)
+                    // A realloc that cannot grow in place keeps both buffers alive while it
+                    // copies, so a growing step admits the old buffer beside the new one --
+                    // the same reason the map below admits both tables. The old buffer alone
+                    // is already in `existing_bytes`; what is added here is the new one.
+                    .saturating_add(group_realloc_bytes)
+                    .saturating_add(if self.group_index.len() == map_capacity {
+                        hash_map_backing_bytes(next_hash_capacity(map_capacity))
+                    } else {
+                        0
+                    })
+                    .saturating_add(workspace_bytes),
+            )?;
+        }
+        let group_capacity_growth = if new_group && self.groups.len() == group_capacity {
+            group_capacity.saturating_mul(2).max(4)
+        } else {
+            group_capacity
+        };
+        if new_group && group_capacity_growth > group_capacity {
+            self.groups.reserve(1);
+        }
+        if new_group && self.group_index.len() == self.group_index.capacity() {
+            self.group_index.reserve(1);
+        }
+        let index = match self.group_index.entry(key) {
             Entry::Occupied(entry) => *entry.get(),
             Entry::Vacant(entry) => {
                 self.groups.push(Group {
-                    metric,
+                    metric: metric.clone(),
                     blocks: BTreeMap::new(),
                 });
-                *entry.insert(self.groups.len() - 1)
+                let index = self.groups.len() - 1;
+                entry.insert(index);
+                index
             }
         };
-        self.groups[index].accumulate(series, self.op);
+        if new_group {
+            self.retained = self
+                .retained
+                .saturating_add(metric_bytes(&metric))
+                .saturating_add(key_retained)
+                .saturating_add(
+                    self.groups
+                        .capacity()
+                        .saturating_sub(group_capacity)
+                        .saturating_mul(size_of::<Group>()),
+                )
+                .saturating_add(
+                    hash_map_backing_bytes(self.group_index.capacity())
+                        .saturating_sub(old_map_backing),
+                );
+        }
+        drop(labels);
+        drop(metric);
+        self.retained = self.groups[index].accumulate(
+            series,
+            self.op,
+            &mut budget,
+            self.retained,
+            self.schema.fields.len(),
+        )?;
         Ok(())
     }
 
+    /// Number of groups currently retained by this fold.
+    pub(crate) fn groups(&self) -> usize {
+        self.groups.len()
+    }
+
+    /// Bytes reserved for this fold, using bounded allowances for its maps and group payloads.
+    pub(crate) fn state_bytes(&self) -> usize {
+        self.retained
+    }
+
+    #[cfg(test)]
+    pub(crate) fn recounted_state_bytes(&self) -> usize {
+        let groups = self.groups.iter().fold(0usize, |total, group| {
+            total
+                .saturating_add(metric_bytes(&group.metric))
+                .saturating_add(
+                    group
+                        .blocks
+                        .len()
+                        .saturating_mul(BLOCK_POINTS)
+                        .saturating_mul(self.schema.fields.len())
+                        .saturating_mul(std::mem::size_of::<FloatState>()),
+                )
+                .saturating_add(btree_backing_bytes(group.blocks.len()))
+        });
+        groups
+            .saturating_add(
+                self.labels
+                    .capacity()
+                    .saturating_mul(std::mem::size_of::<Arc<str>>()),
+            )
+            .saturating_add(hash_map_backing_bytes(self.group_index.capacity()))
+            .saturating_add(self.group_index.keys().fold(0usize, |bytes, key| {
+                bytes.saturating_add(
+                    key.capacity()
+                        .saturating_mul(size_of::<(Arc<str>, Arc<str>)>()),
+                )
+            }))
+            .saturating_add(
+                self.labels
+                    .iter()
+                    .map(|label| label.len() + 2 * std::mem::size_of::<usize>())
+                    .sum::<usize>(),
+            )
+            .saturating_add(
+                self.groups
+                    .capacity()
+                    .saturating_mul(std::mem::size_of::<Group>()),
+            )
+    }
+
     /// The result of every group, in the order the groups first appeared.
     pub fn finish(self) -> Result<Vec<Series>> {
-        finish_groups(self.schema, self.op, self.groups)
+        self.finish_budgeted(|_| Ok(()))
+    }
+
+    pub(crate) fn finish_budgeted(
+        self,
+        budget: impl FnMut(usize) -> Result<()>,
+    ) -> Result<Vec<Series>> {
+        let state = self.state_bytes();
+        finish_groups_budgeted(self.schema, self.op, self.groups, state, budget)
     }
 }
 
-/// Turns the state of `groups` into result series, in group order.
-fn finish_groups(
+/// Turns the state of `groups` into result series, in group order, admitting as it builds.
+fn finish_groups_budgeted(
     schema: Arc<SeriesSchema>,
     op: AggregateOp,
     groups: Vec<Group>,
+    state: usize,
+    mut budget: impl FnMut(usize) -> Result<()>,
 ) -> Result<Vec<Series>> {
     let width = schema.fields.len();
+    let group_count = groups.len();
+    if group_count == 0 {
+        return Ok(Vec::new());
+    }
+    let groups = groups.into_iter();
     let mut builder = GridSeriesBuilder::try_new(schema)?;
-    let mut output = Vec::with_capacity(groups.len());
-    let mut row = Vec::with_capacity(width);
+    let output_capacity_bytes = group_count.saturating_mul(std::mem::size_of::<Series>());
+    let row_capacity_bytes = width.saturating_mul(std::mem::size_of::<Option<f64>>());
+    let base = state
+        .saturating_add(output_capacity_bytes)
+        .saturating_add(row_capacity_bytes);
+    budget(base)?;
+    let mut output = Vec::new();
+    output.try_reserve_exact(group_count).map_err(|error| {
+        DataFusionError::Execution(format!("cannot reserve aggregate outputs: {error}"))
+    })?;
+    let mut row = Vec::new();
+    row.try_reserve_exact(width).map_err(|error| {
+        DataFusionError::Execution(format!("cannot reserve aggregate row: {error}"))
+    })?;
+    let mut completed_output_bytes = 0usize;
     for group in groups {
         // One builder serves every group, so each result series starts explicitly.
         builder.begin_series();
+        let (rows, runs) = group_rows_runs(&group, width);
+        let finish_bytes = finish_working_bytes(rows, runs, width);
+        budget(
+            state
+                .saturating_add(output_capacity_bytes)
+                .saturating_add(row_capacity_bytes)
+                .saturating_add(completed_output_bytes)
+                .saturating_add(finish_bytes),
+        )?;
+        builder.reserve_rows(rows)?;
         for (&block, states) in &group.blocks {
             // The last block can extend past the grid; only points with a sample are pushed.
             for (offset, point) in states.chunks_exact(width).enumerate() {
@@ -174,11 +402,89 @@ fn finish_groups(
                 builder.push(block * BLOCK_POINTS + offset, &row)?;
             }
         }
-        output.extend(builder.finish(group.metric)?);
+        let finished = builder.finish(group.metric)?;
+        completed_output_bytes = completed_output_bytes.saturating_add(
+            finished
+                .iter()
+                .map(crate::grid_series::series_bytes)
+                .sum::<usize>(),
+        );
+        output.extend(finished);
+        budget(
+            state
+                .saturating_add(output_capacity_bytes)
+                .saturating_add(row_capacity_bytes)
+                .saturating_add(completed_output_bytes),
+        )?;
     }
     Ok(output)
 }
 
+/// The result of every group, without a budget: the entry the partial-state merger takes, which
+/// does not hold a query pool to admit against.
+fn finish_groups(
+    schema: Arc<SeriesSchema>,
+    op: AggregateOp,
+    groups: Vec<Group>,
+) -> Result<Vec<Series>> {
+    finish_groups_budgeted(schema, op, groups, 0, |_| Ok(()))
+}
+
+fn group_rows_runs(group: &Group, width: usize) -> (usize, usize) {
+    if width == 0 {
+        return (0, 0);
+    }
+    let mut rows = 0usize;
+    let mut runs = 0usize;
+    let mut previous: Option<usize> = None;
+    for (&block, states) in &group.blocks {
+        for (offset, point) in states.chunks_exact(width).enumerate() {
+            if point.iter().any(|state| state.seen) {
+                let index = block.saturating_mul(BLOCK_POINTS).saturating_add(offset);
+                rows = rows.saturating_add(1);
+                if previous.is_none_or(|previous| previous.checked_add(1) != Some(index)) {
+                    runs = runs.saturating_add(1);
+                }
+                previous = Some(index);
+            }
+        }
+    }
+    (rows, runs)
+}
+
+fn finish_working_bytes(rows: usize, runs: usize, width: usize) -> usize {
+    let values = rows
+        .saturating_mul(width)
+        .saturating_mul(std::mem::size_of::<f64>());
+    let validity_bytes = rows
+        .saturating_add(7)
+        .checked_div(8)
+        .unwrap_or(0)
+        .div_ceil(64)
+        .saturating_mul(64);
+    let fields = crate::grid_series::field_builder_bytes(width).saturating_add(
+        width.saturating_mul(
+            std::mem::size_of::<Float64Array>()
+                .saturating_add(std::mem::size_of::<Samples>())
+                .saturating_add(std::mem::size_of::<usize>()),
+        ),
+    );
+    let indices = rows.saturating_mul(std::mem::size_of::<usize>());
+    let spans = runs.saturating_mul(std::mem::size_of::<GridSpan>());
+    let axis_conversion = runs
+        .saturating_mul(
+            2 * std::mem::size_of::<GridSpan>() + std::mem::size_of::<Arc<[GridSpan]>>(),
+        )
+        .saturating_add(rows.saturating_mul(std::mem::size_of::<usize>()));
+    values
+        .saturating_add(width.saturating_mul(validity_bytes))
+        .saturating_add(fields)
+        .saturating_add(indices)
+        .saturating_add(spans)
+        .saturating_add(axis_conversion)
+        .saturating_add(std::mem::size_of::<Series>() + std::mem::size_of::<TimeAxis>())
+}
+
 /// `quantile(q, v)`, with `q` read at each grid point from an unlabelled single-field scalar on
 /// the same grid, as Prometheus evaluates the parameter at every step. A point where `q` has no
 /// value has no result; `q` outside `[0, 1]` or NaN gives what Prometheus' `quantile` gives.
@@ -438,6 +744,105 @@ impl Grouping<'_> {
 
 type LabelKey = Vec<(Arc<str>, Arc<str>)>;
 
+fn metric_bytes(metric: &MetricName) -> usize {
+    metric
+        .name()
+        .map_or(0, str::len)
+        .saturating_add(
+            metric
+                .tags()
+                .iter()
+                .map(|tag| tag.name.len() + tag.value.len())
+                .sum::<usize>(),
+        )
+        .saturating_add(
+            metric
+                .tags()
+                .len()
+                .saturating_mul(std::mem::size_of::<(Arc<str>, Arc<str>)>()),
+        )
+        .saturating_add(std::mem::size_of::<MetricName>())
+}
+
+fn grouping_workspace_bytes(metric: &MetricName) -> usize {
+    metric_bytes(metric)
+        .saturating_add(
+            metric
+                .tags()
+                .len()
+                .saturating_add(1)
+                .saturating_mul(std::mem::size_of::<(Arc<str>, Arc<str>)>()),
+        )
+        .saturating_add(std::mem::size_of::<LabelKey>())
+}
+
+/// Conservative backing estimate for Rust 1.98.1's insert-only `HashMap` (hashbrown 0.17.1).
+/// Bucket count follows its 7/8 load factor; entries, one control byte per bucket and the
+/// 16-byte control-group tail are counted. This models requested table storage, not allocator slack.
+/// An upper bound on the bytes the group index holds for `capacity` entries.
+///
+/// The table keeps its entries in a power-of-two number of buckets below `16/7` of the usable
+/// capacity, with one control byte beside each bucket. Three buckets per entry therefore covers the
+/// table whatever exact growth the map uses, which is what keeps this a bound rather than a model of
+/// one implementation: a toolchain that changes how the map grows cannot make it under-count.
+fn hash_map_backing_bytes(capacity: usize) -> usize {
+    capacity
+        .saturating_mul(3)
+        .saturating_mul(size_of::<(LabelKey, usize)>() + 1)
+}
+
+/// An upper bound on the capacity a map holding `capacity` entries grows to.
+fn next_hash_capacity(capacity: usize) -> usize {
+    capacity.saturating_mul(2).max(7)
+}
+
+/// An upper bound on the bytes one B-tree node holds.
+///
+/// A node keeps at most eleven key/value pairs and twelve child pointers, plus a header of a few
+/// words; the leaf form is smaller than that. The header allowance is generous on purpose, so the
+/// bound does not depend on how a toolchain lays the node out.
+fn btree_node_bytes() -> usize {
+    11 * size_of::<(usize, Box<[FloatState]>)>() + 16 * size_of::<usize>()
+}
+
+/// An upper bound on the bytes a B-tree of `entries` holds.
+///
+/// Every node but the root keeps at least five entries, so `entries / 5` nodes is enough to hold
+/// them; a tree of one entry is one node.
+fn btree_backing_bytes(entries: usize) -> usize {
+    if entries == 0 {
+        return 0;
+    }
+    entries.div_ceil(5).saturating_mul(btree_node_bytes())
+}
+
+fn pending_bytes(fields: usize) -> usize {
+    working_set_bytes(fields, BLOCK_POINTS)
+}
+
+fn working_set_bytes(fields: usize, pending_capacity: usize) -> usize {
+    pending_capacity
+        .saturating_mul(std::mem::size_of::<(usize, f64)>())
+        .saturating_add(fields.saturating_mul(std::mem::size_of::<Option<f64>>() + 1))
+}
+
+fn labels_capacity_bytes(grouping: Grouping<'_>) -> usize {
+    let labels = match grouping {
+        Grouping::By(labels) | Grouping::Without(labels) => labels,
+    };
+    labels.len().saturating_mul(std::mem::size_of::<Arc<str>>())
+}
+
+fn grouping_bytes(grouping: Grouping<'_>) -> usize {
+    let labels = match grouping {
+        Grouping::By(labels) | Grouping::Without(labels) => labels,
+    };
+    labels
+        .iter()
+        .map(|label| label.len() + std::mem::size_of::<Arc<str>>())
+        .sum::<usize>()
+}
+
 fn label_key(metric: &MetricName) -> LabelKey {
     let mut labels = Vec::with_capacity(metric.tags().len() + 1);
     if let Some(name) = metric.name() {
@@ -463,9 +868,29 @@ struct Group {
     blocks: BTreeMap<usize, Box<[FloatState]>>,
 }
 
+/// Which field of which layout a fold is writing, and how it folds it.
+///
+/// One value carried through the per-block work instead of three separate arguments: the three
+/// never vary independently within one series, and the block loop is the only place that reads
+/// them.
+#[derive(Clone, Copy)]
+struct FieldFold {
+    width: usize,
+    field_index: usize,
+    op: AggregateOp,
+}
+
 impl Group {
-    fn accumulate(&mut self, series: &Series, op: AggregateOp) {
-        let width = series.fields.len();
+    fn accumulate(
+        &mut self,
+        series: &Series,
+        op: AggregateOp,
+        budget: &mut impl FnMut(usize) -> Result<()>,
+        mut retained: usize,
+        width: usize,
+    ) -> Result<usize> {
+        let pending_capacity_bytes = BLOCK_POINTS.saturating_mul(size_of::<(usize, f64)>());
+        budget(retained.saturating_add(pending_capacity_bytes))?;
         let mut pending = Vec::with_capacity(BLOCK_POINTS);
         for (field_index, field) in series.fields.iter().enumerate() {
             // The axis says which absolute positions the field holds and the field iterator
@@ -475,35 +900,81 @@ impl Group {
             let mut block = usize::MAX;
             for ((index, _), sample) in series.time.positions().zip(field.iter()) {
                 if index / BLOCK_POINTS != block {
-                    self.update_block(block, width, field_index, op, &mut pending);
+                    retained = self.update_block(
+                        block,
+                        FieldFold {
+                            width,
+                            field_index,
+                            op,
+                        },
+                        &mut pending,
+                        budget,
+                        retained,
+                    )?;
                     block = index / BLOCK_POINTS;
                 }
                 if let SampleRef::Float(sample) = sample {
                     pending.push((index % BLOCK_POINTS, sample));
                 }
             }
-            self.update_block(block, width, field_index, op, &mut pending);
+            retained = self.update_block(
+                block,
+                FieldFold {
+                    width,
+                    field_index,
+                    op,
+                },
+                &mut pending,
+                budget,
+                retained,
+            )?;
         }
+        Ok(retained)
     }
 
     /// Applies one field's samples within a block, so a block costs one lookup per field run.
     fn update_block(
         &mut self,
         block: usize,
-        width: usize,
-        field_index: usize,
-        op: AggregateOp,
+        fold: FieldFold,
         pending: &mut Vec<(usize, f64)>,
-    ) {
+        budget: &mut impl FnMut(usize) -> Result<()>,
+        retained: usize,
+    ) -> Result<usize> {
         if pending.is_empty() {
-            return;
+            return Ok(retained);
+        }
+        let new_block = !self.blocks.contains_key(&block);
+        let block_bytes = if new_block {
+            BLOCK_POINTS
+                .saturating_mul(fold.width)
+                .saturating_mul(std::mem::size_of::<FloatState>())
+        } else {
+            0
+        };
+        let backing_delta = if new_block {
+            btree_backing_bytes(self.blocks.len() + 1)
+                .saturating_sub(btree_backing_bytes(self.blocks.len()))
+        } else {
+            0
+        };
+        if new_block {
+            budget(
+                retained
+                    .saturating_add(backing_delta)
+                    .saturating_add(block_bytes)
+                    .saturating_add(pending.capacity().saturating_mul(size_of::<(usize, f64)>())),
+            )?;
         }
         let states = self.blocks.entry(block).or_insert_with(|| {
-            vec![FloatState::default(); BLOCK_POINTS * width].into_boxed_slice()
+            vec![FloatState::default(); BLOCK_POINTS * fold.width].into_boxed_slice()
         });
         for (offset, sample) in pending.drain(..) {
-            states[offset * width + field_index].update(op, sample);
+            states[offset * fold.width + fold.field_index].update(fold.op, sample);
         }
+        Ok(retained
+            .saturating_add(backing_delta)
+            .saturating_add(block_bytes))
     }
 }
 

src/promql/src/aggregate/tests.rs

@@ -14,6 +14,7 @@
 
 use std::collections::BTreeMap;
 
+use datafusion::error::DataFusionError;
 use datatypes::arrow::array::{Float64Array, UInt32Array};
 use datatypes::series::Label;
 use datatypes::series::histogram::HistogramBatchBuilder;
@@ -345,6 +346,27 @@ fn avg_compensation_matches_prometheus_bits() {
     }
 }
 
+#[test]
+fn one_field_missing_and_another_nan_keep_row_nullability() {
+    let schema = schema(&["missing", "value"]);
+    let series = input(
+        &schema,
+        metric("m", &[]),
+        0,
+        &[
+            vec![None, None, None],
+            vec![Some(1.0), Some(f64::NAN), Some(2.0)],
+        ],
+    );
+    let output = aggregate_float(&schema, &[series], AggregateOp::Sum, Grouping::By(&[])).unwrap();
+    assert_eq!(output.len(), 1);
+    assert_eq!(values(&output[0], 0).1, vec![None, None, None]);
+    let values = values(&output[0], 1).1;
+    assert_eq!(values[0], Some(1.0));
+    assert!(values[1].unwrap().is_nan());
+    assert_eq!(values[2], Some(2.0));
+}
+
 #[test]
 fn fields_are_aggregated_independently() {
     // A field without input values at a row is null there, even for count and group; the row
@@ -476,6 +498,301 @@ fn fields_are_aggregated_independently() {
     }
 }
 
+#[test]
+fn incremental_retained_bytes_match_full_recount_after_updates() {
+    let schema = schema_with_grid_len(&["value", "other"], 256);
+    let mut aggregator =
+        FloatAggregator::try_new(&schema, AggregateOp::Sum, Grouping::By(&["zone", "rack"]))
+            .unwrap();
+    assert_eq!(
+        aggregator.state_bytes(),
+        aggregator.recounted_state_bytes(),
+        "initial state"
+    );
+    let updates = [
+        input(
+            &schema,
+            metric("m", &[("zone", "a"), ("rack", "1")]),
+            0,
+            &[vec![Some(1.0)], vec![Some(2.0)]],
+        ),
+        input(
+            &schema,
+            metric("m", &[("zone", "a"), ("rack", "1")]),
+            64,
+            &[vec![Some(3.0)], vec![Some(4.0)]],
+        ),
+        input(
+            &schema,
+            metric("m", &[("zone", "a"), ("rack", "1")]),
+            0,
+            &[vec![Some(5.0)], vec![Some(6.0)]],
+        ),
+        input(
+            &schema,
+            metric("m", &[("zone", "b"), ("rack", "2")]),
+            128,
+            &[vec![Some(7.0)], vec![Some(8.0)]],
+        ),
+    ];
+    for (index, series) in updates.iter().enumerate() {
+        let previous = aggregator.state_bytes();
+        aggregator.update(series).unwrap();
+        assert_eq!(
+            aggregator.state_bytes(),
+            aggregator.recounted_state_bytes(),
+            "update {index}"
+        );
+        if index == 2 {
+            assert_eq!(aggregator.state_bytes(), previous);
+        }
+    }
+}
+
+#[test]
+fn incremental_retained_bytes_match_full_recount_for_many_groups() {
+    let schema = schema(&["value"]);
+    let mut aggregator =
+        FloatAggregator::try_new(&schema, AggregateOp::Sum, Grouping::By(&["id"])).unwrap();
+    for id in 0..512 {
+        let id = id.to_string();
+        let series = input(
+            &schema,
+            metric("m", &[("id", &id)]),
+            id.parse::<usize>().unwrap() % 64,
+            &[vec![Some(1.0)]],
+        );
+        aggregator.update(&series).unwrap();
+        assert_eq!(
+            aggregator.state_bytes(),
+            aggregator.recounted_state_bytes(),
+            "group {id}"
+        );
+    }
+}
+
+/// The unbudgeted entry point is the budgeted one with a no-op budget, so this compares one path
+/// with itself: it is a smoke check that the wrapper still runs, not evidence that the kernels are
+/// right. The numbers are checked against Prometheus in `aggregates_match_prometheus` and
+/// `stat_aggregates_match_prometheus`.
+#[test]
+fn every_supported_kind_runs_through_the_unbudgeted_entry_point() {
+    let schema = schema(&["value"]);
+    let series = input(&schema, metric("m", &[]), 0, &[vec![Some(1.0), Some(3.0)]]);
+    for op in [
+        AggregateOp::Sum,
+        AggregateOp::Count,
+        AggregateOp::Avg,
+        AggregateOp::Min,
+        AggregateOp::Max,
+        AggregateOp::Group,
+        AggregateOp::Stddev,
+        AggregateOp::Stdvar,
+    ] {
+        let mut plain = FloatAggregator::try_new(&schema, op, Grouping::By(&[])).unwrap();
+        plain.update(&series).unwrap();
+        let plain = plain.finish().unwrap();
+        let mut budgeted =
+            FloatAggregator::try_new_budgeted(&schema, op, Grouping::By(&[]), |_| Ok(())).unwrap();
+        budgeted.update_budgeted(&series, |_| Ok(())).unwrap();
+        let budgeted = budgeted.finish_budgeted(|_| Ok(())).unwrap();
+        assert_eq!(
+            values_of(&plain, 0, 128),
+            values_of(&budgeted, 0, 128),
+            "{op:?}"
+        );
+    }
+}
+
+#[test]
+fn repeated_updates_in_one_group_and_block_keep_one_block_charge() {
+    let schema = schema(&["value"]);
+    let first = input(&schema, metric("m", &[]), 0, &[vec![Some(1.0)]]);
+    let second = input(&schema, metric("m", &[]), 1, &[vec![Some(3.0)]]);
+    let mut aggregator =
+        FloatAggregator::try_new(&schema, AggregateOp::Sum, Grouping::By(&[])).unwrap();
+    aggregator.update(&first).unwrap();
+    let retained = aggregator.state_bytes();
+    aggregator.update(&second).unwrap();
+    assert_eq!(aggregator.state_bytes(), retained);
+}
+
+#[test]
+fn adjacent_blocks_are_admitted_and_accounted_separately() {
+    let schema = schema(&["value"]);
+    let series = input(&schema, metric("m", &[]), 63, &[vec![Some(1.0), Some(2.0)]]);
+    let mut aggregator =
+        FloatAggregator::try_new(&schema, AggregateOp::Sum, Grouping::By(&[])).unwrap();
+    let mut admissions = Vec::new();
+    aggregator
+        .update_budgeted(&series, |bytes| {
+            admissions.push(bytes);
+            Ok(())
+        })
+        .unwrap();
+    assert_eq!(aggregator.groups(), 1);
+    assert_eq!(aggregator.groups[0].blocks.len(), 2);
+    assert_eq!(
+        admissions.last().copied(),
+        Some(
+            aggregator
+                .state_bytes()
+                .saturating_add(std::mem::size_of::<(usize, f64)>() * BLOCK_POINTS,)
+        )
+    );
+}
+
+#[test]
+fn a_rejected_block_allocation_does_not_insert_a_block() {
+    let schema = schema(&["value"]);
+    let series = input(&schema, metric("m", &[]), 0, &[vec![Some(1.0)]]);
+    // The figure a new block's admission adds: the block's states and its pending buffer, on
+    // top of whatever the fold already holds. Located by the value, not by which call reaches it,
+    // so reordering the earlier admissions cannot make this test pass without testing anything.
+    let block_states = BLOCK_POINTS
+        .saturating_mul(schema.fields.len())
+        .saturating_mul(std::mem::size_of::<FloatState>());
+    let pending = BLOCK_POINTS * std::mem::size_of::<(usize, f64)>();
+    let mut aggregator =
+        FloatAggregator::try_new(&schema, AggregateOp::Sum, Grouping::By(&[])).unwrap();
+    let mut refused = None;
+    let result = aggregator.update_budgeted(&series, |bytes| {
+        // Everything up to the block's own admission is admitted; only the block is refused.
+        if bytes >= block_states.saturating_add(pending) {
+            refused = Some(bytes);
+            return Err(DataFusionError::Execution("block allocation budget".into()));
+        }
+        Ok(())
+    });
+    assert!(
+        result.is_err(),
+        "the block was admitted with no budget left"
+    );
+    let refused = refused.expect("the refusal did not come from a block admission");
+    assert!(
+        refused >= block_states.saturating_add(pending),
+        "the refusal came from an admission smaller than a block"
+    );
+    assert_eq!(aggregator.groups(), 1);
+    assert!(
+        aggregator.groups[0].blocks.is_empty(),
+        "a refused block was inserted anyway"
+    );
+}
+
+#[test]
+fn a_map_growth_rejection_does_not_insert_the_new_group() {
+    let schema = schema(&["value"]);
+    let mut aggregator =
+        FloatAggregator::try_new(&schema, AggregateOp::Sum, Grouping::By(&["i"])).unwrap();
+    let first = input(&schema, metric("m", &[("i", "0")]), 0, &[vec![Some(1.0)]]);
+    aggregator.update(&first).unwrap();
+    let mut index = 1;
+    while aggregator.group_index.len() < aggregator.group_index.capacity() {
+        let label = index.to_string();
+        let series = input(
+            &schema,
+            metric("m", &[("i", &label)]),
+            0,
+            &[vec![Some(1.0)]],
+        );
+        aggregator.update(&series).unwrap();
+        index += 1;
+    }
+    let group_count = aggregator.groups();
+    let old_state = aggregator.state_bytes();
+    assert_eq!(aggregator.group_index.capacity(), group_count);
+
+    // A resize keeps the old table while it fills the new one, so the figure the growth has to
+    // admit is the whole new table on top of the old state -- not the difference between them.
+    // That is the property asserted here: the refusal threshold is the state plus a whole new
+    // table, so the test fails if the admission ever drops to the delta.
+    let new_table = hash_map_backing_bytes(next_hash_capacity(group_count));
+    let threshold = old_state.saturating_add(new_table);
+    let label = index.to_string();
+    let series = input(
+        &schema,
+        metric("m", &[("i", &label)]),
+        0,
+        &[vec![Some(1.0)]],
+    );
+    let mut peak = 0;
+    let result = aggregator.update_budgeted(&series, |bytes| {
+        peak = peak.max(bytes);
+        if bytes >= threshold {
+            Err(DataFusionError::Execution("old+new map budget".into()))
+        } else {
+            Ok(())
+        }
+    });
+    assert!(
+        result.is_err(),
+        "the map grew with no budget for both tables"
+    );
+    assert!(
+        peak >= threshold,
+        "the growth admitted {peak}, less than the old state plus a new table ({threshold})"
+    );
+    assert_eq!(aggregator.group_index.capacity(), group_count);
+    assert_eq!(aggregator.groups(), group_count);
+    assert_eq!(
+        aggregator
+            .groups
+            .iter()
+            .map(|group| group.blocks.len())
+            .sum::<usize>(),
+        group_count
+    );
+}
+
+#[test]
+fn finish_rejection_after_first_group_is_budgeted_before_the_next() {
+    let schema = schema(&["value"]);
+    let mut aggregator =
+        FloatAggregator::try_new(&schema, AggregateOp::Sum, Grouping::By(&["i"])).unwrap();
+    for (label, value) in [("a", 1.0), ("b", 2.0)] {
+        let series = input(
+            &schema,
+            metric("m", &[("i", label)]),
+            0,
+            &[vec![Some(value)]],
+        );
+        aggregator.update(&series).unwrap();
+    }
+    let mut calls = Vec::new();
+    let result = aggregator.finish_budgeted(|bytes| {
+        calls.push(bytes);
+        if calls.len() == 4 {
+            Err(DataFusionError::Execution("budget".into()))
+        } else {
+            Ok(())
+        }
+    });
+    assert!(result.is_err());
+    assert_eq!(calls.len(), 4);
+    assert!(calls[1] > calls[0], "first builder peak");
+    assert!(calls[2] < calls[1], "first group settle");
+    assert!(calls[3] > calls[2], "second builder peak");
+
+    // The base already includes the output Vec backing; later callbacks add only the
+    // builder workspace or completed Series allocation.
+    let builder_workspace = finish_working_bytes(1, 1, 1);
+    let expected_output = aggregate_float(
+        &schema,
+        &[
+            input(&schema, metric("m", &[("i", "a")]), 0, &[vec![Some(1.0)]]),
+            input(&schema, metric("m", &[("i", "b")]), 0, &[vec![Some(2.0)]]),
+        ],
+        AggregateOp::Sum,
+        Grouping::By(&["i"]),
+    )
+    .unwrap();
+    let completed_series = crate::grid_series::series_bytes(&expected_output[0]);
+    assert_eq!(calls[1] - calls[0], builder_workspace);
+    assert_eq!(calls[2] - calls[0], completed_series);
+    assert_eq!(calls[3] - calls[2], builder_workspace);
+}
+
 #[test]
 fn state_blocks_keep_absolute_indices_across_boundaries_and_gaps() {
     // A 512-point grid spans eight state blocks; inputs straddle a block edge, leave whole

src/promql/src/extension_plan.rs

@@ -14,6 +14,7 @@
 
 mod absent;
 mod empty_metric;
+mod fused_series;
 mod histogram_fold;
 mod instant_manipulate;
 mod normalize;
@@ -37,6 +38,7 @@ use datafusion::error::{DataFusionError, Result as DataFusionResult};
 use datafusion::logical_expr::{Expr, Extension, LogicalPlan};
 use datatypes::data_type::DataType as _;
 pub use empty_metric::{EmptyMetric, EmptyMetricExec, EmptyMetricStream, build_special_time_expr};
+pub use fused_series::{FusedSeries, FusedSeriesBinding, FusedSeriesExec, FusedSeriesStream};
 pub use histogram_fold::{
     HistogramFold, HistogramFoldExec, HistogramFoldOperation, HistogramFoldStream,
 };

src/promql/src/grid_series.rs

@@ -19,12 +19,12 @@ use std::collections::hash_map::Entry;
 use std::sync::Arc;
 
 use datafusion::arrow::array::builder::NullBufferBuilder;
-use datafusion::arrow::array::{Float64Array, UInt32Array};
+use datafusion::arrow::array::{Array, Float64Array, UInt32Array};
 use datafusion::error::{DataFusionError, Result};
 use datatypes::series::MetricName;
-use datatypes::series::histogram::HistogramBatchBuilder;
+use datatypes::series::histogram::{HistogramBatch, HistogramBatchBuilder};
 use datatypes::series::samples::{
-    HistogramSamples, SampleRef, Samples, Series, SeriesFields, SeriesSchema,
+    DenseSamplesRef, HistogramSamples, SampleRef, Samples, Series, SeriesFields, SeriesSchema,
 };
 use datatypes::series::time::{EvalGrid, GridSpan, TimeAxis};
 
@@ -35,6 +35,70 @@ pub(crate) mod reference;
 #[cfg(test)]
 mod tests;
 
+pub(crate) fn series_bytes(series: &Series) -> usize {
+    let runs = series.time.runs().count();
+    let segments: usize = series
+        .fields
+        .iter()
+        .map(|field| field.segments().len())
+        .sum();
+    std::mem::size_of::<Series>()
+        .saturating_add(std::mem::size_of::<TimeAxis>())
+        .saturating_add(runs.saturating_mul(std::mem::size_of::<(usize, usize)>()))
+        .saturating_add(
+            segments.saturating_mul(std::mem::size_of::<Samples>() + std::mem::size_of::<usize>()),
+        )
+        .saturating_add(fields_bytes(&series.fields))
+        .saturating_add(series.metric.name().map_or(0, str::len))
+        .saturating_add(std::mem::size_of_val(series.metric.tags()))
+        .saturating_add(
+            series
+                .metric
+                .tags()
+                .iter()
+                .map(|tag| tag.name.len() + tag.value.len())
+                .sum::<usize>(),
+        )
+}
+
+pub(crate) fn fields_bytes(fields: &SeriesFields) -> usize {
+    fields
+        .iter()
+        .map(|field| {
+            field.segments().iter().fold(0usize, |total, segment| {
+                total.saturating_add(segment_bytes(segment))
+            })
+        })
+        .fold(0usize, usize::saturating_add)
+}
+
+/// The bytes one dense segment of samples holds.
+pub(crate) fn segment_bytes(segment: &Samples) -> usize {
+    match segment.dense() {
+        Some(DenseSamplesRef::Float(values)) => values.get_array_memory_size(),
+        Some(DenseSamplesRef::Histogram(histograms)) => histograms
+            .indices()
+            .get_array_memory_size()
+            .saturating_add(table_bytes(histograms.batch())),
+        Some(DenseSamplesRef::Mixed(mixed)) => mixed
+            .floats()
+            .get_array_memory_size()
+            .saturating_add(mixed.histograms().indices().get_array_memory_size())
+            .saturating_add(table_bytes(mixed.histograms().batch())),
+        // `segments` hands out dense segments only, so a nested chunk never reaches here.
+        None => 0,
+    }
+}
+
+/// The bytes one histogram table holds: the batch struct and the arrays its columns view.
+pub(crate) fn table_bytes(table: &Arc<HistogramBatch>) -> usize {
+    std::mem::size_of::<HistogramBatch>().saturating_add(table.as_arrow().get_array_memory_size())
+}
+
+pub(crate) fn field_builder_bytes(width: usize) -> usize {
+    width.saturating_mul(std::mem::size_of::<FieldBuilder>())
+}
+
 /// Builds one grid series from rows pushed at absolute grid indices.
 ///
 /// A push names one position and the value each field holds there. Rows are accumulated
@@ -114,6 +178,36 @@ impl GridSeriesBuilder {
         })
     }
 
+    /// Reserves capacity for the exact number of valued rows in the next series.
+    ///
+    /// Call after `begin_series` and before its first push. Existing public callers may continue
+    /// to rely on normal growth; callers with a budget can reserve the concrete row capacity first.
+    pub(crate) fn reserve_rows(&mut self, rows: usize) -> Result<()> {
+        self.indices
+            .try_reserve_exact(rows.saturating_sub(self.indices.len()))
+            .map_err(|error| {
+                DataFusionError::Execution(format!("cannot reserve grid row indices: {error}"))
+            })?;
+        if rows == 0 {
+            return Ok(());
+        }
+        if self.fields.is_empty() {
+            self.fields = (0..self.schema.fields.len())
+                .map(|_| FieldBuilder::new(rows))
+                .collect();
+        }
+        for field in &mut self.fields {
+            field
+                .values
+                .try_reserve_exact(rows.saturating_sub(field.values.len()))
+                .map_err(|error| {
+                    DataFusionError::Execution(format!("cannot reserve grid values: {error}"))
+                })?;
+            field.nulls = NullBufferBuilder::new(rows);
+        }
+        Ok(())
+    }
+
     /// Pushes the field values at absolute grid `index`; `None` is a missing value.
     ///
     /// Indices must be strictly increasing; skipped indices are positions that hold no value.
@@ -176,7 +270,7 @@ impl GridSeriesBuilder {
             return Ok(false);
         }
         if self.fields.is_empty() {
-            self.fields = (0..len).map(|_| FieldBuilder::new()).collect();
+            self.fields = (0..len).map(|_| FieldBuilder::new(0)).collect();
         }
         self.indices.push(index);
         Ok(true)
@@ -239,24 +333,28 @@ impl GridSeriesBuilder {
         }
         let fields = std::mem::take(&mut self.fields);
         let indices = std::mem::take(&mut self.indices);
-        let samples = fields
-            .into_iter()
-            .map(|mut field| {
-                let floats = Float64Array::new(field.values.into(), field.nulls.finish());
-                let Some(column) = field.histograms else {
-                    return Ok(Samples::from_float(floats));
-                };
+        let mut samples = Vec::new();
+        samples.try_reserve_exact(fields.len()).map_err(|error| {
+            DataFusionError::Execution(format!("cannot reserve grid field descriptions: {error}"))
+        })?;
+        for mut field in fields {
+            let floats = Float64Array::new(field.values.into(), field.nulls.finish());
+            let sample = if let Some(column) = field.histograms {
                 let batch = column.batch.finish().map_err(external)?;
                 let histograms =
                     HistogramSamples::try_new(UInt32Array::from(column.indices), batch)
                         .map_err(external)?;
-                Samples::try_from_mixed(floats, histograms).map_err(external)
-            })
-            .collect::<Result<Vec<_>>>()?;
+                Samples::try_from_mixed(floats, histograms).map_err(external)?
+            } else {
+                Samples::from_float(floats)
+            };
+            samples.push(sample);
+        }
         let fields = SeriesFields::try_from_vec(samples).map_err(external)?;
         // The runs are all the axis has to remember about the gaps, and the rows are already
         // compact, so no position holding no value is visited or stored.
-        let spans = spans_of(&indices);
+        let run_count = count_runs(&indices);
+        let spans = spans_of(&indices, run_count)?;
         let time = TimeAxis::from_grid_spans(self.grid.clone(), spans).map_err(external)?;
         Series::try_new(&self.schema, metric, time, fields)
             .map(Some)
@@ -265,8 +363,19 @@ impl GridSeriesBuilder {
 }
 
 /// Cuts the retained indices into maximal runs of consecutive positions.
-fn spans_of(indices: &[usize]) -> Vec<GridSpan> {
+fn count_runs(indices: &[usize]) -> usize {
+    indices
+        .windows(2)
+        .filter(|pair| pair[1] != pair[0] + 1)
+        .count()
+        .saturating_add(usize::from(!indices.is_empty()))
+}
+
+fn spans_of(indices: &[usize], runs: usize) -> Result<Vec<GridSpan>> {
     let mut spans: Vec<GridSpan> = Vec::new();
+    spans.try_reserve_exact(runs).map_err(|error| {
+        DataFusionError::Execution(format!("cannot reserve grid spans: {error}"))
+    })?;
     let mut start = 0;
     while let Some(&position) = indices.get(start) {
         let mut len = 1;
@@ -276,14 +385,14 @@ fn spans_of(indices: &[usize]) -> Vec<GridSpan> {
         spans.push(GridSpan::new(position, len));
         start += len;
     }
-    spans
+    Ok(spans)
 }
 
 impl FieldBuilder {
-    fn new() -> Self {
+    fn new(rows: usize) -> Self {
         Self {
             values: Vec::new(),
-            nulls: NullBufferBuilder::new(0),
+            nulls: NullBufferBuilder::new(rows),
             histograms: None,
         }
     }
@@ -454,7 +563,8 @@ pub fn merge_same_labelset(schema: &Arc<SeriesSchema>, series: Vec<Series>) -> R
         // The runs are all the axis remembers about the holes, so a sparse merge stores one span
         // per run and no row for the positions between them.
         let time =
-            TimeAxis::from_grid_spans(grid.clone(), spans_of(&positions)).map_err(external)?;
+            TimeAxis::from_grid_spans(grid.clone(), spans_of(&positions, count_runs(&positions))?)
+                .map_err(external)?;
         let fields = SeriesFields::try_from_vec(fields).map_err(external)?;
         // The first fragment names the output, as it always has: a label set ignores empty values,
         // so the metric of the first member is the one the merged series keeps.

src/promql/src/native/aggregate.rs

@@ -0,0 +1,243 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The run-time state of an aggregate node.
+//!
+//! An aggregate folds a whole vector into one series per group, so its node is the one that cannot
+//! answer as its input arrives: a group's value at a grid point is the fold of every value every
+//! member of the group has there, and the last member to arrive is the one that decides it. What the
+//! node keeps between chunks is therefore the fold's states and nothing else -- a chunk is folded as
+//! it arrives and dropped -- and it produces once, when the input ends.
+//!
+//! That is also why this runtime cannot tell an incomplete group from a complete one: it folds what
+//! it is handed. A [`AggregateStage::Complete`] aggregate is correct only if every contribution to
+//! each group lies within the streams this execution reads, which is the physical plan's job -- the
+//! caller that schedules scans and exchanges guarantees it, and a plan that ran a `Complete`
+//! aggregate once per partition and then combined the answers would be combining values that were
+//! each missing the other partitions' contributions. The stages that exist for that shape are
+//! [`AggregateStage::Partial`] and [`AggregateStage::Final`], which this runtime does not execute
+//! yet.
+
+use std::sync::Arc;
+
+use datafusion::error::{DataFusionError, Result};
+use datatypes::series::samples::{Series, SeriesSchema};
+
+use crate::aggregate::{AggregateOp, FloatAggregator, Grouping};
+use crate::evaluated::EvaluatedChunk;
+use crate::grid_series::series_bytes;
+use crate::map::validate_float_series;
+use crate::native::execute::{NodeBuffer, evaluated_bytes};
+use crate::native::metrics::NodePath;
+use crate::native::plan::{
+    AggregateKind, AggregateStage, BoundAggregate, Grouping as BoundGrouping,
+};
+
+/// One aggregate node's states, from the first chunk it is handed to the series it finishes with.
+pub(crate) struct AggregateRuntime {
+    /// What the node computes, as the tree bound it: the kind it folds by and the stage it reports.
+    op: BoundAggregate,
+    /// The layout its input and output share: the fields every chunk has and the grid they are
+    /// addressed on. A `Complete` aggregate neither adds nor drops a field, and it evaluates on its
+    /// input's grid, so one schema describes both sides.
+    schema: Arc<SeriesSchema>,
+    /// The fold. `None` once [`Self::compute`] has taken it, which is also what says the node has
+    /// nothing left to fold into.
+    accumulator: Option<FloatAggregator>,
+    /// Whether the fold has been finished and its series handed over. A node computes once: a second
+    /// pull finds the series it queued rather than a fold that ran again.
+    ///
+    /// The driver reads this to decide whether the node settled: an aggregate answers only once its
+    /// fold has run, so a child with nothing left to hand over is not an answer.
+    pub(crate) computed: bool,
+}
+
+impl AggregateRuntime {
+    /// What this node computes, for a rejection that has to say what it refused.
+    ///
+    /// It names the kind this node computes, which distinguishes it from another node in the tree.
+    fn describe(&self) -> String {
+        format!("{:?}", self.op.kind)
+    }
+
+    /// The states of a `Complete` aggregate over `schema`.
+    ///
+    /// A stage other than `Complete` and a kind without a supported kernel are refused here, before a single
+    /// chunk is folded: a node this runtime cannot compute must not start, because a partial answer
+    /// would be one the caller never asked for. The rejection names what was found, so a reader can
+    /// tell a tree this runtime cannot run from one it cannot run correctly.
+    pub(crate) fn new(
+        op: BoundAggregate,
+        schema: Arc<SeriesSchema>,
+        buffer: &mut NodeBuffer,
+    ) -> Result<Self> {
+        if op.stage != AggregateStage::Complete {
+            return Err(DataFusionError::NotImplemented(format!(
+                "the native runtime does not execute a `{:?}` aggregate yet: a `Partial` aggregate \
+                 reports the state a later stage merges and a `Final` aggregate merges that state, \
+                 and only `Complete` -- every contribution inside one execution -- is computed here",
+                op.stage
+            )));
+        }
+        let kind = match op.kind {
+            AggregateKind::Sum => AggregateOp::Sum,
+            AggregateKind::Count => AggregateOp::Count,
+            AggregateKind::Avg => AggregateOp::Avg,
+            AggregateKind::Min => AggregateOp::Min,
+            AggregateKind::Max => AggregateOp::Max,
+            AggregateKind::Group => AggregateOp::Group,
+            AggregateKind::Stddev => AggregateOp::Stddev,
+            AggregateKind::Stdvar => AggregateOp::Stdvar,
+            other => {
+                return Err(DataFusionError::NotImplemented(format!(
+                    "the native runtime does not execute the `{other:?}` aggregate yet: only the supported `Complete` aggregate kinds are computed here"
+                )));
+            }
+        };
+        let name_count = match &op.group {
+            BoundGrouping::By(labels) | BoundGrouping::Without(labels) => labels.len(),
+        };
+        let name_bytes = name_count.saturating_mul(std::mem::size_of::<&str>());
+        buffer.grow(name_bytes)?;
+        let names: Vec<&str> = match &op.group {
+            BoundGrouping::By(labels) | BoundGrouping::Without(labels) => {
+                labels.iter().map(|label| label.as_ref()).collect()
+            }
+        };
+        let grouping = match &op.group {
+            BoundGrouping::By(_) => Grouping::By(&names),
+            BoundGrouping::Without(_) => Grouping::Without(&names),
+        };
+        let name_bytes = names.len().saturating_mul(std::mem::size_of::<&str>());
+        let initial = FloatAggregator::try_new_budgeted(&schema, kind, grouping, |bytes| {
+            buffer.resize(bytes.saturating_add(name_bytes))
+        })?;
+        drop(names);
+        buffer.resize(initial.state_bytes())?;
+        Ok(Self {
+            op,
+            schema,
+            accumulator: Some(initial),
+            computed: false,
+        })
+    }
+
+    /// What folding `chunk` reserves while it is folded.
+    ///
+    /// The chunk is not retained: it is folded into the states and dropped at the end of the call,
+    /// so what this node holds afterwards is the fold and none of the input. The charge is therefore
+    /// the chunk's own bytes -- the figure [`evaluated_bytes`] reports, which is one pass over the
+    /// chunk and no state between chunks -- and not the `operand_chunk_bytes` figure a binary
+    /// operand keeps, which exists because such an operand holds a series for the whole execution
+    /// and so must count the slices of one buffer instead of the buffer once per chunk.
+    pub(crate) fn pending(&self, chunk: &EvaluatedChunk) -> usize {
+        evaluated_bytes(chunk)
+    }
+
+    /// Folds one chunk into its group.
+    ///
+    /// A chunk that closes its series closes nothing here: a group is not a series, so the last
+    /// contribution of one member says nothing about the group. The fold keeps its states, which is
+    /// what makes a group's answer its whole group's answer once the input ends.
+    pub(crate) fn accumulate(
+        &mut self,
+        path: &NodePath,
+        chunk: EvaluatedChunk,
+        buffer: &mut NodeBuffer,
+        pending: usize,
+    ) -> Result<()> {
+        // `Series` keeps its fields public, so a chunk can be built by struct literal against
+        // another schema; the states are indexed by absolute position on this node's grid, so a
+        // chunk of another field count or another grid would be folded into the wrong places.
+        validate_float_series(&self.schema, &chunk.series)
+            .map_err(|error| folded_chunk(path, &self.describe(), error))?;
+        let accumulator = self.accumulator.as_mut().ok_or_else(|| no_fold(path))?;
+        // Keep the chunk covered while the fold decides whether it needs a new group or block.
+        accumulator.update_budgeted(&chunk.series, |fold_bytes| {
+            buffer.resize(pending.saturating_add(fold_bytes))
+        })?;
+        drop(chunk);
+        buffer.resize(accumulator.state_bytes())
+    }
+
+    /// The number of retained groups, for budgeting the output queue backing storage.
+    pub(crate) fn groups(&self) -> usize {
+        self.accumulator.as_ref().map_or(0, FloatAggregator::groups)
+    }
+
+    /// Finishes the fold and returns one series per group, or `None` if the fold already ran.
+    ///
+    /// `None` is what a node reports after it has queued its series: the caller reads them from the
+    /// queue, exactly as it reads the chunks a binary node queued, so a second pull does not finish
+    /// a fold whose states are gone.
+    pub(crate) fn compute(
+        &mut self,
+        path: &NodePath,
+        buffer: &mut NodeBuffer,
+        queue_bytes: usize,
+    ) -> Result<Option<Vec<Series>>> {
+        if self.computed {
+            return Ok(None);
+        }
+        if self.accumulator.is_none() {
+            return Err(no_fold(path));
+        }
+        self.computed = true;
+        let output = self
+            .accumulator
+            .take()
+            .ok_or_else(|| no_fold(path))?
+            .finish_budgeted(|total| buffer.resize(total.saturating_add(queue_bytes)))?;
+        let output_bytes = output.iter().fold(0usize, |bytes, series| {
+            bytes.saturating_add(series_bytes(series))
+        });
+        let output_vec_bytes = output
+            .capacity()
+            .saturating_mul(std::mem::size_of::<Series>());
+        buffer.resize(
+            output_bytes
+                .saturating_add(output_vec_bytes)
+                .saturating_add(queue_bytes),
+        )?;
+        Ok(Some(output))
+    }
+}
+
+/// The rejection of a chunk that is not the layout this node folds.
+///
+/// The fold indexes its states by absolute grid position and by field, so a chunk of another field
+/// count or another grid would be folded into states that mean something else. `error` is what the
+/// series check reported, which says which of the two disagreed.
+/// Wraps the error of a chunk this node cannot fold.
+///
+/// Only a chunk that does not fit the schema is the node's own refusal; anything else is the
+/// kernel saying it does not compute this shape at all -- a native histogram in a float aggregate,
+/// for instance -- and a caller that falls back to another planner on `NotImplemented` still can.
+/// Wrapping that one in an `Execution` would deny it.
+fn folded_chunk(path: &NodePath, aggregate: &str, error: DataFusionError) -> DataFusionError {
+    match error {
+        DataFusionError::NotImplemented(_) => error,
+        error => DataFusionError::Execution(format!(
+            "the `{aggregate}` aggregate at {path} folds chunks bound to its own schema, and this \
+             one is not: {error}"
+        )),
+    }
+}
+
+/// The error for a fold that has already been finished.
+fn no_fold(path: &NodePath) -> DataFusionError {
+    DataFusionError::Execution(format!(
+        "the aggregate at {path} already finished its fold, so it takes no more input"
+    ))
+}

src/promql/src/native/execute.rs

@@ -91,9 +91,9 @@ use std::sync::Arc;
 use datafusion::error::{DataFusionError, Result};
 use datafusion::execution::memory_pool::{MemoryConsumer, MemoryPool, MemoryReservation};
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
-use datatypes::arrow::array::{Array, Float64Array};
+use datatypes::arrow::array::Float64Array;
 use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
-use datatypes::series::histogram::{HistogramBatch, HistogramView};
+use datatypes::series::histogram::HistogramView;
 use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
 use datatypes::series::samples::{
     DenseSamplesRef, RowCursor, SampleRef, Samples, Series, SeriesFields, SeriesSchema,
@@ -102,7 +102,8 @@ use datatypes::series::time::{EvalGrid, GridRange, GridSpan, GridSpec, TimeAxis}
 use datatypes::series::{MetricName, SeriesOrdering};
 
 use crate::grid_series::{
-    GridSample, GridSeriesBuilder, LabelSet, label_set, merge_same_labelset, owned,
+    GridSample, GridSeriesBuilder, LabelSet, fields_bytes, label_set, merge_same_labelset, owned,
+    series_bytes, table_bytes,
 };
 use crate::histogram_eval::evaluate_histograms;
 use crate::instant_eval::select_instants;
@@ -110,12 +111,16 @@ use crate::map::{
     ArithmeticOp, Comparison, ComparisonOp, ScalarOperand, drop_metric_name, map_arithmetic,
     map_compare, map_math,
 };
+use crate::native::aggregate::AggregateRuntime;
 use crate::native::batch::EvaluatedChunk;
 use crate::native::metrics::{NodeMetrics, NodePath};
 use crate::native::plan::{
     BoundInstant, BoundScalar, BoundVectorOp, Grouping, MapOp, NativeExpr, NativeNode, NativeTree,
     NativeType,
 };
+#[cfg(test)]
+pub(crate) use crate::native::runs::absorb_rebuilds;
+use crate::native::runs::{Run, absorb};
 use crate::range_eval::RangeEvaluator;
 use crate::select::{self, SortOrder};
 use crate::vector_match::{
@@ -189,34 +194,11 @@ fn raw_bytes(chunk: &RawSeriesChunk) -> usize {
 /// The bytes an evaluated chunk holds: its struct and the Arrow buffers behind its fields.
 ///
 /// The grid is shared by every chunk addressed on it, so it is not counted per chunk.
-fn evaluated_bytes(chunk: &EvaluatedChunk) -> usize {
-    std::mem::size_of::<EvaluatedChunk>().saturating_add(fields_bytes(&chunk.series.fields))
-}
-
-/// The bytes one evaluated series holds: its header, the axis that addresses it, the identity it
-/// carries and the sample buffers behind its fields.
 ///
-/// It is the figure a fragment of an affected window is charged with as it arrives. The axis's runs
-/// and the segments of its fields are counted in one pass over the series, so the estimate costs
-/// the fragment rather than a walk over everything the node holds. A payload a fragment shares with
-/// another — a histogram table behind two slices of it, the grid behind every axis — is counted
-/// once per fragment: that over-count is accepted here rather than a deduplication that would have
-/// to keep state across fragments.
-fn series_bytes(series: &Series) -> usize {
-    let runs = series.time.runs().count();
-    let segments: usize = series
-        .fields
-        .iter()
-        .map(|field| field.segments().len())
-        .sum();
-    std::mem::size_of::<Series>()
-        .saturating_add(std::mem::size_of::<TimeAxis>())
-        .saturating_add(runs.saturating_mul(std::mem::size_of::<(usize, usize)>()))
-        .saturating_add(
-            segments.saturating_mul(std::mem::size_of::<Samples>() + std::mem::size_of::<usize>()),
-        )
-        .saturating_add(fields_bytes(&series.fields))
-        .saturating_add(metric_bytes(&series.metric))
+/// It is what an aggregate node charges for a chunk it is folding: that node hands the chunk over
+/// to its fold and drops it, so the chunk alone is what it holds for as long as it holds anything.
+pub(super) fn evaluated_bytes(chunk: &EvaluatedChunk) -> usize {
+    std::mem::size_of::<EvaluatedChunk>().saturating_add(fields_bytes(&chunk.series.fields))
 }
 
 /// The bytes one chunk a binary operand retains holds on its own.
@@ -342,42 +324,6 @@ fn merge_workspace_bytes(rows: usize, fragments: usize, fields: usize) -> usize
         .saturating_add(output)
 }
 
-/// The bytes the fields of a chunk hold: the sample buffers of every segment, and the histogram
-/// tables those segments reference.
-fn fields_bytes(fields: &SeriesFields) -> usize {
-    fields
-        .iter()
-        .map(|field| {
-            field.segments().iter().fold(0usize, |total, segment| {
-                total.saturating_add(segment_bytes(segment))
-            })
-        })
-        .fold(0usize, usize::saturating_add)
-}
-
-/// The bytes one dense segment of samples holds.
-fn segment_bytes(segment: &Samples) -> usize {
-    match segment.dense() {
-        Some(DenseSamplesRef::Float(values)) => values.get_array_memory_size(),
-        Some(DenseSamplesRef::Histogram(histograms)) => histograms
-            .indices()
-            .get_array_memory_size()
-            .saturating_add(table_bytes(histograms.batch())),
-        Some(DenseSamplesRef::Mixed(mixed)) => mixed
-            .floats()
-            .get_array_memory_size()
-            .saturating_add(mixed.histograms().indices().get_array_memory_size())
-            .saturating_add(table_bytes(mixed.histograms().batch())),
-        // `segments` hands out dense segments only, so a nested chunk never reaches here.
-        None => 0,
-    }
-}
-
-/// The bytes one histogram table holds: the batch struct and the arrays its columns view.
-fn table_bytes(table: &Arc<HistogramBatch>) -> usize {
-    std::mem::size_of::<HistogramBatch>().saturating_add(table.as_arrow().get_array_memory_size())
-}
-
 /// The bytes the allocation `buffer` is a slice of.
 ///
 /// The capacity is the whole allocation, which is what a slice keeps alive: a key of a few bytes
@@ -408,6 +354,8 @@ enum NodeTag {
     /// Matches two evaluated vectors and applies one operation to every matched pair of series, or
     /// broadcasts one constant over the chunks of one evaluated vector.
     Binary,
+    /// Folds every series of the input into one series per group, when the input ends.
+    Aggregate,
     /// Keeps some samples of its whole input vector at each grid point.
     Vector,
 }
@@ -422,6 +370,7 @@ impl NodeTag {
             Self::Window => "window",
             Self::Map => "map",
             Self::Binary => "binary",
+            Self::Aggregate => "aggregate",
             Self::Vector => "vector operation",
         }
     }
@@ -576,7 +525,9 @@ impl MergeInput {
 /// in the query; the metrics report the same figure so a slow node can be told apart from a wide
 /// one. The reservation is also the figure itself: a node holds nothing beside what it reserved, so
 /// a gauge reads its size rather than a copy that could drift from it.
-struct NodeBuffer {
+// Visible to the aggregate node so that its fold can reserve for what it is about to allocate; the
+// reservation belongs to a node, not to a module.
+pub(super) struct NodeBuffer {
     metrics: Arc<NodeMetrics>,
     reservation: MemoryReservation,
 }
@@ -592,7 +543,7 @@ impl NodeBuffer {
     }
 
     /// Reserves `bytes` beside what this node already holds.
-    fn grow(&mut self, bytes: usize) -> Result<()> {
+    pub(super) fn grow(&mut self, bytes: usize) -> Result<()> {
         self.reservation.try_grow(bytes)?;
         self.record();
         Ok(())
@@ -603,7 +554,7 @@ impl NodeBuffer {
     /// What a node that consumes its input chunk by chunk holds afterwards is not the sum of what
     /// it consumed, so the reservation is set from the state that is left rather than grown per
     /// chunk.
-    fn resize(&mut self, bytes: usize) -> Result<()> {
+    pub(super) fn resize(&mut self, bytes: usize) -> Result<()> {
         self.reservation.try_resize(bytes)?;
         self.record();
         Ok(())
@@ -628,20 +579,6 @@ impl NodeBuffer {
     }
 }
 
-/// One run of grid positions, half-open, that hold a value in one field.
-#[derive(Clone, Copy, Debug)]
-struct Run {
-    start: usize,
-    end: usize,
-}
-
-impl Run {
-    /// Whether two runs share a position.
-    fn overlaps(self, other: Self) -> bool {
-        self.start < other.end && other.start < self.end
-    }
-}
-
 /// The positions one node has handed out, per output label set and field.
 ///
 /// A node that rewrites labels — a `Map` dropping the metric name or rewriting a label, a `Window`
@@ -876,109 +813,6 @@ fn misaligned_field(
     ))
 }
 
-// How many times `absorb` rebuilt the stored runs instead of appending to them.
-//
-// Test-only, and per thread: the runs stored afterwards are the same either way, so a rebuild shows
-// only in what it costs, and a test that reads its own thread's figure is not disturbed by another
-// test's chunks.
-#[cfg(test)]
-thread_local! {
-    static ABSORB_REBUILDS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
-}
-
-/// How many times this thread's [`absorb`] rebuilt the stored runs.
-#[cfg(test)]
-fn absorb_rebuilds() -> usize {
-    ABSORB_REBUILDS.with(std::cell::Cell::get)
-}
-
-/// Merges `incoming` into `stored`, keeping the result in position order and joining the runs that
-/// touch.
-///
-/// Within a chunk one pass over both sides is what keeps this linear, since each side is sorted and
-/// disjoint. Adding the runs one at a time and rebuilding the array after each would cost the square
-/// of the runs a chunk yields: a chunk of one hundred thousand positions alternating between values
-/// and nulls yields fifty thousand runs, and re-merging after each is over a billion element moves
-/// for a single series.
-///
-/// One pass per chunk is still a pass over everything the series has stored so far, so it is not
-/// free once the *history* grows across chunks: a series split into `K` chunks of one position each,
-/// a hole apart, stores one more run per chunk and moves `K(K+1)/2` of them, with no label collision
-/// and no illegal input anywhere. A chunk that starts no earlier than the end of the last stored run
-/// only ever adds to the end — the stored runs are sorted and disjoint, so every one of them ends
-/// before anything this chunk holds — and it is appended instead, which is amortised constant per
-/// run. That is the shape a scan hands over as its positions move forward chunk after chunk. A chunk
-/// that starts before the last stored run can split a run or reach back past it, so it still costs
-/// the full merge.
-///
-/// Either way the result is sorted, disjoint, and has the runs that touch joined, which is what
-/// [`Emitted::occupy`] looks a position up by.
-fn absorb(stored: &mut Vec<Run>, incoming: &[Run]) {
-    if incoming.is_empty() {
-        return;
-    }
-    if stored
-        .last()
-        .is_none_or(|last| incoming[0].start >= last.end)
-    {
-        append_runs(stored, incoming);
-    } else {
-        merge_runs(stored, incoming);
-    }
-}
-
-/// Appends `incoming` to `stored`, all of which lies past the last stored run.
-///
-/// Nothing already stored has to be placed again: the two sides are each in position order and
-/// nothing `incoming` holds can reach back into what is stored, so its runs are pushed on the end and
-/// the vector's own growth is the only move — amortised over the runs appended. Where its first run
-/// touches the end of the last stored run the two are one run, as they are after a merge, so it is
-/// joined rather than left beside it.
-fn append_runs(stored: &mut Vec<Run>, incoming: &[Run]) {
-    let mut incoming = incoming;
-    if let Some(last) = stored.last_mut()
-        && let Some(first) = incoming.first()
-        && first.start <= last.end
-    {
-        last.end = last.end.max(first.end);
-        incoming = &incoming[1..];
-    }
-    stored.extend_from_slice(incoming);
-}
-
-/// Merges `incoming` into `stored`, where the two can interleave.
-///
-/// One pass is enough because each side is sorted and disjoint: whichever run starts first is
-/// placed, and it joins the run before it where the two touch.
-fn merge_runs(stored: &mut Vec<Run>, incoming: &[Run]) {
-    #[cfg(test)]
-    ABSORB_REBUILDS.with(|rebuilds| rebuilds.set(rebuilds.get() + 1));
-    let mut merged: Vec<Run> = Vec::with_capacity(stored.len() + incoming.len());
-    let (mut a, mut b) = (0, 0);
-    while a < stored.len() || b < incoming.len() {
-        let larger = match (stored.get(a), incoming.get(b)) {
-            (Some(left), Some(right)) if right.start < left.start => {
-                b += 1;
-                *right
-            }
-            (Some(left), _) => {
-                a += 1;
-                *left
-            }
-            (None, Some(right)) => {
-                b += 1;
-                *right
-            }
-            (None, None) => break,
-        };
-        match merged.last_mut() {
-            Some(last) if larger.start <= last.end => last.end = last.end.max(larger.end),
-            _ => merged.push(larger),
-        }
-    }
-    *stored = merged;
-}
-
 /// What a map node applies.
 struct MapRuntime {
     ops: Arc<[MapOp]>,
@@ -2260,6 +2094,14 @@ struct NodeRuntime {
     /// `None` for every other node, and for a binary node that broadcasts a constant instead: that
     /// one holds no operand at all and is described by [`NodeRuntime::broadcast`].
     binary: Option<BinaryRuntime>,
+    /// What an aggregate node holds: the fold over what it was handed.
+    ///
+    /// Unlike a binary node's operands, nothing of the input survives a chunk: the fold keeps its
+    /// states and the chunk it read is dropped, so this is the one node whose reservation does not
+    /// follow what it was handed but what it accumulated. That is also why this node cannot report
+    /// its input complete from what it holds -- there is nothing to hold -- and reads the driver's
+    /// [`NativeExecutor::inputs_finished`] and its child's [`NativeExecutionState::settled`] instead.
+    aggregate: Option<AggregateRuntime>,
     /// What a vector node holds.
     vector: Option<VectorRuntime>,
     /// What a binary node holds when one operand is a constant: the operation and the layout of the
@@ -2674,6 +2516,102 @@ impl NodeRuntime {
         self.buffer.resize(queued)
     }
 
+    /// Folds one chunk into this node's aggregate.
+    ///
+    /// The chunk is reserved for before it is folded, by what the node reports it is worth, and the
+    /// reservation settles afterwards on the fold that is left. Unlike [`Self::retain_binary`], what
+    /// stops being held here is the chunk itself: the fold keeps its states, so a chunk that closes
+    /// its series leaves nothing behind and the node is empty of input again.
+    fn retain_aggregate(&mut self, path: &NodePath, chunk: NativeChunk) -> Result<()> {
+        let NativeChunk::Evaluated(chunk) = chunk else {
+            return Err(DataFusionError::Execution(format!(
+                "the aggregate at {path} consumes evaluated chunks, not raw samples"
+            )));
+        };
+        // The figure is read before the chunk is folded, because reserving after the fact would be
+        // reserving for memory the fold already read.
+        let pending = self
+            .aggregate
+            .as_ref()
+            .ok_or_else(|| no_aggregate(path))?
+            .pending(&chunk);
+        let positions = chunk.series.time.len();
+        self.buffer.grow(pending)?;
+        let metrics = Arc::clone(&self.buffer.metrics);
+        {
+            let _timer = metrics.timer();
+            let (aggregate, buffer) = (&mut self.aggregate, &mut self.buffer);
+            aggregate
+                .as_mut()
+                .ok_or_else(|| no_aggregate(path))?
+                .accumulate(path, chunk, buffer, pending)?;
+        }
+        metrics.record_input(1, positions);
+        // A chunk that closed its series closed it for the node below: a group is not a series, so
+        // the fold completing nothing on it is why nothing is counted here. The series this node
+        // closes are the ones it hands out, counted when it queues them.
+        Ok(())
+    }
+
+    /// Finishes this node's fold and queues one chunk per group it produced.
+    ///
+    /// Each group is one series, decided in full by the fold, so the one chunk it is handed over as
+    /// closes it: a consumer above reads "nothing more here" from that chunk rather than from another
+    /// one arriving.
+    fn compute_aggregate(&mut self, path: &NodePath) -> Result<()> {
+        let metrics = Arc::clone(&self.buffer.metrics);
+        let _timer = metrics.timer();
+        let groups = self
+            .aggregate
+            .as_ref()
+            .map_or(0, |aggregate| aggregate.groups());
+        let queue_capacity = self.queued.capacity();
+        let queue_target = self.queued.len().saturating_add(groups);
+        let queue_growth = queue_target
+            .saturating_sub(queue_capacity)
+            .saturating_mul(std::mem::size_of::<(NativeChunk, usize)>());
+        self.buffer.grow(queue_growth)?;
+        if groups > 0 {
+            self.queued.try_reserve_exact(groups).map_err(|error| {
+                DataFusionError::Execution(format!(
+                    "cannot reserve aggregate output queue: {error}"
+                ))
+            })?;
+        }
+        let queue_bytes = queue_target
+            .max(queue_capacity)
+            .saturating_mul(std::mem::size_of::<(NativeChunk, usize)>());
+        let output = {
+            let aggregate = self.aggregate.as_mut().ok_or_else(|| no_aggregate(path))?;
+            aggregate.compute(path, &mut self.buffer, queue_bytes)?
+        };
+        // The fold has returned its series; reserve their handover before appending them to the queue.
+        let Some(output) = output else {
+            return Ok(());
+        };
+        let mut closed = 0usize;
+        for series in output {
+            let bytes = series_bytes(&series);
+            let chunk = NativeChunk::Evaluated(EvaluatedChunk::try_new(series, true)?);
+            self.queued.push_back((chunk, bytes));
+            closed += 1;
+        }
+        // A group is one series and this node closes each of them exactly once, on the one chunk it
+        // hands it over as.
+        self.buffer.metrics.record_completed_series(closed);
+        // The output Vec has been consumed; the node now holds only queue backing and chunks.
+        let queued_bytes = self
+            .queued
+            .iter()
+            .fold(0usize, |total, (_, bytes)| total.saturating_add(*bytes));
+        let queue_backing = self
+            .queued
+            .capacity()
+            .saturating_mul(std::mem::size_of::<(NativeChunk, usize)>());
+        self.buffer
+            .resize(queued_bytes.saturating_add(queue_backing))
+    }
+
     /// Feeds one chunk to this node's instant selection.
     ///
     /// `None` means the series this chunk belongs to has not ended: the selection reads one whole
@@ -2876,6 +2814,8 @@ impl NodeRuntime {
         self.window = None;
         self.instant = None;
         self.binary = None;
+        // The fold is what an aggregate node holds, and dropping it is what gives its states back.
+        self.aggregate = None;
         self.vector = None;
         self.broadcast = None;
         self.source_schema = None;
@@ -2968,15 +2908,14 @@ impl NativeExecutionState {
                 // two vectors, or a constant broadcast over one -- is decided where the operands'
                 // layouts are read.
                 NativeExpr::Binary { .. } => NodeTag::Binary,
+                // An aggregate is built here and narrowed below, for the same reason: `validate`
+                // admits every kind and every stage as a legal tree, and the runtime computes one
+                // kind at one stage.
+                NativeExpr::Aggregate { .. } => NodeTag::Aggregate,
                 NativeExpr::Vector { .. } => NodeTag::Vector,
                 // A scalar is a value node: it produces the constant it was bound to, hands over no
                 // chunk and holds no state, so it is built like any other node and pulled like none.
                 NativeExpr::Scalar(_) => NodeTag::Scalar,
-                // An aggregate is legal in a bound tree, and `validate` stays the authority on that;
-                // this runtime simply does not execute it yet. Rejected before a single chunk is
-                // read: a tree the runtime cannot execute must not start, because a partial answer
-                // would be one the caller never asked for.
-                NativeExpr::Aggregate { .. } => return Err(not_implemented("Aggregate", &path)),
             };
             let buffer = NodeBuffer::new(&path, handles, &pool);
             let mut runtime = NodeRuntime {
@@ -2987,6 +2926,7 @@ impl NativeExecutionState {
                 window: None,
                 map: None,
                 binary: None,
+                aggregate: None,
                 vector: None,
                 broadcast: None,
                 source_schema: None,
@@ -3086,6 +3026,29 @@ impl NativeExecutionState {
                     // out. A broadcast that keeps the labels it reads is not where two series
                     // become one either, so it keeps none; the one that drops them does, above.
                 }
+                NativeExpr::Aggregate { input, op } => {
+                    // A `Complete` aggregate neither adds nor drops a field and evaluates on its
+                    // input's grid, so the node's own output is the layout it folds. What it does
+                    // not execute -- another kind, or another stage -- is refused here, before the
+                    // pass reaches the input that would feed it, and naming what it found is that
+                    // rejection's job: a `Partial` aggregate's output is state and names no samples
+                    // at all, so the input's layout stands in for it and the stage decides.
+                    let schema = node
+                        .output
+                        .samples()
+                        .or_else(|| input.output.samples())
+                        .cloned()
+                        .ok_or_else(|| {
+                            DataFusionError::Execution(format!(
+                                "the aggregate at {path} does not produce samples"
+                            ))
+                        })?;
+                    runtime.aggregate = Some(AggregateRuntime::new(
+                        op.clone(),
+                        schema,
+                        &mut runtime.buffer,
+                    )?);
+                }
                 NativeExpr::Vector { op, .. } => {
                     let schema = node.output.samples().cloned().ok_or_else(|| {
                         DataFusionError::Execution(format!(
@@ -3108,10 +3071,9 @@ impl NativeExecutionState {
                     // A vector operation keeps the labels of the series it reads, or answers under
                     // label sets of its own, so it keeps no ledger of what it handed out.
                 }
-                NativeExpr::Scalar(_) | NativeExpr::Aggregate { .. } => {
+                NativeExpr::Scalar(_) => {
                     // A scalar holds nothing: what its consumer reads is the constant the node was
-                    // bound to, which is on the tree rather than in this state. The aggregate is
-                    // rejected above, where the pass has the position to name.
+                    // bound to, which is on the tree rather than in this state.
                 }
             }
             // The inputs this node reads, in the order the tree declares them: a unary node's one
@@ -3119,12 +3081,11 @@ impl NativeExecutionState {
             // pushed back to front, so the node the spec names is the one the pass reaches next and
             // nothing but its own subtree lies between the two.
             let reads_input = match &node.expr {
-                NativeExpr::Input(_) | NativeExpr::Scalar(_) | NativeExpr::Aggregate { .. } => {
-                    false
-                }
+                NativeExpr::Input(_) | NativeExpr::Scalar(_) => false,
                 NativeExpr::Instant { input, .. }
                 | NativeExpr::Window { input, .. }
                 | NativeExpr::Map { input, .. }
+                | NativeExpr::Aggregate { input, .. }
                 | NativeExpr::Vector { input, .. } => {
                     pending.push((input.as_ref(), path.input()));
                     true
@@ -3194,6 +3155,14 @@ impl NativeExecutionState {
                         holds_shared[index] = input_holds_shared;
                     }
                     NodeTag::Map => holds_shared[index] = true,
+                    NodeTag::Aggregate => {
+                        // An aggregate hands over one series per group, each under the label set its
+                        // group was keyed by and each closed on its own chunk: two groups are two
+                        // label sets, so nothing above has to join pieces of a series. Its own input
+                        // is read whole -- the fold holds what it accumulated, not the chunks -- so
+                        // nothing about its input's flag reaches it either.
+                        holds_shared[index] = false;
+                    }
                     NodeTag::Scalar => {
                         // A scalar produces one value and no labels, so it is not a stream of series
                         // and nothing above it can read two of them under one label set. The flag
@@ -3747,9 +3716,24 @@ impl NativeExecutor {
             // A scalar produces its value without reading anything, so it is settled from the
             // start: there is no chunk it owes its consumer.
             NodeTag::Scalar => true,
+            // These report as they are handed chunks, so a child with nothing left is all it takes
+            // for them to have nothing left either.
             NodeTag::Map | NodeTag::Window | NodeTag::Instant => self.state.nodes[index]
                 .input
                 .is_none_or(|child| self.settled(child)),
+            // An aggregate answers one series per group and only once its input ended, so its own
+            // fold has to have run as well: a child that has handed everything up is not enough,
+            // because every one of those chunks was folded and none of them answered yet. This is
+            // what keeps it out of the line above, which asks only about the child.
+            NodeTag::Aggregate => {
+                runtime
+                    .aggregate
+                    .as_ref()
+                    .is_some_and(|aggregate| aggregate.computed)
+                    && self.state.nodes[index]
+                        .input
+                        .is_none_or(|child| self.settled(child))
+            }
             NodeTag::Vector => {
                 runtime
                     .vector
@@ -3947,6 +3931,7 @@ impl NativeExecutor {
                 // does, so its values are counted rather than read as its positions.
                 runtime.pop_queued(evaluated_values)
             }
+            NodeTag::Aggregate => self.aggregate_node(index),
             NodeTag::Vector => {
                 let child = input.ok_or_else(|| missing_input(tag))?;
                 // Which samples are kept at a point is not decided until the whole input has
@@ -3966,6 +3951,35 @@ impl NativeExecutor {
         }
     }
 
+    /// Folds everything the aggregate at `index` can read and hands over one group's series.
+    ///
+    /// What it reads is taken to the end of what its child has, because nothing the fold is handed
+    /// is retained: a chunk is folded as it arrives and dropped, so reading the child dry costs the
+    /// fold's states and not the input's bytes. Nothing is answered until the input has ended --
+    /// before that, a group's value at a point is still missing the members that have not arrived.
+    ///
+    /// That is the difference from a binary node. A binary node retains both operands and so can ask
+    /// whether they hold anything still open; an aggregate keeps none of them and asks the driver
+    /// instead: [`Self::inputs_finished`] says the caller will hand over nothing more, and
+    /// [`NativeExecutionState::settled`] says the child has none left to produce.
+    fn aggregate_node(&mut self, index: usize) -> Result<Option<NativeChunk>> {
+        let child = self.state.nodes[index]
+            .input
+            .ok_or_else(|| missing_input(NodeTag::Aggregate))?;
+        while let Some(chunk) = self.pull(child)? {
+            let path = &self.state.paths[index];
+            self.state.runtimes[index].retain_aggregate(path, chunk)?;
+        }
+        if !self.inputs_finished || !self.settled(child) {
+            return Ok(None);
+        }
+        let path = &self.state.paths[index];
+        self.state.runtimes[index].compute_aggregate(path)?;
+        // A group's series holds one value per field and point it decided, so its values are counted
+        // rather than read as its positions, as a binary result's are.
+        self.state.runtimes[index].pop_queued(evaluated_values)
+    }
+
     /// Pulls the vector side of the broadcast at `index` until it hands over a chunk, and returns
     /// that chunk transformed.
     ///
@@ -4211,11 +4225,7 @@ fn valid_values(chunk: &EvaluatedChunk) -> usize {
         .sum()
 }
 
-/// The values a chunk a binary node hands over holds.
-///
-/// A binary result is an evaluated chunk, so this is [`valid_values`]: one value per field sample
-/// that holds one. A raw chunk cannot reach a binary node's queue, and is answered as its positions
-/// so an impossible queue cannot under-count.
+/// The values an evaluated chunk hands over, or the timestamps a raw source hands over.
 fn evaluated_values(chunk: &NativeChunk) -> usize {
     match chunk {
         NativeChunk::Evaluated(chunk) => valid_values(chunk),
@@ -4223,18 +4233,6 @@ fn evaluated_values(chunk: &NativeChunk) -> usize {
     }
 }
 
-/// The rejection of a node kind this runtime does not execute yet.
-///
-/// It names the kind and the node's position: the position is what the metrics and `EXPLAIN` call
-/// the same node, so a reader can find it in the tree. An aggregate is the one kind left here: a
-/// scalar executes as a value node, and a binary node executes both as a vector match and as a
-/// constant broadcast.
-fn not_implemented(kind: &str, path: &NodePath) -> DataFusionError {
-    DataFusionError::NotImplemented(format!(
-        "the native runtime does not execute {kind} nodes yet: {kind} at {path}"
-    ))
-}
-
 /// The error for a window whose evaluator this execution already closed.
 fn closed(path: &NodePath) -> DataFusionError {
     DataFusionError::Execution(format!("the window at {path} is already closed"))
@@ -4316,11 +4314,27 @@ fn truncated_input(path: &NodePath) -> DataFusionError {
     ))
 }
 
+/// The rejection of a node kind this runtime does not execute yet.
+///
+/// It names the kind and the node's position: the position is what the metrics and `EXPLAIN` call
+/// the same node, so a reader can find it in the tree.
+fn not_implemented(kind: &str, path: &NodePath) -> DataFusionError {
+    DataFusionError::NotImplemented(format!(
+        "the native runtime does not execute {kind} nodes yet: {kind} at {path}"
+    ))
+}
+
 /// The error for a node asked to compute without the runtime its kind holds.
 fn no_binary(path: &NodePath) -> DataFusionError {
     DataFusionError::Execution(format!("the node at {path} holds no binary operation"))
 }
 
+/// The error for an aggregate node whose fold is missing.
+fn no_aggregate(path: &NodePath) -> DataFusionError {
+    DataFusionError::Execution(format!("the node at {path} holds no aggregate"))
+}
+
+/// The error for a node asked to run a vector operation without one.
 fn no_vector(path: &NodePath) -> DataFusionError {
     DataFusionError::Execution(format!("the node at {path} holds no vector operation"))
 }

src/promql/src/native/execute/tests.rs

@@ -46,8 +46,9 @@ use crate::map::MathOp;
 use crate::map_label::LabelRewrite;
 use crate::native::metrics::{NodeMetric, NodeMetrics, NodePath};
 use crate::native::plan::{
-    AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar,
-    BoundVectorOp, BoundWindow, Grouping, MapOp, NativeExpr, NativeNode, NativeTree, NativeType,
+    AggregateKind, AggregateStage, AggregateStateSchema, BoundAggregate, BoundInput, BoundInstant,
+    BoundMap, BoundScalar, BoundVectorOp, BoundWindow, Grouping, MapOp, NativeExpr, NativeNode,
+    NativeTree, NativeType,
 };
 use crate::range_eval::{RangeEvaluator, RangeFunction, RangeWindow};
 use crate::select::SortOrder;
@@ -490,6 +491,7 @@ fn source_runtime(pool: &Arc<dyn MemoryPool>) -> NodeRuntime {
         window: None,
         map: None,
         binary: None,
+        aggregate: None,
         broadcast: None,
         vector: None,
         source_schema: None,
@@ -512,6 +514,7 @@ fn binary_runtime(pool: &Arc<dyn MemoryPool>, schema: &Arc<SeriesSchema>) -> Nod
             Arc::new(binary_matching(BinaryOp::Add)),
             [Arc::clone(schema), Arc::clone(schema)],
         )),
+        aggregate: None,
         broadcast: None,
         vector: None,
         source_schema: None,
@@ -1093,30 +1096,17 @@ fn a_downstream_window_emits_before_its_input_series_ends() {
     }
 }
 
-/// Every node kind the runtime does not execute is rejected, naming the kind and its position.
+/// Every binary shape the runtime does not execute is rejected, naming the kind and its position.
 ///
 /// A scalar is not one of them any more: it is a value node, and a binary node over a constant is
-/// executed as a broadcast. The binary shapes that stay refused are the ones no kernel here takes,
-/// and the aggregate is still a kind this runtime does not run at all.
+/// executed as a broadcast. The binary shapes that stay refused are the ones no kernel here takes;
+/// the aggregate kinds and stages this runtime does not execute are refused by
+/// `an_aggregate_kind_or_stage_the_runtime_does_not_compute_is_named`.
 #[test]
 fn an_unimplemented_node_is_reported_not_panicked() {
     // One grid shared by the nodes that validate by allocation identity.
     let statement = || grid(0, 120_000, 2);
 
-    let aggregate_tree = {
-        let out = statement();
-        NativeNode {
-            expr: NativeExpr::Aggregate {
-                input: Box::new(window(input(), Arc::clone(&out))),
-                op: BoundAggregate {
-                    kind: AggregateKind::Sum,
-                    group: Grouping::By(Arc::new(["job".into()])),
-                    stage: AggregateStage::Complete,
-                },
-            },
-            output: NativeType::Samples(evaluated_schema(out)),
-        }
-    };
     // Two constants make a scalar, which is a legal node and a shape no binary here broadcasts
     // over: there is no vector to pair either of them with.
     let two_scalars_tree = {
@@ -1148,12 +1138,6 @@ fn an_unimplemented_node_is_reported_not_panicked() {
     };
 
     let cases = [
-        (
-            "Aggregate",
-            "root",
-            "does not execute Aggregate",
-            aggregate_tree,
-        ),
         ("Binary", "root", "two constants", two_scalars_tree),
         ("Binary", "root", "matching modifiers", matched_scalar_tree),
     ];
@@ -6568,6 +6552,158 @@ fn a_constant_on_the_right_multiplies_every_value() {
     assert_eq!(execution.state().reserved_bytes(), 0);
 }
 
+/// A `Complete` aggregate over `inner`, carrying `inner`'s fields and evaluated on its grid.
+///
+/// A `Complete` aggregate neither adds nor drops a field and lands on its input's grid, so the node's
+/// own output is the layout its input produces.
+fn aggregate(inner: NativeNode, op: BoundAggregate) -> NativeNode {
+    let fields = inner
+        .output
+        .samples()
+        .expect("an aggregate reads samples")
+        .fields
+        .clone();
+    let grid = inner
+        .output
+        .grid()
+        .expect("a complete aggregate reads an evaluated vector")
+        .clone();
+    let output = NativeType::Samples(Arc::new(SeriesSchema {
+        fields,
+        eval_grid: Some(grid),
+    }));
+    NativeNode {
+        expr: NativeExpr::Aggregate {
+            input: Box::new(inner),
+            op,
+        },
+        output,
+    }
+}
+
+/// A `Complete` `sum` grouped as `group`.
+fn sum_grouping(group: Grouping) -> BoundAggregate {
+    BoundAggregate {
+        kind: AggregateKind::Sum,
+        group,
+        stage: AggregateStage::Complete,
+    }
+}
+
+/// `name{job=...,instance=...}`, the two labels these tests group by.
+fn job_instance_metric(name: &str, job: &str, instance: &str) -> MetricName {
+    MetricName::try_new(
+        Some(name.into()),
+        vec![
+            Label {
+                name: "job".into(),
+                value: job.into(),
+            },
+            Label {
+                name: "instance".into(),
+                value: instance.into(),
+            },
+        ],
+    )
+    .unwrap()
+}
+
+/// What one aggregate result reports: its name, its tags and the value at each point it decided.
+type GroupResult = (
+    Option<String>,
+    Vec<(String, String)>,
+    Vec<(usize, Option<f64>)>,
+);
+
+/// The groups a run produced, each with the absolute positions it decided, sorted by identity.
+///
+/// Sorting is the test's, not the runtime's: a group's place in the answer is the order its first
+/// input appeared, which the tests that check it assert on the stream rather than here.
+fn groups_of(chunks: &[EvaluatedChunk]) -> Vec<GroupResult> {
+    let mut results: Vec<GroupResult> = chunks
+        .iter()
+        .map(|chunk| {
+            let mut tags: Vec<(String, String)> = chunk
+                .series
+                .metric
+                .tags()
+                .iter()
+                .map(|tag| (tag.name.to_string(), tag.value.to_string()))
+                .collect();
+            tags.sort();
+            (
+                chunk.series.metric.name().map(str::to_string),
+                tags,
+                decided(std::slice::from_ref(chunk)),
+            )
+        })
+        .collect();
+    results.sort_by(|left, right| (&left.0, &left.1).cmp(&(&right.0, &right.1)));
+    results
+}
+
+/// One group's answer under the name and tags `tags` give it, ordered by label name.
+fn grouped(tags: &[(&str, &str)], points: Vec<(usize, Option<f64>)>) -> GroupResult {
+    let mut tags: Vec<(String, String)> = tags
+        .iter()
+        .map(|(name, value)| ((*name).to_string(), (*value).to_string()))
+        .collect();
+    tags.sort();
+    (None, tags, points)
+}
+
+/// One group's answer when every label of its input was dropped.
+fn unlabelled(points: Vec<(usize, Option<f64>)>) -> GroupResult {
+    grouped(&[], points)
+}
+
+/// `sum` with no grouping drops every label, name and tags alike, and adds the members per point.
+#[test]
+fn an_aggregate_drops_every_label_and_adds_its_group_per_point() {
+    let grid = grid(0, 60_000, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new([]))),
+    ))
+    .unwrap();
+    let set = ExecutionPlanMetricsSet::new();
+    let mut execution = executor(&tree, pool(), &set);
+    for (metric, values) in [
+        (
+            job_instance_metric("requests", "api", "a"),
+            vec![Some(1.0), Some(2.0), Some(3.0), Some(4.0)],
+        ),
+        (
+            job_instance_metric("requests", "db", "b"),
+            vec![Some(10.0), Some(20.0), Some(30.0), Some(40.0)],
+        ),
+    ] {
+        execution
+            .push_evaluated(evaluated_chunk_of(&schema, metric, 0, values, true))
+            .unwrap();
+    }
+
+    let output = execution.finish().unwrap();
+    assert_eq!(
+        groups_of(&output),
+        vec![unlabelled(vec![
+            (0, Some(11.0)),
+            (1, Some(22.0)),
+            (2, Some(33.0)),
+            (3, Some(44.0)),
+        ])],
+        "the two series were not folded into one unlabelled group"
+    );
+    // One group is one series, so it closes exactly once and its chunk says so.
+    assert!(output.len() == 1 && output[0].is_last);
+    assert_eq!(
+        count(&set, &NodePath::root(), NodeMetric::COMPLETED_SERIES),
+        1
+    );
+    assert_eq!(execution.state().reserved_bytes(), 0);
+}
+
 /// The side a constant is on decides which operand the vector's value is.
 ///
 /// Division is not commutative, so `x / 100` and `100 / x` answering different values is what says
@@ -7041,6 +7177,208 @@ fn a_window_above_a_broadcast_merges_the_series_it_collapses() {
     assert_eq!(execution.state().reserved_bytes(), 0);
 }
 
+/// Every supported `Complete` kind reaches its matching float kernel through the native runtime.
+#[test]
+fn complete_aggregate_kinds_fold_two_values() {
+    let cases = [
+        (AggregateKind::Sum, 4.0),
+        (AggregateKind::Count, 2.0),
+        (AggregateKind::Avg, 2.0),
+        (AggregateKind::Min, 1.0),
+        (AggregateKind::Max, 3.0),
+        (AggregateKind::Group, 1.0),
+        (AggregateKind::Stddev, 1.0),
+        (AggregateKind::Stdvar, 1.0),
+    ];
+
+    for (kind, expected) in cases {
+        let grid = grid(0, 60_000, 1);
+        let schema = evaluated_schema(Arc::clone(&grid));
+        let tree = NativeTree::try_new(aggregate(
+            evaluated_input(Arc::clone(&grid)),
+            BoundAggregate {
+                kind,
+                group: Grouping::By(Arc::new([])),
+                stage: AggregateStage::Complete,
+            },
+        ))
+        .unwrap();
+        let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
+        for value in [1.0, 3.0] {
+            execution
+                .push_evaluated(evaluated_chunk_of(
+                    &schema,
+                    metric(),
+                    0,
+                    vec![Some(value)],
+                    true,
+                ))
+                .unwrap();
+        }
+
+        let output = execution.finish().unwrap();
+        assert_eq!(
+            groups_of(&output),
+            vec![unlabelled(vec![(0, Some(expected))])],
+            "{kind:?} did not compute its native complete result"
+        );
+    }
+}
+
+/// The same series, delivered in two chunks, fold to what the whole series folds to.
+#[test]
+fn an_aggregate_folds_a_series_split_across_chunks_like_a_whole_one() {
+    let bound = grid(0, 60_000, 4);
+    let schema = evaluated_schema(Arc::clone(&bound));
+    // A fresh tree per execution: the node's state is the execution's, not the tree's.
+    let tree = || {
+        NativeTree::try_new(aggregate(
+            evaluated_input(grid(0, 60_000, 4)),
+            sum_grouping(Grouping::By(Arc::new(["job".into()]))),
+        ))
+        .unwrap()
+    };
+    let (first, second) = (
+        job_instance_metric("requests", "api", "a"),
+        job_instance_metric("requests", "api", "b"),
+    );
+
+    // One chunk per series.
+    let mut whole = executor(&tree(), pool(), &ExecutionPlanMetricsSet::new());
+    for (metric, values) in [
+        (
+            first.clone(),
+            vec![Some(1.0), Some(2.0), Some(3.0), Some(4.0)],
+        ),
+        (
+            second.clone(),
+            vec![Some(10.0), Some(20.0), Some(30.0), Some(40.0)],
+        ),
+    ] {
+        whole
+            .push_evaluated(evaluated_chunk_of(&schema, metric, 0, values, true))
+            .unwrap();
+    }
+    let whole = whole.finish().unwrap();
+
+    // The first series split in two, the second whole: the group is the same group, folded from
+    // positions the chunks name absolutely.
+    let mut split = executor(&tree(), pool(), &ExecutionPlanMetricsSet::new());
+    split
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            first.clone(),
+            0,
+            vec![Some(1.0), Some(2.0)],
+            false,
+        ))
+        .unwrap();
+    assert!(
+        split.next_chunk().unwrap().is_none(),
+        "a chunk that does not close its series answered for the group"
+    );
+    split
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            first,
+            2,
+            vec![Some(3.0), Some(4.0)],
+            true,
+        ))
+        .unwrap();
+
+    split
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            second,
+            0,
+            vec![Some(10.0), Some(20.0), Some(30.0), Some(40.0)],
+            true,
+        ))
+        .unwrap();
+
+    let split = split.finish().unwrap();
+    assert_eq!(groups_of(&split), groups_of(&whole));
+    assert_eq!(
+        groups_of(&split),
+        vec![grouped(
+            &[("job", "api")],
+            vec![
+                (0, Some(11.0)),
+                (1, Some(22.0)),
+                (2, Some(33.0)),
+                (3, Some(44.0)),
+            ],
+        )],
+        "the split series was folded as two groups or at the wrong positions"
+    );
+}
+
+/// A point no member of a group has a value at is not a point of the group's result.
+///
+/// Sparse input is what says the fold indexes its states by absolute position: a chunk holding
+/// positions 0 and 1 and another holding 1 and 2 leave position 3 to nothing, and the hole between
+/// the rows is never filled with a value.
+#[test]
+fn an_aggregate_reports_only_the_points_a_group_holds_a_value_at() {
+    let grid = grid(0, 60_000, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new([]))),
+    ))
+    .unwrap();
+    let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
+    // Positions 0 and 1 of the first series, 1 and 2 of the second, and nothing at 3.
+    execution
+        .push_evaluated(sparse_chunk(
+            &schema,
+            job_instance_metric("requests", "api", "a"),
+            &[(0, 2)],
+            vec![1.0, 2.0],
+            true,
+        ))
+        .unwrap();
+    execution
+        .push_evaluated(sparse_chunk(
+            &schema,
+            job_instance_metric("requests", "db", "b"),
+            &[(1, 2)],
+            vec![20.0, 300.0],
+            true,
+        ))
+        .unwrap();
+
+    let output = execution.finish().unwrap();
+    assert_eq!(
+        groups_of(&output),
+        vec![
+            // Position 1 is the only one both members hold, and it is their sum.
+            unlabelled(vec![(0, Some(1.0)), (1, Some(22.0)), (2, Some(300.0))]),
+        ],
+        "the group was answered at a point no member holds, or summed the wrong members"
+    );
+}
+
+/// An aggregate handed no series produces nothing at all: no group, so no empty series either.
+#[test]
+fn an_aggregate_over_no_series_produces_nothing() {
+    let grid = grid(0, 60_000, 4);
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new(["job".into()]))),
+    ))
+    .unwrap();
+    let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
+    let output = execution.finish().unwrap();
+    assert!(
+        output.is_empty(),
+        "an aggregate with no input answered {} series",
+        output.len()
+    );
+    assert_eq!(execution.state().reserved_bytes(), 0);
+}
+
 /// A broadcast hands one chunk over for every chunk it reads, and closes the series once.
 ///
 /// The metrics are what says the chunk-per-chunk behaviour held: one input batch per chunk read, one
@@ -7256,6 +7594,411 @@ fn a_broadcast_refuses_two_series_that_share_a_label_set_and_releases_the_reserv
     );
 }
 
+/// A series that ended is one member of a group, not the group: nothing is answered until the input
+/// ends.
+///
+/// The observable state between the two is what this pins: the fold has read the closing chunk of
+/// the first series and answered nothing, and `computed` still says so.
+#[test]
+fn an_aggregate_answers_nothing_until_its_whole_input_ends() {
+    let grid = grid(0, 60_000, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new([]))),
+    ))
+    .unwrap();
+    let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
+    let computed = |execution: &NativeExecutor| {
+        execution.state().runtimes[0]
+            .aggregate
+            .as_ref()
+            .expect("the root holds an aggregate")
+            .computed
+    };
+
+    execution
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            job_instance_metric("requests", "api", "a"),
+            0,
+            vec![Some(1.0), Some(2.0), Some(3.0), Some(4.0)],
+            true,
+        ))
+        .unwrap();
+    assert!(execution.next_chunk().unwrap().is_none());
+    assert!(!computed(&execution), "the fold finished on one series");
+    assert!(
+        execution.next_chunk().unwrap().is_none(),
+        "the closing chunk of one series answered for its group"
+    );
+    // The second series closed too, and the input still has not ended.
+    execution
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            job_instance_metric("requests", "db", "b"),
+            0,
+            vec![Some(10.0), Some(20.0), Some(30.0), Some(40.0)],
+            true,
+        ))
+        .unwrap();
+    assert!(
+        !computed(&execution),
+        "the fold finished before the input ended"
+    );
+    assert!(
+        execution.next_chunk().unwrap().is_none(),
+        "every series closed, and the group was answered before the input ended"
+    );
+
+    execution.end_of_input().unwrap();
+    assert!(
+        !computed(&execution),
+        "ending the input finished the fold before anything pulled for it"
+    );
+    let mut output = Vec::new();
+    while let Some(chunk) = execution.next_chunk().unwrap() {
+        output.push(chunk);
+    }
+    assert_eq!(
+        groups_of(&output),
+        vec![unlabelled(vec![
+            (0, Some(11.0)),
+            (1, Some(22.0)),
+            (2, Some(33.0)),
+            (3, Some(44.0)),
+        ])],
+        "the group was not answered once the input ended"
+    );
+}
+
+/// A chunk of another layout is refused, naming the aggregate and its position, and the execution
+/// stops with nothing reserved.
+///
+/// The fold indexes its states by absolute position and by field, so a chunk of another field count
+/// would be folded into states that mean something else. It is refused rather than folded, and what
+/// the refusal leaves behind is a stopped execution whose reservation went back to the pool.
+#[test]
+fn an_aggregate_refuses_a_chunk_of_another_layout_and_returns_its_reservation() {
+    let grid = grid(0, 60_000, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new([]))),
+    ))
+    .unwrap();
+    let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
+    execution
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            metric(),
+            0,
+            vec![Some(1.0), Some(2.0), Some(3.0), Some(4.0)],
+            true,
+        ))
+        .unwrap();
+    // Two fields where the tree bound one: the source node accepts it, the fold refuses it.
+    let wider = Arc::new(SeriesSchema {
+        fields: field_count(2),
+        eval_grid: Some(Arc::clone(&grid)),
+    });
+    execution
+        .push_evaluated(evaluated_columns(
+            &wider,
+            metric(),
+            0,
+            vec![vec![Some(1.0); 4], vec![Some(2.0); 4]],
+            true,
+        ))
+        .unwrap();
+
+    let error = execution
+        .next_chunk()
+        .expect_err("a chunk of another field count was folded");
+    assert!(
+        matches!(error, DataFusionError::Execution(_)),
+        "the refusal is not an execution error: {error}"
+    );
+    let message = error.to_string();
+    assert!(
+        message.contains("aggregate at root"),
+        "the refusal does not name the aggregate: {message}"
+    );
+    assert!(
+        message.contains("field count"),
+        "the refusal does not say what disagreed: {message}"
+    );
+    assert_eq!(execution.state().reserved_bytes(), 0);
+    // A stopped execution is not resumed: the fold must not be asked to continue after this.
+    assert!(execution.next_chunk().is_err());
+}
+
+/// A pool rejection while the first aggregate block is admitted releases the source and fold.
+#[test]
+fn aggregate_budget_rejects_first_block_and_releases_everything() {
+    let grid = grid(0, 1, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new([]))),
+    ))
+    .unwrap();
+    let pool = bounded_pool(2_000);
+    let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
+    execution
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            metric(),
+            0,
+            vec![Some(2.0)],
+            true,
+        ))
+        .unwrap();
+    assert!(execution.next_chunk().is_err());
+    assert_eq!(pool.reserved(), 0);
+    assert_eq!(execution.state().reserved_bytes(), 0);
+}
+
+/// The aggregate runtime charges its output queue, releases each output as it is pulled, and keeps
+/// queue backing until the runtime itself is released.
+#[test]
+fn aggregate_runtime_queue_keeps_only_backing_after_pops_until_release() {
+    let grid = grid(0, 1, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new(["job".into()]))),
+    ))
+    .unwrap();
+    let pool = bounded_pool(1 << 20);
+    let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
+    for (job, value) in [("api", 1.0), ("db", 2.0)] {
+        execution
+            .push_evaluated(evaluated_chunk_of(
+                &schema,
+                job_instance_metric("requests", job, "a"),
+                0,
+                vec![Some(value)],
+                true,
+            ))
+            .unwrap();
+    }
+    execution.end_of_input().unwrap();
+    let first = execution.next_chunk().unwrap().expect("first group output");
+    assert_eq!(first.series.time.positions().count(), 1);
+    let runtime = &execution.state().runtimes[0];
+    let entry = std::mem::size_of::<(NativeChunk, usize)>();
+    assert_eq!(runtime.queued.len(), 1);
+    let remaining = runtime.queued[0].1;
+    let expected = runtime.queued.capacity() * entry + remaining;
+    assert_eq!(runtime.buffer.reservation.size(), expected);
+
+    let second = execution
+        .next_chunk()
+        .unwrap()
+        .expect("second group output");
+    assert_eq!(second.series.time.positions().count(), 1);
+    let runtime = &execution.state().runtimes[0];
+    assert!(runtime.queued.is_empty());
+    let backing = runtime.queued.capacity() * entry;
+    assert!(backing > 0);
+    assert_eq!(runtime.buffer.reservation.size(), backing);
+    assert!(execution.next_chunk().unwrap().is_none());
+    // Pulling the terminal None releases the queue backing; only explicit close checks the
+    // runtime's final cleanup below.
+    assert_eq!(execution.state().runtimes[0].buffer.reservation.size(), 0);
+
+    execution.state.close().unwrap();
+    assert_eq!(execution.state().runtimes[0].queued.capacity(), 0);
+    assert_eq!(execution.state().runtimes[0].buffer.reservation.size(), 0);
+    assert_eq!(pool.reserved(), 0);
+}
+
+/// An aggregate whose sparse output has a few points on a large grid can finish in a small pool.
+#[test]
+fn aggregate_sparse_output_budget_uses_rows_not_grid_length() {
+    let grid = grid(0, 1, 100_000);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new([]))),
+    ))
+    .unwrap();
+    let pool = bounded_pool(64 * 1024);
+    let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
+    execution
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            metric(),
+            10,
+            vec![Some(2.0)],
+            true,
+        ))
+        .unwrap();
+    let output = execution.finish().unwrap();
+    assert_eq!(groups_of(&output), vec![unlabelled(vec![(10, Some(2.0))])]);
+    assert_eq!(pool.reserved(), 0);
+}
+
+/// An aggregate retains its fold before finish and release returns its reservation.
+#[test]
+fn aggregate_fold_state_is_released_after_input() {
+    let grid = grid(0, 1, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = NativeTree::try_new(aggregate(
+        evaluated_input(Arc::clone(&grid)),
+        sum_grouping(Grouping::By(Arc::new([]))),
+    ))
+    .unwrap();
+    let pool = bounded_pool(1 << 20);
+    let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
+    execution
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            metric(),
+            0,
+            vec![Some(2.0)],
+            true,
+        ))
+        .unwrap();
+    assert!(execution.next_chunk().unwrap().is_none());
+    let runtime = &execution.state.runtimes[0];
+    let aggregate = runtime.aggregate.as_ref().unwrap();
+    assert_eq!(aggregate.groups(), 1);
+    assert!(!aggregate.computed);
+    assert!(pool.reserved() > 0);
+    execution.state.close().unwrap();
+    assert_eq!(pool.reserved(), 0);
+}
+
+/// An aggregate kind or stage this runtime does not compute is refused, naming what was found.
+#[test]
+fn an_aggregate_kind_or_stage_the_runtime_does_not_compute_is_named() {
+    let grid = || grid(0, 60_000, 4);
+    let kind_tree = NativeTree::try_new(aggregate(
+        evaluated_input(grid()),
+        BoundAggregate {
+            kind: AggregateKind::Quantile { quantile: 0.5 },
+            group: Grouping::By(Arc::new([])),
+            stage: AggregateStage::Complete,
+        },
+    ))
+    .unwrap();
+    let partial_tree = {
+        let out = grid();
+        let group = Grouping::By(Arc::new(["job".into()]));
+        let state = Arc::new(AggregateStateSchema {
+            group: group.clone(),
+            accumulators: fields(),
+            grid: Arc::clone(&out),
+        });
+        NativeTree::try_new(NativeNode {
+            expr: NativeExpr::Aggregate {
+                input: Box::new(evaluated_input(Arc::clone(&out))),
+                op: BoundAggregate {
+                    kind: AggregateKind::Sum,
+                    group,
+                    stage: AggregateStage::Partial,
+                },
+            },
+            output: NativeType::AggregateState(state),
+        })
+        .unwrap()
+    };
+
+    for (found, tree) in [("Quantile", kind_tree), ("Partial", partial_tree)] {
+        let error = match NativeExecutionState::try_new(
+            &tree,
+            pool(),
+            &ExecutionPlanMetricsSet::new(),
+            0,
+        ) {
+            Ok(_) => panic!("the runtime accepted a {found} aggregate"),
+            Err(error) => error,
+        };
+        assert!(
+            matches!(error, DataFusionError::NotImplemented(_)),
+            "{found}: {error}"
+        );
+        let message = error.to_string();
+        assert!(message.contains(found), "{found}: {message}");
+        // A `Partial` rejection has to say why it is not computed: the stages it would need.
+        if found == "Partial" {
+            assert!(message.contains("Final"), "{message}");
+        }
+    }
+}
+
+/// An aggregate above a window that collects and merges its input waits for that merge.
+///
+/// The multiplication drops the metric name, so the two input series land on one label set and the
+/// `last_over_time` above them has to collect and merge before it evaluates. The aggregate above the
+/// window must wait for that merge: its guard is `inputs_finished && settled(child)`, and with only
+/// the first half it finalizes on the pull that follows the caller's last chunk, before the window
+/// has answered.
+#[test]
+fn an_aggregate_above_a_collecting_window_waits_for_the_merge() {
+    let grid = grid(0, 60_000, 4);
+    let schema = evaluated_schema(Arc::clone(&grid));
+    let tree = || {
+        NativeTree::try_new(aggregate(
+            window_with(
+                broadcast_right(
+                    evaluated_input(Arc::clone(&grid)),
+                    2.0,
+                    BinaryOp::Mul,
+                    false,
+                ),
+                RangeFunction::LastOverTime,
+                60_000,
+                Some(Arc::clone(&grid)),
+                Arc::clone(&grid),
+            ),
+            sum_grouping(Grouping::By(Arc::new([]))),
+        ))
+        .unwrap()
+    };
+
+    let mut execution = executor(&tree(), pool(), &ExecutionPlanMetricsSet::new());
+    // Two series that only differ by name, each holding half of the grid the other leaves empty:
+    // after the name is dropped they are one series only once merged.
+    execution
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            named_metric("requests", "x"),
+            0,
+            vec![Some(1.0), Some(2.0)],
+            true,
+        ))
+        .unwrap();
+    assert!(
+        execution.next_chunk().unwrap().is_none(),
+        "the aggregate answered before the window above it had merged"
+    );
+    execution
+        .push_evaluated(evaluated_chunk_of(
+            &schema,
+            named_metric("requests_total", "x"),
+            2,
+            vec![Some(3.0), Some(4.0)],
+            true,
+        ))
+        .unwrap();
+
+    let out = run(&mut execution);
+    assert_eq!(
+        decided(&out),
+        vec![
+            (0, Some(2.0)),
+            (1, Some(4.0)),
+            (2, Some(6.0)),
+            (3, Some(8.0)),
+        ],
+        "the aggregate did not fold the merged window result"
+    );
+    assert_eq!(execution.state().reserved_bytes(), 0);
+}
+
 fn vector_op(inner: NativeNode, op: BoundVectorOp) -> NativeNode {
     let output = inner.output.clone();
     NativeNode {

src/promql/src/native/mod.rs

@@ -27,8 +27,10 @@
 //! - A series closes exactly once, on the chunk that says so, so a consumer can tell "nothing more
 //!   here" from "no more input".
 
+mod aggregate;
 pub mod batch;
 mod display;
 pub mod execute;
 pub mod metrics;
 pub mod plan;
+pub(crate) mod runs;

src/promql/src/native/runs.rs

@@ -0,0 +1,146 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! The grid positions a field occupies, as sorted, disjoint runs.
+//!
+//! A node that rewrites labels — a `Map` dropping the metric name, a `Window` dropping the name and
+//! the empty-valued labels — can turn two input series into one label set, and every layer above
+//! then reads the two chunks as one series. What decides whether that pair is legal is which
+//! positions of each field already hold a value: two values of one field at one position cannot be
+//! merged away, while a chunk that fills positions the set left empty merges cleanly, as
+//! [`crate::grid_series::merge_same_labelset`] does.
+//!
+//! The state is kept per label set, per field, as runs, so a chunk of many positions costs one run
+//! rather than one entry per position. Keeping it in one place is what keeps the two layers that
+//! check it — the native runtime's own [`crate::native::execute::Emitted`] table and the
+//! cross-partition [`crate::extension_plan::fused_series::LabelSets`] — from drifting apart: both
+//! ask the same functions for the same answers, so the fast path one of them earned is the fast path
+//! the other gets.
+
+/// One run of grid positions, half-open, that hold a value in one field.
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct Run {
+    pub(crate) start: usize,
+    pub(crate) end: usize,
+}
+
+impl Run {
+    /// Whether two runs share a position.
+    pub(crate) fn overlaps(self, other: Self) -> bool {
+        self.start < other.end && other.start < self.end
+    }
+}
+
+// How many times `absorb` rebuilt the stored runs instead of appending to them.
+//
+// Test-only, and per thread: the runs stored afterwards are the same either way, so a rebuild shows
+// only in what it costs, and a test that reads its own thread's figure is not disturbed by another
+// test's chunks.
+#[cfg(test)]
+thread_local! {
+    static ABSORB_REBUILDS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
+}
+
+/// How many times this thread's [`absorb`] rebuilt the stored runs.
+#[cfg(test)]
+pub(crate) fn absorb_rebuilds() -> usize {
+    ABSORB_REBUILDS.with(std::cell::Cell::get)
+}
+
+/// Merges `incoming` into `stored`, keeping the result in position order and joining the runs that
+/// touch.
+///
+/// Within a chunk one pass over both sides is what keeps this linear, since each side is sorted and
+/// disjoint. Adding the runs one at a time and rebuilding the array after each would cost the square
+/// of the runs a chunk yields: a chunk of one hundred thousand positions alternating between values
+/// and nulls yields fifty thousand runs, and re-merging after each is over a billion element moves
+/// for a single series.
+///
+/// One pass per chunk is still a pass over everything the series has stored so far, so it is not
+/// free once the *history* grows across chunks: a series split into `K` chunks of one position each,
+/// a hole apart, stores one more run per chunk and moves `K(K+1)/2` of them, with no label collision
+/// and no illegal input anywhere. A chunk that starts no earlier than the end of the last stored run
+/// only ever adds to the end — the stored runs are sorted and disjoint, so every one of them ends
+/// before anything this chunk holds — and it is appended instead, which is amortised constant per
+/// run. That is the shape a scan hands over as its positions move forward chunk after chunk. A chunk
+/// that starts before the last stored run can split a run or reach back past it, so it still costs
+/// the full merge.
+///
+/// Either way the result is sorted, disjoint, and has the runs that touch joined, which is what a
+/// look-up by [`Run::overlaps`] on the runs stored here relies on.
+pub(crate) fn absorb(stored: &mut Vec<Run>, incoming: &[Run]) {
+    if incoming.is_empty() {
+        return;
+    }
+    if stored
+        .last()
+        .is_none_or(|last| incoming[0].start >= last.end)
+    {
+        append_runs(stored, incoming);
+    } else {
+        merge_runs(stored, incoming);
+    }
+}
+
+/// Appends `incoming` to `stored`, all of which lies past the last stored run.
+///
+/// Nothing already stored has to be placed again: the two sides are each in position order and
+/// nothing `incoming` holds can reach back into what is stored, so its runs are pushed on the end and
+/// the vector's own growth is the only move — amortised over the runs appended. Where its first run
+/// touches the end of the last stored run the two are one run, as they are after a merge, so it is
+/// joined rather than left beside it.
+fn append_runs(stored: &mut Vec<Run>, incoming: &[Run]) {
+    let mut incoming = incoming;
+    if let Some(last) = stored.last_mut()
+        && let Some(first) = incoming.first()
+        && first.start <= last.end
+    {
+        last.end = last.end.max(first.end);
+        incoming = &incoming[1..];
+    }
+    stored.extend_from_slice(incoming);
+}
+
+/// Merges `incoming` into `stored`, where the two can interleave.
+///
+/// One pass is enough because each side is sorted and disjoint: whichever run starts first is
+/// placed, and it joins the run before it where the two touch.
+fn merge_runs(stored: &mut Vec<Run>, incoming: &[Run]) {
+    #[cfg(test)]
+    ABSORB_REBUILDS.with(|rebuilds| rebuilds.set(rebuilds.get() + 1));
+    let mut merged: Vec<Run> = Vec::with_capacity(stored.len() + incoming.len());
+    let (mut a, mut b) = (0, 0);
+    while a < stored.len() || b < incoming.len() {
+        let larger = match (stored.get(a), incoming.get(b)) {
+            (Some(left), Some(right)) if right.start < left.start => {
+                b += 1;
+                *right
+            }
+            (Some(left), _) => {
+                a += 1;
+                *left
+            }
+            (None, Some(right)) => {
+                b += 1;
+                *right
+            }
+            (None, None) => break,
+        };
+        match merged.last_mut() {
+            Some(last) if larger.start <= last.end => last.end = last.end.max(larger.end),
+            _ => merged.push(larger),
+        }
+    }
+    *stored = merged;
+}

src/query/src/promql/native_compile/tests.rs

@@ -13,17 +13,17 @@
 // limitations under the License.
 
 use std::sync::Arc;
-use std::time::{Duration, UNIX_EPOCH};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use datafusion::error::DataFusionError;
 use datafusion::execution::memory_pool::{GreedyMemoryPool, MemoryPool};
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use datatypes::arrow::array::Float64Array;
 use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
-use datatypes::series::MetricName;
 use datatypes::series::SeriesOrdering;
 use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
 use datatypes::series::samples::{FieldSpec, SampleRef, Samples, SeriesFields, SeriesSchema};
+use datatypes::series::{Label, MetricName};
 use promql::evaluated::EvaluatedChunk;
 use promql::map::MathOp;
 use promql::native::execute::{NativeExecutionState, NativeExecutor};
@@ -3138,11 +3138,25 @@ fn large_subquery_child_grids_compile_deferred() {
     assert_eq!(grid.2, range_ms);
 }
 
+/// The largest millisecond endpoint the platform can represent as a `SystemTime`.
+///
+/// The tests below need an endpoint close enough to `i64::MAX` milliseconds that a one-second
+/// rounding or a one-millisecond offset crosses the representable range. Unix-like platforms can
+/// hold that instant; Windows represents `SystemTime` as a 64-bit count of 100-nanosecond
+/// intervals since 1601, so the same instant does not fit and `UNIX_EPOCH + duration` panics. The
+/// behaviour under test is about arithmetic on the millisecond value, not about the platform's
+/// clock range, so a platform that cannot represent the instant skips instead of failing.
+fn extreme_millisecond_endpoint() -> Option<SystemTime> {
+    UNIX_EPOCH.checked_add(Duration::from_millis(i64::MAX as u64))
+}
+
 /// A sampled child endpoint can fit even when the unrounded read-window endpoint does not.
 #[test]
 fn a_subquery_accepts_an_in_range_sampled_endpoint() {
+    let Some(endpoint) = extreme_millisecond_endpoint() else {
+        return;
+    };
     let mut stmt = eval_stmt("rate(x[1s:1s] offset -1ms)");
-    let endpoint = UNIX_EPOCH + Duration::from_millis(i64::MAX as u64);
     stmt.start = endpoint;
     stmt.end = endpoint;
     stmt.interval = Duration::from_millis(1);
@@ -3161,6 +3175,9 @@ fn a_subquery_accepts_an_in_range_sampled_endpoint() {
 /// An anchored fold offset that cannot be represented as i64 is a plan error.
 #[test]
 fn a_subquery_with_an_overflowing_fold_offset_is_refused() {
+    let Some(endpoint) = extreme_millisecond_endpoint() else {
+        return;
+    };
     let mut stmt = eval_stmt("rate(x[5m:1m] @ 0)");
     let PromExpr::Call(call) = &mut stmt.expr else {
         unreachable!("query parses to a call");
@@ -3168,9 +3185,7 @@ fn a_subquery_with_an_overflowing_fold_offset_is_refused() {
     let PromExpr::Subquery(subquery) = &mut *call.args.args[0] else {
         unreachable!("call argument parses to a subquery");
     };
-    subquery.at = Some(AtModifier::At(
-        UNIX_EPOCH + Duration::from_millis(i64::MAX as u64),
-    ));
+    subquery.at = Some(AtModifier::At(endpoint));
     subquery.offset = Some(Offset::Neg(Duration::from_millis(1)));
 
     let error = compile(&stmt, raw_schema(), SeriesOrdering::KeyThenTime)
@@ -3724,6 +3739,529 @@ fn whole_vector_functions_bind_a_vector_operation() {
     }
 }
 
+/// One raw chunk of `metric`, keyed by `key`, closing the series.
+///
+/// `timestamps` are milliseconds and `values` one per timestamp, which is what a scan of the series
+/// hands the tree.
+fn agg_raw_chunk(
+    schema: &SeriesSchema,
+    key: &[u8],
+    metric: MetricName,
+    timestamps: &[i64],
+    values: &[f64],
+) -> RawSeriesChunk {
+    agg_raw_chunk_with_last(schema, key, metric, timestamps, values, true)
+}
+
+/// One raw chunk of `metric`, keyed by `key`, with an explicit series-ending flag.
+fn agg_raw_chunk_with_last(
+    schema: &SeriesSchema,
+    key: &[u8],
+    metric: MetricName,
+    timestamps: &[i64],
+    values: &[f64],
+    is_last: bool,
+) -> RawSeriesChunk {
+    RawSeriesChunk::try_new(
+        schema,
+        Buffer::from_slice_ref(key),
+        Some(metric),
+        RawTimestamps::try_new(ScalarBuffer::from(timestamps.to_vec())).unwrap(),
+        SeriesFields::one(Samples::from_float(Float64Array::from(values.to_vec()))),
+        is_last,
+    )
+    .unwrap()
+}
+
+/// `name` with the labels in `tags`, in the order given.
+fn agg_metric(name: &str, tags: &[(&str, &str)]) -> MetricName {
+    MetricName::try_new(
+        Some(name.into()),
+        tags.iter()
+            .map(|(name, value)| Label {
+                name: (*name).into(),
+                value: (*value).into(),
+            })
+            .collect(),
+    )
+    .unwrap()
+}
+
+/// What one result series of a native execution reports.
+type AggSeries = (
+    Option<String>,
+    Vec<(String, String)>,
+    Vec<(usize, Option<f64>)>,
+);
+
+/// Reads the outputs a native execution produced.
+fn aggregate_results(output: &[EvaluatedChunk]) -> Vec<AggSeries> {
+    let mut results: Vec<AggSeries> = output
+        .iter()
+        .map(|chunk| {
+            let mut tags: Vec<(String, String)> = chunk
+                .series
+                .metric
+                .tags()
+                .iter()
+                .map(|tag| (tag.name.to_string(), tag.value.to_string()))
+                .collect();
+            tags.sort();
+            let points = chunk
+                .series
+                .time
+                .positions()
+                .map(|(position, row)| {
+                    let sample = chunk.series.fields.as_slice()[0]
+                        .iter()
+                        .nth(row)
+                        .expect("the axis and the field are aligned");
+                    (
+                        position,
+                        match sample {
+                            SampleRef::Float(value) => Some(value),
+                            _ => None,
+                        },
+                    )
+                })
+                .collect();
+            (chunk.series.metric.name().map(str::to_string), tags, points)
+        })
+        .collect();
+    results.sort_by(|left, right| (&left.0, &left.1).cmp(&(&right.0, &right.1)));
+    results
+}
+
+/// Runs `tree` over `chunks`, which enter at the tree's only input, and reads the answer.
+///
+/// The answer is read the way a caller reads it: the whole input is pushed, the input is ended, and
+/// what the root reports is collected. Each series is reported by its identity and the value it
+/// decided at each absolute grid position.
+fn run_aggregate_tree(tree: &NativeTree, chunks: Vec<RawSeriesChunk>) -> Vec<AggSeries> {
+    let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 26));
+    let state =
+        NativeExecutionState::try_new(tree, pool, &ExecutionPlanMetricsSet::new(), 0).unwrap();
+    let mut execution = NativeExecutor::new(state);
+    for chunk in chunks {
+        execution
+            .push_raw(chunk)
+            .unwrap_or_else(|error| panic!("a source chunk was refused: {error}"));
+    }
+    let output = execution
+        .finish()
+        .unwrap_or_else(|error| panic!("the tree did not answer: {error}"));
+    aggregate_results(&output)
+}
+
+/// The 21 grid points of a statement over 100 seconds in 5-second steps.
+fn agg_points_0_to_100() -> Vec<usize> {
+    (0..21).collect()
+}
+
+/// The timestamps of the same 21 points, in milliseconds.
+fn agg_timestamps_0_to_100() -> Vec<i64> {
+    (0..21).map(|index| index * 5_000).collect()
+}
+
+/// `sum(x)` over two series: one unlabelled group, and every label of both inputs dropped.
+#[test]
+fn a_compiled_sum_drops_every_label_and_adds_its_group() {
+    let tree = compile(
+        &eval_stmt("sum(x)"),
+        raw_schema(),
+        SeriesOrdering::KeyThenTime,
+    )
+    .unwrap();
+    let schema = raw_schema();
+    let timestamps = agg_timestamps_0_to_100();
+    let chunks = vec![
+        agg_raw_chunk(
+            &schema,
+            b"a",
+            agg_metric("x", &[("job", "api"), ("instance", "a")]),
+            &timestamps,
+            &(1..=21).map(f64::from).collect::<Vec<_>>(),
+        ),
+        agg_raw_chunk(
+            &schema,
+            b"b",
+            agg_metric("x", &[("job", "db"), ("instance", "b")]),
+            &timestamps,
+            &(1..=21).map(|value| (2 * value) as f64).collect::<Vec<_>>(),
+        ),
+    ];
+
+    let output = run_aggregate_tree(&tree, chunks);
+    // `3` at the first point, three more at each of the twenty after it: the two members summed.
+    let values: Vec<(usize, Option<f64>)> = agg_points_0_to_100()
+        .into_iter()
+        .map(|position| (position, Some(3.0 * (position as f64 + 1.0))))
+        .collect();
+    assert_eq!(output, vec![(None, Vec::new(), values)]);
+}
+
+#[test]
+fn a_compiled_sum_aggregates_series_chunks_after_input_ends() {
+    let schema = raw_schema();
+    let timestamps = agg_timestamps_0_to_100();
+    let values: Vec<f64> = (1..=21).map(f64::from).collect();
+    let cases = [
+        ("sum(x)", None, Vec::new()),
+        (
+            "sum by(job)(x)",
+            None,
+            vec![("job".to_string(), "api".to_string())],
+        ),
+        (
+            "sum without(instance)(x)",
+            None,
+            vec![("job".to_string(), "api".to_string())],
+        ),
+    ];
+
+    for (query, name, tags) in cases {
+        let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime).unwrap();
+        let metric = agg_metric("x", &[("job", "api"), ("instance", "a")]);
+        let chunks = vec![
+            agg_raw_chunk_with_last(
+                &schema,
+                b"a",
+                metric.clone(),
+                &timestamps[..10],
+                &values[..10],
+                false,
+            ),
+            agg_raw_chunk_with_last(
+                &schema,
+                b"a",
+                metric,
+                &timestamps[10..],
+                &values[10..],
+                true,
+            ),
+        ];
+        let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 26));
+        let state =
+            NativeExecutionState::try_new(&tree, pool, &ExecutionPlanMetricsSet::new(), 0).unwrap();
+        let mut execution = NativeExecutor::new(state);
+        for chunk in chunks {
+            execution.push_raw(chunk).unwrap();
+            assert!(
+                execution.next_chunk().unwrap().is_none(),
+                "`{query}` produced output before the whole input ended"
+            );
+        }
+        let output = execution.finish().unwrap();
+        let results = aggregate_results(&output);
+        let expected_points = values
+            .iter()
+            .enumerate()
+            .map(|(position, value)| (position, Some(*value)))
+            .collect();
+        assert_eq!(
+            results,
+            vec![(name.map(str::to_string), tags, expected_points)],
+            "`{query}`"
+        );
+    }
+}
+
+/// Empty source input is not a member of an aggregate's implicit group.
+#[test]
+fn a_compiled_aggregate_over_empty_input_produces_no_groups() {
+    for query in ["sum(x)", "sum by(job)(x)", "sum without(instance)(x)"] {
+        let tree = compile(&eval_stmt(query), raw_schema(), SeriesOrdering::KeyThenTime).unwrap();
+        assert_eq!(
+            run_aggregate_tree(&tree, Vec::new()),
+            Vec::new(),
+            "`{query}` invented a group for empty input"
+        );
+    }
+}
+
+/// `sum by(job)(x)` over five series: two groups, each the sum of its own members.
+///
+/// The members cover different points, so each group is a sum over a subset that changes from point
+/// to point: one series holds the whole grid, one holds a single sample between two grid points, and
+/// one stops a whole lookback before the grid does. No last sample sits exactly one lookback behind
+/// a grid point, so which points each member covers does not depend on how the boundary is read.
+#[test]
+fn a_compiled_sum_by_groups_its_members_and_sums_each_group() {
+    let tree = compile(
+        &eval_stmt("sum by(job)(x)"),
+        raw_schema(),
+        SeriesOrdering::KeyThenTime,
+    )
+    .unwrap();
+    let schema = raw_schema();
+    let all = agg_timestamps_0_to_100();
+    let chunks = vec![
+        // The whole grid, holding one.
+        agg_raw_chunk(
+            &schema,
+            b"a",
+            agg_metric("x", &[("job", "api"), ("instance", "a")]),
+            &all,
+            &[1.0; 21],
+        ),
+        // One sample halfway between grid points 0 and 1, so only grid point 1 selects it.
+        agg_raw_chunk(
+            &schema,
+            b"b",
+            agg_metric("x", &[("job", "api"), ("instance", "b")]),
+            &[2_500],
+            &[2.0],
+        ),
+        // Four samples between grid points, covering grid points 1 to 4.
+        agg_raw_chunk(
+            &schema,
+            b"e",
+            agg_metric("x", &[("job", "api"), ("instance", "e")]),
+            &[2_400, 7_400, 12_400, 17_400],
+            &[5.0; 4],
+        ),
+        // The whole grid, holding ten.
+        agg_raw_chunk(
+            &schema,
+            b"c",
+            agg_metric("x", &[("job", "db"), ("instance", "c")]),
+            &all,
+            &[10.0; 21],
+        ),
+        // One sample halfway between grid points 0 and 1, holding twenty.
+        agg_raw_chunk(
+            &schema,
+            b"d",
+            agg_metric("x", &[("job", "db"), ("instance", "d")]),
+            &[2_500],
+            &[20.0],
+        ),
+    ];
+
+    let output = run_aggregate_tree(&tree, chunks);
+    // `api` is `a` alone at points 0 and 5 to 20, and every member at the points they cover:
+    // point 1 is `1 + 2 + 5`, points 2 to 4 are `1 + 5`.
+    let api: Vec<(usize, Option<f64>)> = agg_points_0_to_100()
+        .into_iter()
+        .map(|position| {
+            let value = match position {
+                1 => 8.0,
+                2..=4 => 6.0,
+                _ => 1.0,
+            };
+            (position, Some(value))
+        })
+        .collect();
+    // `db` is the same shape over its own members: point 1 is `10 + 20`, the rest is `c` alone.
+    let db: Vec<(usize, Option<f64>)> = agg_points_0_to_100()
+        .into_iter()
+        .map(|position| (position, Some(if position == 1 { 30.0 } else { 10.0 })))
+        .collect();
+    assert_eq!(
+        output,
+        vec![
+            (None, vec![("job".to_string(), "api".to_string())], api),
+            (None, vec![("job".to_string(), "db".to_string())], db),
+        ]
+    );
+}
+
+/// `sum without(instance)(x)`: `instance` goes, `job` and `zone` stay, and so does `__name__` going.
+#[test]
+fn a_compiled_sum_without_drops_the_named_label_and_the_name() {
+    let tree = compile(
+        &eval_stmt("sum without(instance)(x)"),
+        raw_schema(),
+        SeriesOrdering::KeyThenTime,
+    )
+    .unwrap();
+    let schema = raw_schema();
+    let timestamps = agg_timestamps_0_to_100();
+    let chunks = vec![
+        agg_raw_chunk(
+            &schema,
+            b"a",
+            agg_metric("x", &[("job", "api"), ("instance", "a"), ("zone", "west")]),
+            &timestamps,
+            &[1.0; 21],
+        ),
+        agg_raw_chunk(
+            &schema,
+            b"b",
+            agg_metric("x", &[("job", "api"), ("instance", "b"), ("zone", "east")]),
+            &timestamps,
+            &[2.0; 21],
+        ),
+    ];
+
+    let output = run_aggregate_tree(&tree, chunks);
+    let values_of = |value: f64| -> Vec<(usize, Option<f64>)> {
+        agg_points_0_to_100()
+            .into_iter()
+            .map(|position| (position, Some(value)))
+            .collect()
+    };
+    // Two groups, because `zone` is not the dropped label: the two series differ by more than the
+    // label the grouping removes.
+    assert_eq!(
+        output,
+        vec![
+            (
+                None,
+                vec![
+                    ("job".to_string(), "api".to_string()),
+                    ("zone".to_string(), "east".to_string()),
+                ],
+                values_of(2.0),
+            ),
+            (
+                None,
+                vec![
+                    ("job".to_string(), "api".to_string()),
+                    ("zone".to_string(), "west".to_string()),
+                ],
+                values_of(1.0),
+            ),
+        ]
+    );
+}
+
+/// `sum by(job)(rate(x[5m]))`: an aggregate over a window, mid-chain.
+///
+/// Every series is a counter rising by one per five seconds, and every grid point's window ends on a
+/// sample and begins five seconds before one, so the extrapolated rate is the same at every point:
+/// `59 / 295 = 0.2` per second, which the double arithmetic lands on the literals below.
+#[test]
+fn a_compiled_sum_over_a_rate_groups_the_window_it_reads() {
+    let statement = EvalStmt {
+        expr: promql_parser::parser::parse("sum by(job)(rate(x[5m]))").unwrap(),
+        start: UNIX_EPOCH + Duration::from_secs(300),
+        end: UNIX_EPOCH + Duration::from_secs(400),
+        interval: Duration::from_secs(5),
+        lookback_delta: Duration::from_secs(5),
+    };
+    let tree = compile(&statement, raw_schema(), SeriesOrdering::KeyThenTime).unwrap();
+    // The window reads 300 seconds, so the samples start a full window before the first grid point.
+    let schema = raw_schema();
+    let timestamps: Vec<i64> = (0..=80).map(|index| index * 5_000).collect();
+    let values: Vec<f64> = (0..=80).map(|index| f64::from(index) + 1.0).collect();
+    let chunks = vec![
+        agg_raw_chunk(
+            &schema,
+            b"a",
+            agg_metric("x", &[("job", "api"), ("instance", "a")]),
+            &timestamps,
+            &values,
+        ),
+        agg_raw_chunk(
+            &schema,
+            b"b",
+            agg_metric("x", &[("job", "api"), ("instance", "b")]),
+            &timestamps,
+            &values,
+        ),
+        agg_raw_chunk(
+            &schema,
+            b"c",
+            agg_metric("x", &[("job", "db"), ("instance", "c")]),
+            &timestamps,
+            &values,
+        ),
+    ];
+
+    let output = run_aggregate_tree(&tree, chunks);
+    let at_every_point = |value: f64| -> Vec<(usize, Option<f64>)> {
+        (0..21).map(|position| (position, Some(value))).collect()
+    };
+    // One `rate` is `0.19999999999999998`, and the pair of them add to `0.39999999999999997`.
+    assert_eq!(
+        output,
+        vec![
+            (
+                None,
+                vec![("job".to_string(), "api".to_string())],
+                at_every_point(0.39999999999999997),
+            ),
+            (
+                None,
+                vec![("job".to_string(), "db".to_string())],
+                at_every_point(0.19999999999999998),
+            ),
+        ]
+    );
+}
+
+/// Splitting one counter's samples does not change the rate aggregate or make it answer early.
+#[test]
+fn a_compiled_sum_over_a_rate_aggregates_series_chunks_after_input_ends() {
+    let statement = EvalStmt {
+        expr: promql_parser::parser::parse("sum by(job)(rate(x[5m]))").unwrap(),
+        start: UNIX_EPOCH + Duration::from_secs(300),
+        end: UNIX_EPOCH + Duration::from_secs(400),
+        interval: Duration::from_secs(5),
+        lookback_delta: Duration::from_secs(5),
+    };
+    let tree = compile(&statement, raw_schema(), SeriesOrdering::KeyThenTime).unwrap();
+    let schema = raw_schema();
+    let timestamps: Vec<i64> = (0..=80).map(|index| index * 5_000).collect();
+    let values: Vec<f64> = (0..=80).map(|index| f64::from(index) + 1.0).collect();
+    let metric = agg_metric("x", &[("job", "api"), ("instance", "a")]);
+    let chunks = [
+        agg_raw_chunk_with_last(
+            &schema,
+            b"a",
+            metric.clone(),
+            &timestamps[..41],
+            &values[..41],
+            false,
+        ),
+        agg_raw_chunk_with_last(
+            &schema,
+            b"a",
+            metric,
+            &timestamps[41..],
+            &values[41..],
+            true,
+        ),
+    ];
+    let pool: Arc<dyn MemoryPool> = Arc::new(GreedyMemoryPool::new(1 << 26));
+    let state =
+        NativeExecutionState::try_new(&tree, pool, &ExecutionPlanMetricsSet::new(), 0).unwrap();
+    let mut execution = NativeExecutor::new(state);
+    for chunk in chunks {
+        execution.push_raw(chunk).unwrap();
+        assert!(
+            execution.next_chunk().unwrap().is_none(),
+            "the rate aggregate produced output before the whole input ended"
+        );
+    }
+    let output = execution.finish().unwrap();
+    assert_eq!(
+        aggregate_results(&output),
+        vec![(
+            None,
+            vec![("job".to_string(), "api".to_string())],
+            (0..21)
+                .map(|position| (position, Some(0.19999999999999998)))
+                .collect(),
+        )]
+    );
+}
+
+/// Empty source input is not a member of this aggregate's implicit group, with or without a window.
+#[test]
+fn a_compiled_sum_over_a_rate_with_empty_input_produces_no_groups() {
+    let statement = EvalStmt {
+        expr: promql_parser::parser::parse("sum by(job)(rate(x[5m]))").unwrap(),
+        start: UNIX_EPOCH + Duration::from_secs(300),
+        end: UNIX_EPOCH + Duration::from_secs(400),
+        interval: Duration::from_secs(5),
+        lookback_delta: Duration::from_secs(5),
+    };
+    let tree = compile(&statement, raw_schema(), SeriesOrdering::KeyThenTime).unwrap();
+    assert_eq!(run_aggregate_tree(&tree, Vec::new()), Vec::new());
+}
+
 /// `label_replace` and `label_join` bind their label change into a map over the vector they are
 /// given, with the strings the call wrote.
 #[test]