diff --git a/src/promql/src/native/execute.rs b/src/promql/src/native/execute.rs index 6578ab3b7ec..9500be2f82b 100644 --- a/src/promql/src/native/execute.rs +++ b/src/promql/src/native/execute.rs @@ -709,17 +709,83 @@ fn occupied_runs(field: &Samples, time: &TimeAxis) -> Vec { 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 = 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, 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, 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, incoming: &[Run]) { + #[cfg(test)] + ABSORB_REBUILDS.with(|rebuilds| rebuilds.set(rebuilds.get() + 1)); let mut merged: Vec = Vec::with_capacity(stored.len() + incoming.len()); let (mut a, mut b) = (0, 0); while a < stored.len() || b < incoming.len() { diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index 3c93cdc09ad..56bc886014b 100644 --- a/src/promql/src/native/execute/tests.rs +++ b/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:?}" + ); +}