diff --git a/src/promql/src/native/display.rs b/src/promql/src/native/display.rs index 2124f692aec..fd22e771db0 100644 --- a/src/promql/src/native/display.rs +++ b/src/promql/src/native/display.rs @@ -28,7 +28,7 @@ use datatypes::series::{MetricName, SeriesOrdering}; use crate::map::MathOp; use crate::native::metrics::{self, NodePath}; use crate::native::plan::{ - AggregateKind, AggregateStage, BoundSelect, BoundWindow, Grouping, NativeExpr, NativeNode, + AggregateKind, AggregateStage, BoundVectorOp, BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, }; use crate::range_eval::RangeFunction; @@ -103,15 +103,15 @@ fn describe(node: &NativeNode) -> String { let ops: Vec = op.ops.iter().map(math_op).collect(); ops.join(", ") } - NativeExpr::Select { op, .. } => match op { - BoundSelect::TopK { k, top, group } => format!( + NativeExpr::Vector { op, .. } => match op { + BoundVectorOp::TopK { k, top, group } => format!( "{} {k} {}", if *top { "topk" } else { "bottomk" }, grouping(group) ), - BoundSelect::LimitK { k, group } => format!("limitk {k} {}", grouping(group)), - BoundSelect::LimitRatio { ratio } => format!("limit_ratio {ratio}"), - BoundSelect::Sort(order) => match order { + BoundVectorOp::LimitK { k, group } => format!("limitk {k} {}", grouping(group)), + BoundVectorOp::LimitRatio { ratio } => format!("limit_ratio {ratio}"), + BoundVectorOp::Sort(order) => match order { SortOrder::Asc => "sort".to_string(), SortOrder::Desc => "sort_desc".to_string(), SortOrder::ByLabel(labels) => format!("sort_by_label [{}]", label_list(labels)), @@ -119,7 +119,7 @@ fn describe(node: &NativeNode) -> String { format!("sort_by_label_desc [{}]", label_list(labels)) } }, - BoundSelect::Absent { labels } => format!("absent {{{}}}", label_pairs(labels)), + BoundVectorOp::Absent { labels } => format!("absent {{{}}}", label_pairs(labels)), }, NativeExpr::Aggregate { op, .. } => format!( "{} {} {}", @@ -298,7 +298,7 @@ fn kind(expr: &NativeExpr) -> &'static str { NativeExpr::Window { .. } => "Window", NativeExpr::Map { .. } => "Map", NativeExpr::Aggregate { .. } => "Aggregate", - NativeExpr::Select { .. } => "Select", + NativeExpr::Vector { .. } => "Vector", NativeExpr::Binary { .. } => "Binary", } } diff --git a/src/promql/src/native/display/tests.rs b/src/promql/src/native/display/tests.rs index d259ce28eee..5b088f560d8 100644 --- a/src/promql/src/native/display/tests.rs +++ b/src/promql/src/native/display/tests.rs @@ -20,7 +20,7 @@ use datatypes::series::{Label, MetricName, SeriesOrdering}; use super::super::plan::{ AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar, - BoundSelect, BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, + BoundVectorOp, BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, }; use crate::map::MathOp; use crate::range_eval::{RangeFunction, RangeWindow}; @@ -396,9 +396,9 @@ fn a_without_grouping_does_not_read_as_by() { ); } -/// A selection names its operator, constant and grouping. +/// A vector operation names its operator, constant and grouping. #[test] -fn a_selection_names_its_binding() { +fn a_vector_operation_names_its_binding() { let out = grid(0, 60_000, 4); let window = window_over( raw_input(), @@ -410,30 +410,30 @@ fn a_selection_names_its_binding() { ); let lines = [ ( - BoundSelect::TopK { + BoundVectorOp::TopK { k: 3.0, top: false, group: Grouping::Without(Arc::new(["job".into()])), }, - " root Select bottomk 3 without [job]", + " root Vector bottomk 3 without [job]", ), ( - BoundSelect::LimitK { + BoundVectorOp::LimitK { k: 2.0, group: Grouping::By(Arc::new([])), }, - " root Select limitk 2 by []", + " root Vector limitk 2 by []", ), ( - BoundSelect::LimitRatio { ratio: -0.5 }, - " root Select limit_ratio -0.5", + BoundVectorOp::LimitRatio { ratio: -0.5 }, + " root Vector limit_ratio -0.5", ), ( - BoundSelect::Sort(SortOrder::ByLabelDesc(vec!["job".into(), "le".into()])), - " root Select sort_by_label_desc [job, le]", + BoundVectorOp::Sort(SortOrder::ByLabelDesc(vec!["job".into(), "le".into()])), + " root Vector sort_by_label_desc [job, le]", ), ( - BoundSelect::Absent { + BoundVectorOp::Absent { labels: MetricName::try_new( None, vec![Label { @@ -443,12 +443,12 @@ fn a_selection_names_its_binding() { ) .unwrap(), }, - r#" root Select absent {job="x"}"#, + r#" root Vector absent {job="x"}"#, ), ]; for (op, line) in lines { let tree = NativeTree::try_new(NativeNode { - expr: NativeExpr::Select { + expr: NativeExpr::Vector { input: Box::new(window.clone()), op, }, diff --git a/src/promql/src/native/execute.rs b/src/promql/src/native/execute.rs index 66b17c63f97..b4147bdab6e 100644 --- a/src/promql/src/native/execute.rs +++ b/src/promql/src/native/execute.rs @@ -40,7 +40,7 @@ //! operands and runs the matching kernel once the input has ended, because the operation pairs two //! whole vectors point by point and nothing is decided for a point until both sides are complete. //! The chunks of one logical series are assembled into the single series the kernel matches before it -//! runs, so a series that arrived in many chunks is one candidate, not one per chunk. A selection +//! runs, so a series that arrived in many chunks is one candidate, not one per chunk. A vector operation //! (`topk`, `limitk`, `limit_ratio`) reads its one input the same way, because which samples it //! keeps at a point depends on every series present there. //! @@ -99,7 +99,7 @@ use crate::map::{MathOp, drop_metric_name, map_math}; use crate::native::batch::EvaluatedChunk; use crate::native::metrics::{NodeMetrics, NodePath}; use crate::native::plan::{ - BoundInstant, BoundSelect, Grouping, NativeExpr, NativeNode, NativeTree, + BoundInstant, BoundVectorOp, Grouping, NativeExpr, NativeNode, NativeTree, }; use crate::range_eval::RangeEvaluator; use crate::select::{self, SortOrder}; @@ -385,7 +385,7 @@ enum NodeTag { /// Matches two evaluated vectors and applies one operation to every matched pair of series. Binary, /// Keeps some samples of its whole input vector at each grid point. - Select, + Vector, } impl NodeTag { @@ -397,7 +397,7 @@ impl NodeTag { Self::Window => "window", Self::Map => "map", Self::Binary => "binary", - Self::Select => "selection", + Self::Vector => "vector operation", } } } @@ -1252,7 +1252,7 @@ impl InstantRuntime { } /// An input a node reads whole before it computes -- one operand of a binary node, or the input of a -/// selection -- chunk by chunk, until it ends. +/// vector operation -- chunk by chunk, until it ends. /// /// What is kept is the series each chunk carries, not a concatenation of them: a chunk is retained /// as it arrives and nothing is copied on the way in, because the chunks of a series can share an @@ -1591,22 +1591,22 @@ fn side_name(index: usize) -> &'static str { enum Reader<'a> { /// The operand at `side` of the binary node at `path`. Operand { path: &'a NodePath, side: usize }, - /// The input of the selection at the path. - Select(&'a NodePath), + /// The input of the vector operation at the path. + Vector(&'a NodePath), } impl Reader<'_> { fn owner(self) -> String { match self { Self::Operand { path, .. } => format!("the binary node at {path}"), - Self::Select(path) => format!("the selection at {path}"), + Self::Vector(path) => format!("the vector operation at {path}"), } } fn input(self) -> String { match self { Self::Operand { side, .. } => format!("its {} operand", side_name(side)), - Self::Select(_) => "its input".to_string(), + Self::Vector(_) => "its input".to_string(), } } } @@ -1776,27 +1776,31 @@ impl BinaryRuntime { } } -/// What a selection node holds: its whole input until it ends. +/// What a vector node holds: its whole input until it ends. /// /// Which samples `topk` or `limitk` keeps at a point depends on every series present there, so, /// like a binary node, this node retains every chunk its input hands over, assembles the chunks of /// each logical series, and runs its kernel once, after the input has ended. The results are queued /// in the node's own output queue and handed over from there. -struct SelectRuntime { - op: BoundSelect, +struct VectorRuntime { + op: BoundVectorOp, /// The layout of the input, which is also what the node produces. schema: Arc, input: RetainedInput, + /// Whether the input's pieces of one label set are merged before the kernel reads them, which + /// `limitk` needs when its input can hold them. + merge_input: bool, /// Whether the kernel has run. It runs at most once: the input it read is gone afterwards. computed: bool, } -impl SelectRuntime { - fn new(op: BoundSelect, schema: Arc) -> Self { +impl VectorRuntime { + fn new(op: BoundVectorOp, schema: Arc) -> Self { Self { op, schema, input: RetainedInput::default(), + merge_input: false, computed: false, } } @@ -1805,44 +1809,47 @@ impl SelectRuntime { /// already ran. /// /// The reservation covers the retained chunks, the assembled series and a rough estimate of the kernel's workspace and of the series it rebuilds - /// ([`selection_workspace_bytes`]). The kernels do not report their workspace as they allocate, + /// ([`vector_workspace_bytes`]). The kernels do not report their workspace as they allocate, /// so the estimate is reserved once, before the kernel runs. fn compute(&mut self, path: &NodePath, buffer: &mut NodeBuffer) -> Result>> { if self.computed { return Ok(None); } - self.input.ensure_closed(Reader::Select(path))?; + self.input.ensure_closed(Reader::Vector(path))?; let retained = self.input.held; let assembly = self .input .assembly_bytes(&self.schema) - .ok_or_else(|| select_overflow(path))?; + .ok_or_else(|| vector_overflow(path))?; buffer.grow(assembly)?; - let assembled = match self.input.assemble(&self.schema, Reader::Select(path)) { + let assembled = match self.input.assemble(&self.schema, Reader::Vector(path)) { Ok(assembled) => assembled, Err(error) => { buffer.resize(retained)?; return Err(error); } }; - let mut assembled = assembled; - if matches!(self.op, BoundSelect::LimitK { .. }) { - // `limitk` keeps the first series in input order. Prometheus reads them in label order - // once a query spans its head and a block, while a scan here need not order distinct - // series at all, so label order is what makes the choice the same on every run. The - // sort is stable, so the pieces of one label set stay in the order they arrived and, - // holding disjoint points, are read as the one series they make up. - assembled.sort_by(|a, b| select::compare_labels(&a.metric, &b.metric)); - } - buffer.grow(selection_workspace_bytes( - &self.op, - &self.schema, - &assembled, - ))?; + // `limitk` keeps the first series in the order its input hands them over, as Prometheus + // takes the order of the vector it is given -- a sort below it decides that order. Pieces + // of one label set are one series there, placed where the label set first arrived, so + // they are merged first; ranked apart, a later piece would be counted after the series + // that arrived between them. + let assembled = if self.merge_input { + let rows = assembled.iter().map(|series| series.time.len()).sum(); + buffer.grow(merge_workspace_bytes( + rows, + assembled.len(), + self.schema.fields.len(), + ))?; + merge_same_labelset(&self.schema, assembled)? + } else { + assembled + }; + buffer.grow(vector_workspace_bytes(&self.op, &self.schema, &assembled))?; let metrics = Arc::clone(&buffer.metrics); let output = { let _timer = metrics.timer(); - select_series(&self.op, &self.schema, assembled)? + vector_series(&self.op, &self.schema, assembled)? }; self.input.release(); self.computed = true; @@ -1850,17 +1857,17 @@ impl SelectRuntime { } } -/// The error for a selection whose reservation does not fit an address space. -fn select_overflow(path: &NodePath) -> DataFusionError { +/// The error for a vector operation whose reservation does not fit an address space. +fn vector_overflow(path: &NodePath) -> DataFusionError { DataFusionError::Execution(format!( - "the input of the selection at {path} does not fit one address space: the bytes it would \ + "the input of the vector operation at {path} does not fit one address space: the bytes it would \ reserve overflow" )) } /// Runs the kernel `op` binds over the whole input. -fn select_series( - op: &BoundSelect, +fn vector_series( + op: &BoundVectorOp, schema: &SeriesSchema, series: Vec, ) -> Result> { @@ -1869,21 +1876,21 @@ fn select_series( Grouping::Without(labels) => select::Grouping::Without(labels.to_vec()), }; match op { - BoundSelect::TopK { k, top, group } => { + BoundVectorOp::TopK { k, top, group } => { select::topk(*k, *top, &grouping(group), schema, &series) } - BoundSelect::LimitK { k, group } => select::limitk(*k, &grouping(group), schema, &series), - BoundSelect::LimitRatio { ratio } => select::limit_ratio(*ratio, schema, &series), + BoundVectorOp::LimitK { k, group } => select::limitk(*k, &grouping(group), schema, &series), + BoundVectorOp::LimitRatio { ratio } => select::limit_ratio(*ratio, schema, &series), // A sort hands back the series it was given, rebuilding only those `sort` and `sort_desc` // drop histogram samples from. - BoundSelect::Sort(order) => select::sort_instant(order, schema, series), - BoundSelect::Absent { labels } => Ok(select::absent(schema, &series, labels.clone())? + BoundVectorOp::Sort(order) => select::sort_instant(order, schema, series), + BoundVectorOp::Absent { labels } => Ok(select::absent(schema, &series, labels.clone())? .into_iter() .collect()), } } -/// A rough estimate of what the selection `op` allocates over `series`, beside the input it reads. +/// A rough estimate of what the vector operation `op` allocates over `series`, beside the input it reads. /// Like [`merge_workspace_bytes`] it is not an allocation bound: vector slack and the hash table's /// own layout are not measured. /// @@ -1895,10 +1902,10 @@ fn select_series( /// - A sort: the value each series is keyed by, and the floats `sort` and `sort_desc` rebuild a /// series into when it holds histograms; the label sorts move the series they were given. /// - `absent`: a presence flag and a result value per grid point and field. -fn selection_workspace_bytes(op: &BoundSelect, schema: &SeriesSchema, series: &[Series]) -> usize { +fn vector_workspace_bytes(op: &BoundVectorOp, schema: &SeriesSchema, series: &[Series]) -> usize { let fields = schema.fields.len(); match op { - BoundSelect::Sort(order) => { + BoundVectorOp::Sort(order) => { let rebuilds = matches!(order, SortOrder::Asc | SortOrder::Desc); return series.iter().fold(0usize, |total, item| { let floats = if rebuilds @@ -1916,14 +1923,16 @@ fn selection_workspace_bytes(op: &BoundSelect, schema: &SeriesSchema, series: &[ .saturating_add(floats) }); } - BoundSelect::Absent { .. } => { + BoundVectorOp::Absent { .. } => { let points = schema .eval_grid .as_ref() .map_or(0, |grid| grid.timestamps().len()); return points.saturating_mul(fields.saturating_mul(1 + std::mem::size_of::())); } - BoundSelect::TopK { .. } | BoundSelect::LimitK { .. } | BoundSelect::LimitRatio { .. } => {} + BoundVectorOp::TopK { .. } + | BoundVectorOp::LimitK { .. } + | BoundVectorOp::LimitRatio { .. } => {} } let row = std::mem::size_of::<(usize, usize)>() + std::mem::size_of::<(usize, usize, f64)>() @@ -2100,11 +2109,15 @@ struct NodeRuntime { /// The chunks this node holds for its consumer, each with the estimate it was queued with. /// /// An input node's queue is filled by the caller, a window's by the chunks it computed while - /// flushing its collected input, and a binary or selection node's by the results its kernel + /// flushing its collected input, and a binary or vector node's by the results its kernel /// produced; no node /// uses it for both. A chunk is measured once, as it enters, and the figure it was queued with /// is given back as it leaves, so neither the queue nor the handover walks it. queued: VecDeque<(NativeChunk, usize)>, + /// The histogram tables queued results share with each other, charged to the queue as a whole + /// rather than to one result: a table stays alive while any result referencing it is queued, + /// so its charge is given back when the queue empties, not when the first result leaves. + queued_tables: usize, /// What an instant node holds: the chunks of the series it is selecting, and, while it gathers /// that series, the workspace the gather reserved for. `None` for every other kind of node. instant: Option, @@ -2115,8 +2128,8 @@ struct NodeRuntime { map: Option, /// What a binary node holds. binary: Option, - /// What a selection node holds. - select: Option, + /// What a vector node holds. + vector: Option, /// The layout the caller's chunks of an input node are checked against. /// /// A tree with a binary node reads two inputs, and each of them is bound to its own layout, so @@ -2166,7 +2179,7 @@ impl NodeRuntime { /// Hands over the next queued chunk, if there is one, giving back what it was queued with. /// /// It is the one way a chunk leaves this node's queue: an input node hands over what the caller - /// queued, and a binary or selection node hands over what its kernel produced. `values` is how + /// queued, and a binary or vector node hands over what its kernel produced. `values` is how /// that node counts the values it hands over, because the two do not count the same thing: an /// input's figure has always been its positions, while a kernel's result holds one value per /// field and a field may have none at a position it decided. @@ -2181,6 +2194,7 @@ impl NodeRuntime { // hold it. let _timer = metrics.timer(); self.buffer.shrink(bytes)?; + self.release_queued_tables()?; } metrics.record_output(1, values(&chunk)); // An evaluated chunk is addressed on its grid, so its decided positions are grid points; a @@ -2368,44 +2382,44 @@ impl NodeRuntime { self.queue_results(output, false) } - /// Retains one chunk of a selection's input, reserved for before it is retained, as + /// Retains one chunk of a vector node's input, reserved for before it is retained, as /// [`Self::retain_binary`] retains an operand's. - fn retain_select(&mut self, path: &NodePath, chunk: NativeChunk) -> Result<()> { + fn retain_vector(&mut self, path: &NodePath, chunk: NativeChunk) -> Result<()> { let NativeChunk::Evaluated(chunk) = chunk else { return Err(DataFusionError::Execution(format!( - "the selection at {path} consumes evaluated chunks, not raw samples" + "the vector operation at {path} consumes evaluated chunks, not raw samples" ))); }; - let select = self.select.as_mut().ok_or_else(|| no_select(path))?; + let vector = self.vector.as_mut().ok_or_else(|| no_vector(path))?; // What a bound tree says the input produces is not proof of what a chunk carries, so the // chunk is checked against the layout before it is retained. - validate_binary_operand(select.schema.as_ref(), std::slice::from_ref(&chunk.series)) + validate_binary_operand(vector.schema.as_ref(), std::slice::from_ref(&chunk.series)) .map_err(|error| { DataFusionError::Execution(format!( - "the selection at {path} rejected its input: {error}" + "the vector operation at {path} rejected its input: {error}" )) })?; - let pending = select.input.pending(&chunk); + let pending = vector.input.pending(&chunk); let positions = chunk.series.time.len(); self.buffer.grow(pending)?; let metrics = Arc::clone(&self.buffer.metrics); { let _timer = metrics.timer(); - let select = self.select.as_mut().ok_or_else(|| no_select(path))?; - select.input.retain(chunk, Reader::Select(path))?; - let retained = select.input.held; + let vector = self.vector.as_mut().ok_or_else(|| no_vector(path))?; + vector.input.retain(chunk, Reader::Vector(path))?; + let retained = vector.input.held; self.buffer.resize(retained)?; } metrics.record_input(1, positions); Ok(()) } - /// Runs this node's selection kernel and queues the series it kept. - fn compute_select(&mut self, path: &NodePath) -> Result<()> { + /// Runs this node's vector kernel and queues the series it kept. + fn compute_vector(&mut self, path: &NodePath) -> Result<()> { let (output, shares_input) = { - let select = self.select.as_mut().ok_or_else(|| no_select(path))?; - let shares_input = matches!(select.op, BoundSelect::Sort(_)); - (select.compute(path, &mut self.buffer)?, shares_input) + let vector = self.vector.as_mut().ok_or_else(|| no_vector(path))?; + let shares_input = matches!(vector.op, BoundVectorOp::Sort(_)); + (vector.compute(path, &mut self.buffer)?, shares_input) }; let Some(output) = output else { return Ok(()); @@ -2418,9 +2432,10 @@ impl NodeRuntime { /// /// `shares_input` says the series are the assembled input itself, as a sort hands it back: a /// field there is one slice per input chunk of buffers and histogram tables the chunks shared, - /// so each series is measured the way the input was ([`operand_chunk_bytes`], each table billed - /// once) rather than by the allocations its slices belong to, which would count a buffer once - /// per chunk. + /// so each series is measured the way the input was ([`operand_chunk_bytes`]) rather than by + /// the allocations its slices belong to, which would count a buffer once per chunk. The tables + /// are billed once, to the queue ([`NodeRuntime::queued_tables`]), since several results can + /// keep one alive. fn queue_results(&mut self, output: Vec, shares_input: bool) -> Result<()> { let mut tables = HashSet::new(); // The results are queued the way a node that produces a chunk per input queues what it @@ -2432,7 +2447,8 @@ impl NodeRuntime { let shared = shares_input.then(|| { let (fresh, table_bytes) = unbilled_tables(&series, &tables); tables.extend(fresh); - operand_chunk_bytes(&series).saturating_add(table_bytes) + self.queued_tables = self.queued_tables.saturating_add(table_bytes); + operand_chunk_bytes(&series) }); let chunk = NativeChunk::Evaluated(EvaluatedChunk::try_new(series, true)?); // The result is already admitted, within the workspace reserved for the kernel. What the @@ -2451,7 +2467,9 @@ impl NodeRuntime { let queued = self .queued .iter() - .fold(0usize, |total, (_, bytes)| total.saturating_add(*bytes)); + .fold(self.queued_tables, |total, (_, bytes)| { + total.saturating_add(*bytes) + }); self.buffer.resize(queued) } @@ -2613,9 +2631,19 @@ impl NodeRuntime { return Ok(None); }; self.buffer.shrink(bytes)?; + self.release_queued_tables()?; Ok(Some(chunk)) } + /// Gives back the shared tables' charge once no queued result references them. + fn release_queued_tables(&mut self) -> Result<()> { + if self.queued.is_empty() && self.queued_tables > 0 { + self.buffer + .shrink(std::mem::take(&mut self.queued_tables))?; + } + Ok(()) + } + /// Records what this node hands out under its output label set. /// /// A node that rewrites labels is where two input series become indistinguishable, so this is @@ -2647,7 +2675,7 @@ impl NodeRuntime { self.window = None; self.instant = None; self.binary = None; - self.select = None; + self.vector = None; self.source_schema = None; self.emitted = None; self.merge = None; @@ -2657,6 +2685,7 @@ impl NodeRuntime { self.open_source_series = false; self.queued.clear(); self.queued.shrink_to_fit(); + self.queued_tables = 0; // The reservation is returned once, below, for everything this node still held. self.buffer.release(); } @@ -2736,7 +2765,7 @@ impl NativeExecutionState { // types than the kernel takes, so the shape is checked where the operands' layouts // are read. NativeExpr::Binary { .. } => NodeTag::Binary, - NativeExpr::Select { .. } => NodeTag::Select, + NativeExpr::Vector { .. } => NodeTag::Vector, // The kinds below are legal in a bound tree, and `validate` stays the authority on // that; this runtime simply does not execute them yet. Rejected before a single // chunk is read: a tree the runtime cannot execute must not start, because a @@ -2748,11 +2777,12 @@ impl NativeExecutionState { let mut runtime = NodeRuntime { buffer, queued: VecDeque::new(), + queued_tables: 0, instant: None, window: None, map: None, binary: None, - select: None, + vector: None, source_schema: None, emitted: None, merge: None, @@ -2838,10 +2868,10 @@ impl NativeExecutionState { // produced at points that do not collide; the kernel groups and merges them // itself, which is why this node keeps no ledger of what it handed out. } - NativeExpr::Select { op, .. } => { + NativeExpr::Vector { op, .. } => { let schema = node.output.samples().cloned().ok_or_else(|| { DataFusionError::Execution(format!( - "the selection at {path} does not produce samples" + "the vector operation at {path} does not produce samples" )) })?; // `topk`, `bottomk`, `limitk` and the value sorts read one value per sample, @@ -2849,16 +2879,16 @@ impl NativeExecutionState { // take any number. let one_field = matches!( op, - BoundSelect::TopK { .. } - | BoundSelect::LimitK { .. } - | BoundSelect::Sort(SortOrder::Asc | SortOrder::Desc) + BoundVectorOp::TopK { .. } + | BoundVectorOp::LimitK { .. } + | BoundVectorOp::Sort(SortOrder::Asc | SortOrder::Desc) ); if one_field && schema.fields.len() != 1 { - return Err(not_implemented("multi-field Select", &path)); + return Err(not_implemented("multi-field Vector", &path)); } - runtime.select = Some(SelectRuntime::new(op.clone(), schema)); - // A selection keeps the labels of the series it keeps, so it keeps no ledger of - // what it handed out. + runtime.vector = Some(VectorRuntime::new(op.clone(), schema)); + // 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 { .. } => { // Rejected above, where the pass has the position to name. @@ -2875,7 +2905,7 @@ impl NativeExecutionState { NativeExpr::Instant { input, .. } | NativeExpr::Window { input, .. } | NativeExpr::Map { input, .. } - | NativeExpr::Select { input, .. } => { + | NativeExpr::Vector { input, .. } => { pending.push((input.as_ref(), path.input())); true } @@ -2942,20 +2972,26 @@ impl NativeExecutionState { // them until the input ends -- so its flag is false whatever theirs are. holds_shared[index] = false; } - NodeTag::Select => { - // A selection keeps samples of the series it read, so pieces of one label + NodeTag::Vector => { + // Most vector operations keep samples of the series they read, so pieces of one label // set stay pieces: they hold disjoint points, so ranking them apart picks // what ranking the merged series would, and what is above treats them as - // it treats a map's. `absent` is the exception: it answers with one series - // under the labels it was bound to. - let absent = matches!( - expr, - NativeExpr::Select { - op: BoundSelect::Absent { .. }, - .. - } - ); - holds_shared[index] = input_holds_shared && !absent; + // it treats a map's. `absent` answers with one series under the labels it + // was bound to, and `limitk` counts series in input order, so it merges the + // pieces first and hands over whole series. + let op = match expr { + NativeExpr::Vector { op, .. } => Some(op), + _ => None, + }; + let limitk = matches!(op, Some(BoundVectorOp::LimitK { .. })); + if limitk + && input_holds_shared + && let Some(vector) = runtimes[index].vector.as_mut() + { + vector.merge_input = true; + } + let whole = limitk || matches!(op, Some(BoundVectorOp::Absent { .. })); + holds_shared[index] = input_holds_shared && !whole; } NodeTag::Window => { let (drops_metric_name, input_schema) = match expr { @@ -3133,9 +3169,9 @@ impl NativeExecutionState { /// Reports a stream that ended inside a series, without closing any node. /// /// Closing is destructive -- a window's evaluator takes itself apart -- and the binary and - /// selection nodes still have to compute after this check, so what is read here is only the + /// vector nodes still have to compute after this check, so what is read here is only the /// state that follows what each node was handed: an input node reports the series it handed - /// over without the chunk that ends it, and a binary or selection node reports an input that is + /// over without the chunk that ends it, and a binary or vector node reports an input that is /// still open. A truncation is therefore refused before a kernel allocates its workspace, rather /// than after one has already been computed on half a vector. fn unclosed(&self) -> Result<()> { @@ -3148,8 +3184,8 @@ impl NativeExecutionState { if let Some(binary) = &runtime.binary { binary.closed(path)?; } - if let Some(select) = &runtime.select { - select.input.ensure_closed(Reader::Select(path))?; + if let Some(vector) = &runtime.vector { + vector.input.ensure_closed(Reader::Vector(path))?; } } Ok(()) @@ -3271,7 +3307,7 @@ impl NativeExecutor { /// is materialised on their edge -- except below a window that collects its input, which holds /// that input until [`Self::finish`]. A binary node is the other exception: it pairs two whole /// vectors point by point, so it retains every chunk each operand hands over and reports `None` - /// until [`Self::finish`] ends the input, which is when it computes its result. A selection does + /// until [`Self::finish`] ends the input, which is when it computes its result. A vector node does /// the same with its one input. /// /// A chunk's `is_last` says the series whose data it carries ended. It does not say its label @@ -3454,11 +3490,11 @@ impl NativeExecutor { NodeTag::Map | NodeTag::Window | NodeTag::Instant => self.state.nodes[index] .input .is_none_or(|child| self.settled(child)), - NodeTag::Select => { + NodeTag::Vector => { runtime - .select + .vector .as_ref() - .is_some_and(|select| select.computed) + .is_some_and(|vector| vector.computed) && self.state.nodes[index] .input .is_none_or(|child| self.settled(child)) @@ -3630,20 +3666,20 @@ impl NativeExecutor { // does, so its values are counted rather than read as its positions. runtime.pop_queued(evaluated_values) } - NodeTag::Select => { + 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 // arrived, so the input is read to the end before anything is handed over. while let Some(chunk) = self.pull(child)? { let path = &self.state.paths[index]; - self.state.runtimes[index].retain_select(path, chunk)?; + self.state.runtimes[index].retain_vector(path, chunk)?; } if !self.inputs_finished || !self.settled(child) { return Ok(None); } let path = &self.state.paths[index]; let runtime = &mut self.state.runtimes[index]; - runtime.compute_select(path)?; + runtime.compute_vector(path)?; runtime.pop_queued(evaluated_values) } } @@ -3939,8 +3975,8 @@ fn no_binary(path: &NodePath) -> DataFusionError { DataFusionError::Execution(format!("the node at {path} holds no binary operation")) } -fn no_select(path: &NodePath) -> DataFusionError { - DataFusionError::Execution(format!("the node at {path} holds no selection")) +fn no_vector(path: &NodePath) -> DataFusionError { + DataFusionError::Execution(format!("the node at {path} holds no vector operation")) } /// The error for a reservation the platform cannot express. diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index b77e35e5685..bdc428aca96 100644 --- a/src/promql/src/native/execute/tests.rs +++ b/src/promql/src/native/execute/tests.rs @@ -45,7 +45,7 @@ use crate::map::MathOp; use crate::native::metrics::{NodeMetric, NodeMetrics, NodePath}; use crate::native::plan::{ AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar, - BoundSelect, BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, + BoundVectorOp, BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, }; use crate::range_eval::{RangeEvaluator, RangeFunction, RangeWindow}; use crate::select::SortOrder; @@ -479,11 +479,12 @@ fn source_runtime(pool: &Arc) -> NodeRuntime { pool, ), queued: VecDeque::new(), + queued_tables: 0, instant: None, window: None, map: None, binary: None, - select: None, + vector: None, source_schema: None, emitted: None, merge: None, @@ -496,6 +497,7 @@ fn binary_runtime(pool: &Arc, schema: &Arc) -> Nod NodeRuntime { buffer: NodeBuffer::new(&NodePath::root(), Arc::new(NodeMetrics::default()), pool), queued: VecDeque::new(), + queued_tables: 0, instant: None, window: None, map: None, @@ -503,7 +505,7 @@ fn binary_runtime(pool: &Arc, schema: &Arc) -> Nod Arc::new(binary_matching(BinaryOp::Add)), [Arc::clone(schema), Arc::clone(schema)], )), - select: None, + vector: None, source_schema: None, emitted: None, merge: None, @@ -6422,10 +6424,10 @@ fn an_all_missing_match_group_fits_its_pool() { assert_eq!(execution.state().reserved_bytes(), 0); } -fn select(inner: NativeNode, op: BoundSelect) -> NativeNode { +fn vector_op(inner: NativeNode, op: BoundVectorOp) -> NativeNode { let output = inner.output.clone(); NativeNode { - expr: NativeExpr::Select { + expr: NativeExpr::Vector { input: Box::new(inner), op, }, @@ -6460,9 +6462,9 @@ fn decided_by_job(chunks: &[EvaluatedChunk]) -> Vec<(String, Decided)> { fn topk_ranks_whole_series_at_each_point() { let grid = grid(0, 60_000, 3); let schema = evaluated_schema(Arc::clone(&grid)); - let tree = NativeTree::try_new(select( + let tree = NativeTree::try_new(vector_op( evaluated_input(Arc::clone(&grid)), - BoundSelect::TopK { + BoundVectorOp::TopK { k: 1.0, top: true, group: Grouping::By(Arc::from([])), @@ -6503,16 +6505,16 @@ fn topk_ranks_whole_series_at_each_point() { assert_eq!(pool.reserved(), 0); } -/// `limitk` takes series in label order, not in the order they arrived: `{job="x"}` is kept at -/// both points although `{job="y"}` arrived first. The two series a map folds into `{job="x"}` -/// stay pieces, each kept where it holds the point. +/// `limitk` takes series in the order its input hands them over, not in label order, and the two +/// series a map folds into `{job="z"}` are one series placed where it first arrived: it is kept at +/// both points although `{job="y"}` sorts first and arrived before the piece that fills point 0. #[test] -fn limitk_takes_series_in_label_order() { +fn limitk_takes_series_in_input_order() { let grid = grid(0, 60_000, 2); let schema = evaluated_schema(Arc::clone(&grid)); - let tree = NativeTree::try_new(select( + let tree = NativeTree::try_new(vector_op( map(evaluated_input(Arc::clone(&grid)), vec![MathOp::Abs]), - BoundSelect::LimitK { + BoundVectorOp::LimitK { k: 1.0, group: Grouping::By(Arc::from([])), }, @@ -6520,6 +6522,7 @@ fn limitk_takes_series_in_label_order() { .unwrap(); let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); for chunk in [ + evaluated_chunk_of(&schema, named_metric("a", "z"), 1, vec![Some(1.0)], true), evaluated_chunk_of( &schema, named_metric("b", "y"), @@ -6527,32 +6530,28 @@ fn limitk_takes_series_in_label_order() { vec![Some(2.0), Some(2.0)], true, ), - evaluated_chunk_of(&schema, named_metric("a", "x"), 1, vec![Some(1.0)], true), - evaluated_chunk_of(&schema, named_metric("c", "x"), 0, vec![Some(3.0)], true), + evaluated_chunk_of(&schema, named_metric("c", "z"), 0, vec![Some(3.0)], true), ] { execution.push_evaluated(chunk).unwrap(); } let chunks = run(&mut execution); assert_eq!( decided_by_job(&chunks), - vec![ - ("x".to_string(), vec![(1, Some(1.0))]), - ("x".to_string(), vec![(0, Some(3.0))]), - ] + vec![("z".to_string(), vec![(0, Some(3.0)), (1, Some(1.0))])] ); } -/// A binary node above a selection reads what the selection kept, once the selection has read its +/// A binary node above a vector node reads what `topk` kept, once the vector node has read its /// whole input: `topk` keeps `{job="b"}` at point 0 and `{job="a"}` at point 1, so only point 1 /// matches the right side's `{job="a"}`. #[test] -fn a_binary_node_reads_what_a_selection_kept() { +fn a_binary_node_reads_what_a_vector_node_kept() { let grid = grid(0, 60_000, 2); let schema = evaluated_schema(Arc::clone(&grid)); let tree = NativeTree::try_new(binary_of( - select( + vector_op( evaluated_input(Arc::clone(&grid)), - BoundSelect::TopK { + BoundVectorOp::TopK { k: 1.0, top: true, group: Grouping::By(Arc::from([])), @@ -6616,9 +6615,9 @@ fn absent_answers_where_the_input_holds_nothing() { }], ) .unwrap(); - let tree = NativeTree::try_new(select( + let tree = NativeTree::try_new(vector_op( evaluated_input(Arc::clone(&grid)), - BoundSelect::Absent { + BoundVectorOp::Absent { labels: labels.clone(), }, )) @@ -6651,9 +6650,9 @@ fn absent_answers_where_the_input_holds_nothing() { fn sort_desc_hands_over_series_in_value_order() { let grid = grid(0, 60_000, 1); let schema = evaluated_schema(Arc::clone(&grid)); - let tree = NativeTree::try_new(select( + let tree = NativeTree::try_new(vector_op( evaluated_input(Arc::clone(&grid)), - BoundSelect::Sort(SortOrder::Desc), + BoundVectorOp::Sort(SortOrder::Desc), )) .unwrap(); let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); @@ -6683,9 +6682,9 @@ fn a_sorted_series_is_queued_at_what_its_slices_hold() { let points = 1_000; let grid = grid(0, 60_000, points); let schema = evaluated_schema(Arc::clone(&grid)); - let tree = NativeTree::try_new(select( + let tree = NativeTree::try_new(vector_op( evaluated_input(Arc::clone(&grid)), - BoundSelect::Sort(SortOrder::ByLabel(vec!["job".into()])), + BoundVectorOp::Sort(SortOrder::ByLabel(vec!["job".into()])), )) .unwrap(); let pool = bounded_pool(48 * 1024); @@ -6701,10 +6700,82 @@ fn a_sorted_series_is_queued_at_what_its_slices_hold() { start + 100 == points as usize, )) .unwrap(); - // The selection retains each chunk as it arrives, at what its slice holds. + // The vector node retains each chunk as it arrives, at what its slice holds. assert!(execution.next_chunk().unwrap().is_none()); } let chunks = run(&mut execution); assert_eq!(decided(&chunks).len(), points as usize); assert_eq!(pool.reserved(), 0); } + +/// Two sorted results share one histogram table, so the table stays charged until the second one +/// leaves, not only until the first one does. +#[test] +fn a_table_shared_by_queued_results_stays_charged_until_the_last_leaves() { + let grid = grid(0, 60_000, 1); + let schema = evaluated_schema(Arc::clone(&grid)); + let tree = NativeTree::try_new(vector_op( + evaluated_input(Arc::clone(&grid)), + BoundVectorOp::Sort(SortOrder::ByLabel(vec!["job".into()])), + )) + .unwrap(); + let table = histogram_table(2, 64); + let pool = pool(); + let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new()); + for (job, index) in [("a", 0), ("b", 1)] { + execution + .push_evaluated(evaluated_histogram_chunk_of( + &schema, + named_metric("m", job), + 0, + vec![Some(index)], + &table, + true, + )) + .unwrap(); + } + execution.end_of_input().unwrap(); + let first = execution.next_chunk().unwrap().expect("two sorted results"); + assert_eq!(first.series.metric.get("job"), Some("a")); + drop(first); + assert!(pool.reserved() >= histogram_table_bytes(&table)); + let second = execution.next_chunk().unwrap().expect("two sorted results"); + assert_eq!(second.series.metric.get("job"), Some("b")); + assert!(execution.next_chunk().unwrap().is_none()); + assert_eq!(pool.reserved(), 0); +} + +/// `limitk` takes the order a sort below it decided: `sort_desc` puts `{job="z"}` first, so it is +/// the one `limitk(1, ...)` keeps, although `{job="a"}` sorts first by label. +#[test] +fn limitk_keeps_the_order_a_sort_below_it_decided() { + let grid = grid(0, 60_000, 1); + let schema = evaluated_schema(Arc::clone(&grid)); + let tree = NativeTree::try_new(vector_op( + vector_op( + evaluated_input(Arc::clone(&grid)), + BoundVectorOp::Sort(SortOrder::Desc), + ), + BoundVectorOp::LimitK { + k: 1.0, + group: Grouping::By(Arc::from([])), + }, + )) + .unwrap(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + for (job, value) in [("a", 1.0), ("z", 2.0)] { + execution + .push_evaluated(evaluated_chunk_of( + &schema, + named_metric("v", job), + 0, + vec![Some(value)], + true, + )) + .unwrap(); + } + assert_eq!( + decided_by_job(&run(&mut execution)), + vec![("z".to_string(), vec![(0, Some(2.0))])] + ); +} diff --git a/src/promql/src/native/metrics.rs b/src/promql/src/native/metrics.rs index 6e1a2635d6c..c894c9607b8 100644 --- a/src/promql/src/native/metrics.rs +++ b/src/promql/src/native/metrics.rs @@ -101,7 +101,7 @@ pub fn walk(root: &NativeNode, visit: &mut dyn FnMut(&NativeNode, &NodePath)) { | NativeExpr::Window { input, .. } | NativeExpr::Map { input, .. } | NativeExpr::Aggregate { input, .. } - | NativeExpr::Select { input, .. } => go(input, &path.input(), visit), + | NativeExpr::Vector { input, .. } => go(input, &path.input(), visit), NativeExpr::Binary { left, right, .. } => { go(left, &path.left(), visit); go(right, &path.right(), visit); diff --git a/src/promql/src/native/plan.rs b/src/promql/src/native/plan.rs index 8e827b78819..d520ddd3036 100644 --- a/src/promql/src/native/plan.rs +++ b/src/promql/src/native/plan.rs @@ -229,7 +229,7 @@ pub struct BoundAggregate { /// Its constants are bound here, as a window's are: a `k` or a ratio only known at run time has no /// field. #[derive(Clone, Debug, PartialEq)] -pub enum BoundSelect { +pub enum BoundVectorOp { /// `topk` (`top`) or `bottomk`: the `k` largest or smallest values of each group. TopK { k: f64, top: bool, group: Grouping }, /// `limitk`: the first `k` series of each group that hold a sample, in input order. @@ -273,10 +273,10 @@ pub enum NativeExpr { input: Box, op: BoundAggregate, }, - /// A selection of samples out of the whole input vector. - Select { + /// An operation over the whole input vector at each grid point. + Vector { input: Box, - op: BoundSelect, + op: BoundVectorOp, }, /// A binary operation between two inputs, with its matching rules. /// @@ -419,19 +419,19 @@ impl NativeNode { ensure_same_fields(&input.output, &self.output, "a map")?; ensure_grid(&self.output, input_grid, "a map")?; } - NativeExpr::Select { input, .. } => { + NativeExpr::Vector { input, .. } => { input.validate()?; - // A selection hands on samples of its input unchanged, so it consumes and produces + // A vector operation keeps the input's fields and grid, so it consumes and produces // the same evaluated vector. let Some((_, input_grid)) = evaluated_vector(&input.output) else { return Err(invalid(&format!( - "a selection consumes an evaluated vector, not {}", + "a vector operation consumes an evaluated vector, not {}", input.output.describe() ))); }; - ensure_samples(&self.output, "a selection")?; - ensure_same_fields(&input.output, &self.output, "a selection")?; - ensure_grid(&self.output, input_grid, "a selection")?; + ensure_samples(&self.output, "a vector operation")?; + ensure_same_fields(&input.output, &self.output, "a vector operation")?; + ensure_grid(&self.output, input_grid, "a vector operation")?; } NativeExpr::Aggregate { input, op } => { input.validate()?; diff --git a/src/promql/src/native/plan/tests.rs b/src/promql/src/native/plan/tests.rs index f4f24a99dc7..f4d14b58b9c 100644 --- a/src/promql/src/native/plan/tests.rs +++ b/src/promql/src/native/plan/tests.rs @@ -20,7 +20,7 @@ use datatypes::series::time::{EvalGrid, GridSpec}; use super::{ AggregateKind, AggregateStage, BoundAggregate, BoundInput, BoundInstant, BoundMap, BoundScalar, - BoundSelect, BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, + BoundVectorOp, BoundWindow, Grouping, NativeExpr, NativeNode, NativeTree, NativeType, }; use crate::map::MathOp; use crate::range_eval::{RangeFunction, RangeWindow}; @@ -1219,14 +1219,14 @@ fn an_input_can_bind_aggregate_state() { assert!(error.to_string().contains("not the binding"), "{error}"); } -/// A selection ranks the values at each grid point, so raw samples, which have no shared points, +/// A vector operation reads the values at each grid point, so raw samples, which have no shared points, /// are not something it can consume. #[test] -fn a_selection_cannot_consume_raw_samples() { +fn a_vector_operation_cannot_consume_raw_samples() { let node = NativeNode { - expr: NativeExpr::Select { + expr: NativeExpr::Vector { input: Box::new(input()), - op: BoundSelect::LimitRatio { ratio: 0.5 }, + op: BoundVectorOp::LimitRatio { ratio: 0.5 }, }, output: evaluated(Arc::new([]), grid(0, 60_000, 4)), }; diff --git a/src/promql/src/select.rs b/src/promql/src/select.rs index e88de344778..e57cd16db7d 100644 --- a/src/promql/src/select.rs +++ b/src/promql/src/select.rs @@ -574,7 +574,7 @@ fn labels_with_name(metric: &MetricName) -> Vec<(Arc, Arc)> { } /// Prometheus `labels.Compare`: label pairs in name order, then the shorter set first. -pub(crate) fn compare_labels(a: &MetricName, b: &MetricName) -> Ordering { +fn compare_labels(a: &MetricName, b: &MetricName) -> Ordering { labels_with_name(a).cmp(&labels_with_name(b)) }