perf(promql): append a forward-moving chunk to the emitted history

`absorb` rebuilt the whole stored run array on every non-empty chunk. Legal input
made that quadratic across chunks with no label collision: one series in `K` chunks,
chunk `i` occupying only grid position `2*i`, stores one more run per chunk and moves
`K(K+1)/2` of them -- about five billion for a hundred thousand rows.

A chunk that starts at or past the end of the last stored run only ever adds to the
end, so its runs are appended and the boundary run is joined where it touches. One
that can interleave or reach backwards still takes the full one-pass merge, which can
split a run. Both leave the runs sorted, disjoint and joined, which is what
`Emitted::occupy` looks a position up by.

The test records the chunks through `Emitted::occupy` itself and reads the history
back from it, covering the forward case, the touching boundary, a hole that has to
join both neighbours, and a position before everything stored.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>

src/promql/src/native/execute.rs

@@ -709,17 +709,83 @@ fn occupied_runs(field: &Samples, time: &TimeAxis) -> Vec<Run> {
     runs
 }
 
+// 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.
 ///
-/// Both sides are sorted and disjoint, so one pass merges them. Adding the runs one at a time and
-/// rebuilding the whole 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.
+/// 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() {

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

@@ -35,7 +35,7 @@ use datatypes::series::{Label, MetricName, SeriesOrdering};
 
 use super::{
     ChargedAllocations, Emitted, History, NativeChunk, NativeExecutionState, NativeExecutor,
-    NodeBuffer, NodeRuntime, QueuedCharge, Run, label_bytes, map_chunk, raw_bytes,
+    NodeBuffer, NodeRuntime, QueuedCharge, Run, absorb_rebuilds, label_bytes, map_chunk, raw_bytes,
 };
 use crate::evaluated::EvaluatedChunk;
 use crate::grid_series::{LabelSet, label_set};
@@ -2511,3 +2511,161 @@ fn a_history_entry_is_charged_once_for_its_header() {
         "the table's capacity is where a label-set slot and one `History` per entry are charged",
     );
 }
+
+/// A sparse series that only ever moves forward is appended to its history, not merged into it again
+/// chunk after chunk.
+///
+/// One legal series arrives as eight chunks of one position each, every position a hole apart from
+/// the last: no label set is reached twice, no position of a field is occupied twice, and this is
+/// what a scan produces as its positions advance. The stored runs still grow by one per chunk, so
+/// rebuilding the array after each chunk moves `K(K+1)/2` runs to hold `K` rows of real data, which
+/// is the cost this asserts away.
+///
+/// The chunks go through [`super::Emitted::occupy`], the helper the runtime records an output with,
+/// so what is checked is the history that helper keeps. Three shapes are pushed: chunks landing past
+/// every stored run, which have to be appended to; a chunk starting where the last stored run ends,
+/// which has to be joined to it rather than left beside it; and chunks landing before what is
+/// stored, which cannot be appended to and have to go through the full merge while still leaving the
+/// history sorted and disjoint.
+#[test]
+fn a_sparse_series_that_moves_forward_is_appended_to_its_history() {
+    let map_grid = grid(0, 60_000, 21);
+    let schema = evaluated_schema(Arc::clone(&map_grid));
+    let path = NodePath::root();
+    let mut emitted = Emitted::default();
+
+    // One row at `position`, which is one run of one position: the same series every time, since a
+    // map dropping the name leaves them all one label set.
+    let push = |emitted: &mut Emitted, position: usize| {
+        let chunk = sparse_chunk(
+            &schema,
+            named_metric("a", "x"),
+            &[(position, 1)],
+            vec![-1.0],
+            false,
+        );
+        emitted
+            .occupy(&path, &chunk)
+            .unwrap_or_else(|error| panic!("position {position} was refused: {error}"));
+    };
+    // The runs stored under the one label set, which is what `absorb` keeps in position order.
+    let stored = |emitted: &Emitted| -> Vec<(usize, usize)> {
+        emitted
+            .sets
+            .values()
+            .next()
+            .expect("the chunk was recorded under one label set")
+            .fields
+            .first()
+            .expect("the series has one field")
+            .iter()
+            .map(|run| (run.start, run.end))
+            .collect()
+    };
+    let ordered = |runs: &[(usize, usize)]| {
+        runs.iter().all(|(start, end)| start < end)
+            && runs.windows(2).all(|pair| pair[0].1 < pair[1].0)
+    };
+
+    // Eight chunks, one position each, a hole between them.
+    let positions = [4, 6, 8, 10, 12, 14, 16, 18];
+    let before = absorb_rebuilds();
+    for position in positions {
+        push(&mut emitted, position);
+    }
+    assert_eq!(
+        stored(&emitted),
+        vec![
+            (4, 5),
+            (6, 7),
+            (8, 9),
+            (10, 11),
+            (12, 13),
+            (14, 15),
+            (16, 17),
+            (18, 19),
+        ],
+        "a sparse series was not stored as the one run each of its chunks occupies"
+    );
+    assert_eq!(
+        absorb_rebuilds() - before,
+        0,
+        "the {} chunks of a series moving forward rebuilt its history instead of appending to it",
+        positions.len()
+    );
+
+    // A chunk that starts where the last stored run ends touches it, and the two are one run: the
+    // fast path appends, and the boundary is the part it has to join.
+    let before = absorb_rebuilds();
+    push(&mut emitted, 19);
+    assert_eq!(
+        stored(&emitted),
+        vec![
+            (4, 5),
+            (6, 7),
+            (8, 9),
+            (10, 11),
+            (12, 13),
+            (14, 15),
+            (16, 17),
+            (18, 20),
+        ],
+        "a chunk touching the last stored run was left beside it rather than joined to it"
+    );
+    assert_eq!(
+        absorb_rebuilds() - before,
+        0,
+        "a chunk touching the last stored run was merged rather than appended to it"
+    );
+
+    // A chunk landing in a hole before the last stored run cannot be appended to: it joins the runs
+    // on both sides of it, so it goes through the full merge, which has to join them all the same.
+    let before = absorb_rebuilds();
+    push(&mut emitted, 5);
+    assert_eq!(
+        stored(&emitted),
+        vec![
+            (4, 7),
+            (8, 9),
+            (10, 11),
+            (12, 13),
+            (14, 15),
+            (16, 17),
+            (18, 20),
+        ],
+        "a chunk in a hole did not join the runs on either side of it"
+    );
+    assert_eq!(
+        absorb_rebuilds() - before,
+        1,
+        "a chunk landing before the last stored run was appended rather than merged"
+    );
+
+    // A chunk before every stored run is the other shape the fast path cannot take.
+    let before = absorb_rebuilds();
+    push(&mut emitted, 1);
+    let runs = stored(&emitted);
+    assert_eq!(
+        runs,
+        vec![
+            (1, 2),
+            (4, 7),
+            (8, 9),
+            (10, 11),
+            (12, 13),
+            (14, 15),
+            (16, 17),
+            (18, 20),
+        ],
+        "a chunk before every stored run did not land in front of them"
+    );
+    assert_eq!(
+        absorb_rebuilds() - before,
+        1,
+        "a chunk landing before every stored run was appended rather than merged"
+    );
+    assert!(
+        ordered(&runs),
+        "the merged history is not sorted, disjoint and joined: {runs:?}"
+    );
+}