diff --git a/src/promql/src/histogram_eval/float_histogram.rs b/src/promql/src/histogram_eval/float_histogram.rs index 71f9598bd81..30147d01645 100644 --- a/src/promql/src/histogram_eval/float_histogram.rs +++ b/src/promql/src/histogram_eval/float_histogram.rs @@ -80,6 +80,40 @@ impl FloatHistogram { }) } + /// The bytes this histogram's own heap buffers hold, by capacity. + /// + /// Everything a histogram keeps besides its scalars is on the heap: the bucket lists of both + /// sides, the custom bounds, and the span layout of a histogram read from a view, with the + /// `[Vec; 2]` array that holds the two layout vectors. No scalar of the struct shows any of it: + /// a histogram of 4096 positive buckets is 64 KiB of buckets, and a window that retains one + /// histogram per grid point retains that much per point. + /// + /// The layout array is stored inline in the histogram, so a ledger charging + /// `size_of::()` per slot already pays for those 48 bytes. Counting the whole + /// footprint as one figure here costs a header of over-estimate on a reservation, against a + /// payload of tens of kilobytes that would otherwise be missing from it entirely. + pub(crate) fn heap_bytes(&self) -> usize { + fn buckets(side: &Buckets) -> usize { + side.capacity() * std::mem::size_of::<(i32, f64)>() + } + let layout = self.layout.as_ref().map_or(0, |sides| { + std::mem::size_of::<[Vec<(i32, i32)>; 2]>().saturating_add( + sides + .iter() + .map(|side| side.capacity() * std::mem::size_of::<(i32, i32)>()) + .fold(0, usize::saturating_add), + ) + }); + buckets(&self.positive) + .saturating_add(buckets(&self.negative)) + .saturating_add( + self.custom_values + .capacity() + .saturating_mul(std::mem::size_of::()), + ) + .saturating_add(layout) + } + /// The spans of each side. fn spans(&self) -> [Vec<(i32, i32)>; 2] { self.layout diff --git a/src/promql/src/native/execute.rs b/src/promql/src/native/execute.rs index aa68bbdd8b8..f2296252194 100644 --- a/src/promql/src/native/execute.rs +++ b/src/promql/src/native/execute.rs @@ -49,10 +49,14 @@ 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::Float64Array; +use datatypes::arrow::array::{Array, Float64Array}; +use datatypes::arrow::buffer::Buffer; use datatypes::series::SeriesOrdering; +use datatypes::series::histogram::HistogramBatch; use datatypes::series::raw::RawSeriesChunk; -use datatypes::series::samples::{SampleRef, Samples, Series, SeriesFields, SeriesSchema}; +use datatypes::series::samples::{ + DenseSamplesRef, SampleRef, Samples, Series, SeriesFields, SeriesSchema, +}; use datatypes::series::time::{GridRange, TimeAxis}; use crate::grid_series::{LabelSet, label_set}; @@ -95,32 +99,132 @@ impl NativeChunk { } } - /// An estimate of the bytes this chunk holds. - fn bytes(&self) -> usize { + /// The bytes this chunk holds, counting each shared allocation it references that `charged` has + /// not met yet. + fn bytes(&self, charged: &mut ChargedAllocations) -> usize { match self { - Self::Raw(chunk) => raw_bytes(chunk), - Self::Evaluated(chunk) => evaluated_bytes(chunk), + Self::Raw(chunk) => raw_bytes(chunk, charged), + Self::Evaluated(chunk) => evaluated_bytes(chunk, charged), } } } -/// An estimate of the bytes a raw chunk holds: its timestamps, one float per field and its -/// identity. -fn raw_bytes(chunk: &RawSeriesChunk) -> usize { - std::mem::size_of::() - + chunk.primary_key.len() - + chunk.time.len() - * (std::mem::size_of::() + chunk.fields.len() * std::mem::size_of::()) +/// The shared allocations one accounting pass has already counted. +/// +/// A chunk holds very little of what it costs: a field of native histograms holds no floats at all, +/// only indices into an immutable `Arc` table, and both that table and the buffers +/// behind the field's Arrow arrays are batch-wide allocations that a chunk can hold a slice of. Two +/// chunks of one series can share one table by cloning the `Arc`, and both can hold sibling slices +/// of one buffer, so a pass counts an allocation the first time it meets it. Charging it per +/// referencing chunk instead would make the reservation grow with the number of chunks a node holds +/// rather than with the data in them -- the figure the pool has to rank. +/// +/// One pass walks one chunk, or the queue of one node, so the list stays as short as the +/// allocations it met and is dropped with the pass; nothing outlives the call it was made for. +#[derive(Default)] +struct ChargedAllocations { + /// The address of every histogram table this pass has counted. + tables: Vec<*const HistogramBatch>, + /// The address of every Arrow buffer this pass has counted. + buffers: Vec<*const u8>, } -/// An estimate of the bytes an evaluated chunk holds: one float per field and decided position. +impl ChargedAllocations { + /// The bytes `table` holds, or zero when this pass has already counted it. + fn table(&mut self, table: &Arc) -> usize { + let address = Arc::as_ptr(table); + if self.tables.contains(&address) { + return 0; + } + self.tables.push(address); + // The batch struct holds the struct array and the column arrays beside it, which are views + // of that array's children: the array reports the buffers of all of them, once. + std::mem::size_of::() + .saturating_add(table.as_arrow().get_array_memory_size()) + } + + /// The bytes the allocation `buffer` is a slice of, or zero when this pass has already counted + /// it. + /// + /// The capacity is the whole allocation, which is what a slice keeps alive: a key of a few + /// bytes sliced out of a decoded dictionary values buffer pins that buffer. A buffer that owns + /// no memory of its own reports no capacity, and then the bytes it shows are all there is. + fn buffer(&mut self, buffer: &Buffer) -> usize { + let address = buffer.data_ptr().as_ptr().cast_const(); + if self.buffers.contains(&address) { + return 0; + } + self.buffers.push(address); + buffer.capacity().max(buffer.len()) + } + + /// The bytes the buffers of one Arrow array hold. + fn array(&mut self, array: &dyn Array) -> usize { + let data = array.to_data(); + let values = data.buffers().iter().fold(0usize, |total, buffer| { + total.saturating_add(self.buffer(buffer)) + }); + // The validity bitmap is not one of `buffers`: a field of a series with a missing sample + // carries one, and it is an allocation like any other. + let nulls = data + .nulls() + .map_or(0, |nulls| self.buffer(nulls.inner().inner())); + values.saturating_add(nulls) + } +} + +/// The bytes a raw chunk holds: its identity, its timestamps and the Arrow buffers behind its +/// fields. /// -/// The grid itself is shared by every chunk addressed on it, so it is not counted per chunk. The -/// figure only has to rank one chunk against another for the memory pool, which is what a -/// reservation is for. -fn evaluated_bytes(chunk: &EvaluatedChunk) -> usize { +/// The figure is what the chunk keeps alive rather than one float per field and row: a histogram +/// field holds no floats at all, only indices into the table its samples live in. +fn raw_bytes(chunk: &RawSeriesChunk, charged: &mut ChargedAllocations) -> usize { + std::mem::size_of::() + .saturating_add(charged.buffer(&chunk.primary_key)) + .saturating_add(charged.buffer(chunk.time.as_buffer().inner())) + .saturating_add(fields_bytes(&chunk.fields, charged)) +} + +/// The bytes an evaluated chunk holds: its struct and the Arrow buffers behind its fields. +/// +/// The figure is what the chunk keeps alive rather than one float per field and decided position: +/// a histogram field holds no floats at all, only indices into the table its samples live in. The +/// grid itself is shared by every chunk addressed on it, so it is not counted per chunk. +fn evaluated_bytes(chunk: &EvaluatedChunk, charged: &mut ChargedAllocations) -> usize { std::mem::size_of::() - + chunk.series.time.len() * chunk.series.fields.len() * std::mem::size_of::() + .saturating_add(fields_bytes(&chunk.series.fields, charged)) +} + +/// 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, charged: &mut ChargedAllocations) -> usize { + fields + .iter() + .map(|field| { + field + .segments() + .iter() + .map(|segment| segment_bytes(segment, charged)) + .fold(0usize, usize::saturating_add) + }) + .fold(0usize, usize::saturating_add) +} + +/// The bytes one dense segment of samples holds. +fn segment_bytes(segment: &Samples, charged: &mut ChargedAllocations) -> usize { + match segment.dense() { + Some(DenseSamplesRef::Float(values)) => charged.array(values), + Some(DenseSamplesRef::Histogram(histograms)) => charged + .array(histograms.indices()) + .saturating_add(charged.table(histograms.batch())), + Some(DenseSamplesRef::Mixed(mixed)) => charged + .array(mixed.floats()) + .saturating_add(charged.array(mixed.histograms().indices())) + .saturating_add(charged.table(mixed.histograms().batch())), + // `segments` hands out dense segments only, so a nested chunk never reaches here; one that + // somehow did is charged for the descriptor alone rather than for buffers it nests. + None => 0, + } } /// Which node kinds the runtime executes. @@ -181,7 +285,7 @@ struct NodeSpec { struct NodeBuffer { metrics: Arc, reservation: MemoryReservation, - /// The bytes `reservation` holds, mirrored so a chunk leaving shrinks it exactly. + /// The bytes `reservation` holds, mirrored so a chunk leaving returns exactly them. held: usize, } @@ -204,15 +308,6 @@ impl NodeBuffer { Ok(()) } - /// Returns the bytes of a chunk this node no longer holds. - fn shrink(&mut self, bytes: usize) -> Result<()> { - let held = self.held.saturating_sub(bytes); - self.reservation.try_resize(held)?; - self.held = held; - self.metrics.record_reserved_bytes(held); - Ok(()) - } - /// Reserves exactly `bytes`, which is what this node holds from now on. /// /// A node that consumes its input chunk by chunk what it holds afterwards is not the sum of @@ -277,7 +372,8 @@ struct History { fields: Vec>, /// Whether the series on this label set closed. A second series cannot close it again. closed: bool, - /// The bytes this entry's labels and runs hold, kept current by [`Emitted::occupy`]. + /// The heap this entry owns: the runs its per-field vectors hold, kept current by + /// [`Emitted::occupy`]. The header and the slot it sits in are charged by the table's capacity. bytes: usize, } @@ -289,10 +385,14 @@ fn label_bytes(labels: &LabelSet) -> usize { .sum() } -/// The bytes a history holds: the entry itself, the per-field vectors and the runs in them. +/// The bytes a history holds: the per-field vectors and the runs in them. +/// +/// This is what the entry itself allocates, not the slot it is filed under: the `History` lives in +/// the map's slot, and that slot is what [`Emitted::bytes`] charges with the table's capacity. A +/// `size_of::()` term here billed every entry one header for a value it does not own, +/// once in this figure and again in the slot term beside it. fn history_bytes(history: &History) -> usize { - std::mem::size_of::() - + history.fields.capacity() * std::mem::size_of::>() + history.fields.capacity() * std::mem::size_of::>() + history .fields .iter() @@ -378,10 +478,11 @@ impl Emitted { /// An estimate of the bytes this table holds, which the node reserves for. /// - /// It counts each label's name and value bytes with a tuple per label, a history with its runs, - /// and the table's capacity at a label-set slot plus a history per entry. It leaves out the - /// map's buckets and the `Arc` headers of the strings it shares, so it is a lower bound rather - /// than a measurement. + /// It counts each label's name and value bytes with a tuple per label, the runs a history keeps + /// in its per-field vectors, and the table's capacity at one label-set slot plus one history + /// slot per entry -- which is where a `History` header itself is charged, once, for the entry + /// that lives in it. It leaves out the map's buckets and the `Arc` headers of the strings it + /// shares, so it is a lower bound rather than a measurement. /// /// The per-entry part is a running total, so reading this after every output does not walk the /// table: only the entry that changed and the table's own capacity are read here. @@ -519,15 +620,15 @@ impl WindowRuntime { /// The bytes this window's charge accounts for, as its evaluator's storage stands now. /// /// Between pushes this is the accounted retained size: the sample buffers the evaluator keeps - /// per field, by capacity, the row a push fills, the pending batch's inner vectors, and the - /// identity payload of the series still open. It is a lower bound on what is held, not a - /// measurement of it. + /// per field, by capacity, with the payload of every native histogram they retain, the row a + /// push fills with the histograms it boxes, the pending batch's inner vectors, and the identity + /// payload of the series still open -- the allocation behind its primary key, not the bytes the + /// key shows. It is a lower bound on what is held, not a measurement of it. /// /// The kernel's scratch is left out, being a constant factor of the window samples already /// counted. The `Emitted` table's own map overhead and the `Arc` headers of the strings it /// shares are charged for beside this figure rather than inside it. Of the pending batch only - /// the inner vectors are counted, not the outer `spans` and `columns` that hold them. And a - /// sliced buffer's backing can be larger than the bytes its primary key shows. + /// the inner vectors are counted, not the outer `spans` and `columns` that hold them. /// /// During a push the evaluator also holds the incoming chunk, which the caller charges while /// the call runs; vector growth during a push is not reserved at all, and is not bounded by one @@ -587,26 +688,42 @@ struct NodeRuntime { impl NodeRuntime { /// Queues a source chunk, holding the bytes it needs until it leaves. + /// + /// What this node holds is its queue and nothing else, so the queue is counted in one pass and + /// the reservation set from that figure: a chunk of a series can share its histogram table with + /// a chunk of the same series beside it, and counting the queue chunk by chunk would charge + /// that one table once per chunk instead of once for the allocation. fn enqueue(&mut self, chunk: NativeChunk) -> Result<()> { - let (bytes, positions) = (chunk.bytes(), chunk.positions()); - self.buffer.hold(bytes)?; - self.buffer.metrics.record_input(1, positions); + let positions = chunk.positions(); self.queued.push_back(chunk); + let bytes = self.queued_bytes(); + self.buffer.hold_exactly(bytes)?; + self.buffer.metrics.record_input(1, positions); Ok(()) } + /// The bytes the chunks this node still queues hold, counted as one accounting pass. + fn queued_bytes(&self) -> usize { + let mut charged = ChargedAllocations::default(); + self.queued + .iter() + .map(|chunk| chunk.bytes(&mut charged)) + .fold(0usize, usize::saturating_add) + } + /// Hands over the next queued source chunk, if there is one. fn pop_source(&mut self) -> Result> { let Some(chunk) = self.queued.pop_front() else { return Ok(None); }; + let bytes = self.queued_bytes(); let metrics = Arc::clone(&self.buffer.metrics); { // An input node's own work is handing the chunk over, so its self time covers that and // nothing its producer did. let _timer = metrics.timer(); // The chunk leaves this node, so the pool no longer has to hold it. - self.buffer.shrink(chunk.bytes())?; + self.buffer.hold_exactly(bytes)?; } metrics.record_output(1, chunk.positions()); // An evaluated source is already addressed on its grid, so its decided positions are grid @@ -626,7 +743,7 @@ impl NodeRuntime { /// there is nothing to hand on, and the input may still decide more. fn map(&mut self, path: &NodePath, chunk: EvaluatedChunk) -> Result> { let (bytes, positions, closing) = ( - evaluated_bytes(&chunk), + evaluated_bytes(&chunk, &mut ChargedAllocations::default()), chunk.series.time.len(), chunk.is_last, ); @@ -662,7 +779,11 @@ impl NodeRuntime { /// Feeds one chunk to this node's window. fn window(&mut self, path: &NodePath, chunk: NativeChunk) -> Result> { - let (bytes, positions, closing) = (chunk.bytes(), chunk.positions(), chunk.is_last()); + let (bytes, positions, closing) = ( + chunk.bytes(&mut ChargedAllocations::default()), + chunk.positions(), + chunk.is_last(), + ); // The chunk is held for as long as the evaluator consumes it. self.buffer.hold(bytes)?; let metrics = Arc::clone(&self.buffer.metrics); @@ -692,13 +813,14 @@ impl NodeRuntime { // The chunk has been consumed: what this node still holds is what the evaluator kept of it // and the label sets it has already handed out. Between pushes the charge for both is the // accounted size of that storage, a lower bound on what is held rather than a measurement: - // the evaluator's buffers by capacity and every label set handed out, leaving out the - // kernel's scratch, this table's map overhead with the `Arc` headers of its strings, the - // outer vectors of the pending batch and a sliced buffer's extra backing. During one push - // the evaluator additionally holds the incoming chunk, charged by `hold` above for as long - // as the call runs, and vector growth is not reserved and not bounded by one doubling — - // a bulk append can allocate more than twice the previous capacity, and several vectors can - // grow at once — so this figure is what the storage stands at, not a peak over the push. + // the evaluator's buffers by capacity -- with the payload of the histograms they retain and + // of the ones the row boxes -- and every label set handed out, leaving out the kernel's + // scratch, this table's map overhead with the `Arc` headers of its strings and the outer + // vectors of the pending batch. During one push the evaluator additionally holds the + // incoming chunk, charged by `hold` above for as long as the call runs, and vector growth + // is not reserved and not bounded by one doubling — a bulk append can allocate more than + // twice the previous capacity, and several vectors can grow at once — so this figure is + // what the storage stands at, not a peak over the push. // // The reason the figure must not be per-chunk: charging every chunk consumed makes the // reservation grow with the chunk count, so a fixed window over a long series reads as an diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index 8ffa6f5fc53..14a3606d1cc 100644 --- a/src/promql/src/native/execute/tests.rs +++ b/src/promql/src/native/execute/tests.rs @@ -20,8 +20,11 @@ use std::sync::Arc; use datafusion::error::DataFusionError; use datafusion::execution::memory_pool::{GreedyMemoryPool, MemoryPool}; use datafusion::physical_plan::metrics::{ExecutionPlanMetricsSet, MetricValue}; -use datatypes::arrow::array::Float64Array; +use datatypes::arrow::array::{Float64Array, UInt32Array}; use datatypes::arrow::buffer::{Buffer, ScalarBuffer}; +use datatypes::series::histogram::{ + BucketSideView, CounterResetHint, HistogramBatchBuilder, HistogramView, +}; use datatypes::series::raw::{RawSeriesChunk, RawTimestamps}; use datatypes::series::samples::{ FieldSpec, SampleRef, Samples, Series, SeriesFields, SeriesSchema, @@ -29,8 +32,12 @@ use datatypes::series::samples::{ use datatypes::series::time::{EvalGrid, GridRange, GridSpan, GridSpec, TimeAxis}; use datatypes::series::{Label, MetricName, SeriesOrdering}; -use super::{NativeExecutionState, NativeExecutor, map_chunk}; +use super::{ + ChargedAllocations, Emitted, History, NativeExecutionState, NativeExecutor, Run, label_bytes, + map_chunk, raw_bytes, +}; use crate::evaluated::EvaluatedChunk; +use crate::grid_series::{LabelSet, label_set}; use crate::map::MathOp; use crate::native::metrics::{NodeMetric, NodePath}; use crate::native::plan::{ @@ -257,6 +264,61 @@ fn raw_chunk_repeated( .unwrap() } +/// Pushes one native histogram whose positive side is one span of `buckets` buckets of one count +/// into `builder`, and returns the index it was filed under. +fn push_histogram(builder: &mut HistogramBatchBuilder, buckets: i32) -> u32 { + let offsets = [0]; + let lengths = [buckets]; + let counts = vec![1.0; usize::try_from(buckets).unwrap()]; + builder + .push(HistogramView { + schema: 0, + zero_threshold: 0.0, + zero_count: 0.0, + count: f64::from(buckets), + sum: f64::from(buckets), + reset_hint: CounterResetHint::Unknown, + start_timestamp: None, + custom_values: &[], + positive: BucketSideView { + span_offsets: &offsets, + span_lengths: &lengths, + buckets: &counts, + }, + negative: BucketSideView { + span_offsets: &[], + span_lengths: &[], + buckets: &[], + }, + }) + .unwrap() +} + +/// One raw chunk of `series-a` whose one field holds a `buckets`-wide native histogram at each of +/// `timestamps`. +fn raw_histogram_chunk( + schema: &SeriesSchema, + timestamps: &[i64], + buckets: i32, + is_last: bool, +) -> RawSeriesChunk { + let mut builder = HistogramBatchBuilder::new(); + let indices: UInt32Array = timestamps + .iter() + .map(|_| Some(push_histogram(&mut builder, buckets))) + .collect(); + let samples = Samples::try_from_histogram(indices, builder.finish().unwrap()).unwrap(); + RawSeriesChunk::try_new( + schema, + Buffer::from_slice_ref(b"series-a"), + Some(metric()), + RawTimestamps::try_new(ScalarBuffer::from(timestamps.to_vec())).unwrap(), + SeriesFields::one(samples), + is_last, + ) + .unwrap() +} + /// One evaluated chunk of an input series: `values` at `start` and onwards on `schema`'s grid. fn evaluated_chunk( schema: &SeriesSchema, @@ -1952,3 +2014,154 @@ fn an_emptying_operation_hands_on_nothing_until_it_closes_the_series() { ); assert!(out.last().unwrap().is_last); } + +/// A window over native histograms reserves what it retains, payload included, and the label-set +/// ledger it charges beside that agrees with a full walk of the table. +/// +/// The chunk accounting walks the samples a chunk holds instead of charging one float per field and +/// row, so a histogram field -- which holds no float at all -- reaches both figures as the table its +/// indices point into. Neither may drift from what the node says it holds: the reservation is the +/// retained bytes plus the emitted ones, and the running total of the emitted table is the same +/// figure recounted in full. +#[test] +fn a_window_over_histograms_reserves_what_it_retains() { + let out = grid(0, 10_000, 2); + let tree = NativeTree::try_new(window_with( + input(), + RangeFunction::LastOverTime, + 20_000, + None, + Arc::clone(&out), + )) + .unwrap(); + let schema = raw_schema(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + + for (timestamp, is_last) in [(0, false), (10_000, true)] { + execution + .push_raw(raw_histogram_chunk(&schema, &[timestamp], 2, is_last)) + .unwrap(); + while execution.next_chunk().unwrap().is_some() {} + let (retained, emitted) = execution.state().window_charge(&NodePath::root()); + assert!( + emitted > 0, + "the window handed nothing out, so the ledger below charges for nothing" + ); + assert_eq!( + execution.state().reserved_bytes(), + retained.saturating_add(emitted), + "the reservation is not what the window retains ({retained} bytes) plus what it handed \ + out ({emitted} bytes)", + ); + let (folded, walked) = execution.state().emitted_charge_pair(&NodePath::root()); + assert_eq!( + folded, walked, + "the folded charge of {folded} bytes is not the {walked} bytes the table holds", + ); + } +} + +/// One histogram table is charged once, however many chunks of the series reference it. +/// +/// The chunks of one series can share the immutable table their histogram indices point into, and a +/// node holding both of them holds one table: the source charges its queue as one accounting pass, +/// which counts the table the first time it meets it and the address the `Arc` holds is what makes +/// two clones one table. +#[test] +fn a_histogram_table_is_charged_once_per_pass() { + let schema = raw_schema(); + let mut builder = HistogramBatchBuilder::new(); + let index = push_histogram(&mut builder, 2); + let table = builder.finish().unwrap(); + let chunk = |timestamp: i64, is_last: bool| { + let samples = + Samples::try_from_histogram(UInt32Array::from(vec![Some(index)]), Arc::clone(&table)) + .unwrap(); + RawSeriesChunk::try_new( + &schema, + Buffer::from_slice_ref(b"series-a"), + Some(metric()), + RawTimestamps::try_new(ScalarBuffer::from(vec![timestamp])).unwrap(), + SeriesFields::one(samples), + is_last, + ) + .unwrap() + }; + let (first, second) = (chunk(0, false), chunk(10_000, true)); + + let mut charged = ChargedAllocations::default(); + let queue = raw_bytes(&first, &mut charged).saturating_add(raw_bytes(&second, &mut charged)); + let alone = raw_bytes(&first, &mut ChargedAllocations::default()) + .saturating_add(raw_bytes(&second, &mut ChargedAllocations::default())); + assert!( + queue < alone, + "a queue of two chunks sharing one table is charged {queue} bytes, no less than the \ + {alone} bytes they are charged apart: the table was counted twice", + ); + + // What the source reserves is exactly that one pass: it holds its queue and nothing else, and + // nothing has been pulled yet. + let tree = NativeTree::try_new(window(input(), grid(0, 10_000, 1))).unwrap(); + let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new()); + execution.push_raw(first).unwrap(); + execution.push_raw(second).unwrap(); + assert_eq!( + execution.state().reserved_bytes(), + queue, + "the queue of two chunks sharing one table was not charged as one pass", + ); +} + +/// One entry's figure is the heap it owns, and its `History` header is charged once, by its slot. +/// +/// The `History` of a label set lives in the map's slot, so a `size_of::()` term in the +/// entry's own figure billed one header twice: once here and once in the capacity term beside it. +/// The two assertions below are the definition of the whole figure -- the labels, the vectors and +/// their runs, and one label-set slot plus one history slot per entry -- so either header counted +/// in the wrong place fails them. +#[test] +fn a_history_entry_is_charged_once_for_its_header() { + let grid = grid(0, 10_000, 2); + let schema = evaluated_schema(Arc::clone(&grid)); + let metric = named_metric("requests", "job"); + let labels = label_set(&metric); + // Not the closing chunk: a closed label set drops its runs, and the runs are what the entry's + // own figure is. + let chunk = evaluated_chunk_of(&schema, metric, 0, vec![Some(1.0), Some(2.0)], false); + + let mut emitted = Emitted::default(); + emitted.occupy(&NodePath::root(), &chunk).unwrap(); + + let history = emitted + .sets + .get(&labels) + .expect("the chunk recorded its label set"); + let entry = label_bytes(&labels) + .saturating_add(history.fields.capacity() * std::mem::size_of::>()) + .saturating_add( + history + .fields + .iter() + .map(|runs| runs.capacity() * std::mem::size_of::()) + .sum::(), + ); + assert!( + history.fields.iter().any(|runs| !runs.is_empty()), + "the entry kept no run, so it has no heap of its own to charge for" + ); + assert_eq!( + emitted.entries, + entry, + "the entry's own bytes are not its labels and runs alone: a `History` header in the figure \ + would add the {} bytes its slot already charges", + std::mem::size_of::(), + ); + assert_eq!( + emitted.bytes(), + entry.saturating_add( + emitted.sets.capacity() + * (std::mem::size_of::() + std::mem::size_of::()) + ), + "the table's capacity is where a label-set slot and one `History` per entry are charged", + ); +} diff --git a/src/promql/src/range_eval.rs b/src/promql/src/range_eval.rs index 6b3b7667b03..8575c91260a 100644 --- a/src/promql/src/range_eval.rs +++ b/src/promql/src/range_eval.rs @@ -303,7 +303,20 @@ struct SeriesState { impl SeriesState { /// The bytes this series' identity holds, which the rows handed out do not cover. fn identity_bytes(&self) -> usize { - let primary_key = self.primary_key.as_ref().map_or(0, |key| key.len()); + // A scan slices the primary key out of a decoded dictionary-values buffer, so the bytes the + // key shows are not the bytes it keeps: a key of a few bytes pins the whole buffer it was + // sliced out of, which `capacity` reports for the allocation behind the slice. Charging + // the key's length instead would show a handful of bytes for an allocation the size of a + // decoded batch. (A buffer that does not own its memory reports no capacity, and then the + // bytes it shows are all there is.) + // + // No dedupe is needed against a second slice of the same allocation: one evaluator holds + // one series at a time, and a window over an evaluated input holds no key at all, so a + // pass over this state charges one slice of one allocation rather than one per slice. + let primary_key = self + .primary_key + .as_ref() + .map_or(0, |key| key.capacity().max(key.len())); let metric: usize = self .metric .name() @@ -380,7 +393,17 @@ fn resets_between(resets: &[CounterReset], start: u64, end: u64) -> &[CounterRes impl FieldSamples { /// The bytes this field's buffers hold, by capacity. + /// + /// The histogram term is the slot of every retained sample plus the payload the histogram in + /// that slot owns: a slot is one `(i64, FloatHistogram)` worth of struct, and the buckets the + /// histogram keeps are on the heap behind it, up to tens of kilobytes per sample. Only the + /// retained slice is walked, which is the storage the windows still read; the samples before + /// `head` are gone, their payloads with them. fn bytes(&self) -> usize { + let histograms = self.histograms[self.histogram_head..] + .iter() + .map(|(_, histogram)| histogram.heap_bytes()) + .fold(0usize, usize::saturating_add); self.timestamps .capacity() .saturating_mul(std::mem::size_of::()) @@ -394,6 +417,7 @@ impl FieldSamples { .capacity() .saturating_mul(std::mem::size_of::<(i64, FloatHistogram)>()), ) + .saturating_add(histograms) .saturating_add( self.resets .capacity() @@ -481,6 +505,26 @@ impl FieldSamples { } } +/// The bytes the histograms a row holds box, beyond the slots they sit in. +/// +/// A slot is an `Option`: a pointer and a tag, which is all +/// `size_of::>()` is. A result that is a native histogram does not live in that +/// slot itself: the slot points at a `FloatHistogram` on the heap, which owns its bucket +/// lists in turn, and both are what the row keeps alive between pushes. Only slots that hold a +/// histogram are counted, and only those the last evaluated point left in place: a push clears the +/// row, so the slots past its length hold nothing. +fn boxed_bytes(row: &[Option]) -> usize { + row.iter() + .flatten() + .filter_map(|sample| match sample { + GridSample::Histogram(histogram) => { + Some(std::mem::size_of::().saturating_add(histogram.heap_bytes())) + } + GridSample::Float(_) => None, + }) + .fold(0usize, usize::saturating_add) +} + impl RangeEvaluator { /// Creates an evaluator of `function` for the series of `input` onto `grid`. pub fn try_new( @@ -604,9 +648,10 @@ impl RangeEvaluator { /// An accounted lower bound on the bytes this evaluator keeps between pushes. /// /// Counted by capacity, so it is what the storage holds rather than what it uses: the retained - /// samples per field (timestamps, values and native histograms), the reset index, the row a push - /// fills, and the pending batch's own vectors. Left out are the kernel's scratch buffers, this - /// struct's `Vec` headers and the grid, which every chunk addressed on it shares. + /// samples per field (timestamps, values and the native histograms with the payload each of + /// them owns), the reset index, the row a push fills with the histograms it boxes, and the + /// pending batch's own vectors. Left out are the kernel's scratch buffers, this struct's `Vec` + /// headers and the grid, which every chunk addressed on it shares. /// /// A push is the only caller that also holds the incoming chunk, and it charges that itself for /// as long as the call runs; vector growth during a push is neither reserved nor bounded by one @@ -621,9 +666,11 @@ impl RangeEvaluator { .map(FieldSamples::bytes) .fold(0usize, usize::saturating_add); let row = self.row.capacity() * std::mem::size_of::>(); + let boxed = boxed_bytes(&self.row); let series = self.series.as_ref().map_or(0, SeriesState::identity_bytes); per_field .saturating_add(row) + .saturating_add(boxed) .saturating_add(series) .saturating_add(self.pending.bytes()) } diff --git a/src/promql/src/range_eval/tests.rs b/src/promql/src/range_eval/tests.rs index c11af5e3c02..beea2c27a31 100644 --- a/src/promql/src/range_eval/tests.rs +++ b/src/promql/src/range_eval/tests.rs @@ -786,6 +786,93 @@ fn a_chunk_with_histograms_is_not_appended_in_bulk() { assert_eq!(evaluator.bulk_ingest_count(), 0); } +/// One raw chunk of one field holding a single native histogram sample at `ts`, whose positive side +/// is one span of `buckets` buckets of one count each. +fn histogram_chunk(schema: &SeriesSchema, buckets: i32, ts: i64) -> RawSeriesChunk { + let offsets = [0]; + let lengths = [buckets]; + let counts = vec![1.0; usize::try_from(buckets).unwrap()]; + let mut builder = HistogramBatchBuilder::new(); + let index = builder + .push(HistogramView { + schema: 0, + zero_threshold: 0.0, + zero_count: 0.0, + count: f64::from(buckets), + sum: f64::from(buckets), + reset_hint: CounterResetHint::Unknown, + start_timestamp: None, + custom_values: &[], + positive: BucketSideView { + span_offsets: &offsets, + span_lengths: &lengths, + buckets: &counts, + }, + negative: BucketSideView { + span_offsets: &[], + span_lengths: &[], + buckets: &[], + }, + }) + .unwrap(); + let samples = Samples::try_from_histogram( + UInt32Array::from(vec![Some(index)]), + builder.finish().unwrap(), + ) + .unwrap(); + RawSeriesChunk::try_new( + schema, + Buffer::from(0u64.to_le_bytes().to_vec()), + Some(metric(0)), + RawTimestamps::try_new(ScalarBuffer::from(vec![ts])).unwrap(), + SeriesFields::try_from_vec(vec![samples]).unwrap(), + false, + ) + .unwrap() +} + +/// A window has to reserve what a native histogram it retains owns, not only the slot it sits in. +/// +/// The histogram sample below stays ahead of the second grid point's window, so the evaluator keeps +/// it, and the two runs differ in nothing but the number of buckets on the positive side: 4 and +/// 4096. Those buckets are `(index, count)` pairs on the heap, which no `size_of` of the struct +/// shows -- 64 KiB for the wider one -- so the bytes the evaluator reports have to grow by at least +/// that payload between the two. A figure counting slots alone would report one number for both. +#[test] +fn a_retained_histogram_is_charged_for_its_buckets() { + let retained = |buckets: i32| -> usize { + let schema = raw_schema(1); + let mut evaluator = RangeEvaluator::try_new( + &schema, + grid(0, 10_000, 2), + // Far longer than the grid: the second point's window still reaches the sample, so the + // push retains it instead of advancing past it. + RangeWindow { + range_ms: 600_000, + offset_ms: 0, + anchor_ms: None, + }, + // A function that counts the histogram, so what the row holds is a float: the two + // figures then differ in the payload of the retained sample alone, not in what the row + // boxes beside it. + RangeFunction::CountOverTime, + ) + .unwrap(); + evaluator + .push(&histogram_chunk(&schema, buckets, 0)) + .unwrap(); + evaluator.retained_bytes() + }; + let (few, many) = (4, 4096); + let payload = usize::try_from(many - few).unwrap() * std::mem::size_of::<(i32, f64)>(); + let (narrow, wide) = (retained(few), retained(many)); + assert!( + wide >= narrow + payload, + "a window retaining one histogram of {many} buckets reports {wide} bytes, not the {narrow} \ + it reports for {few} buckets plus the {payload} bytes of bucket payload between them", + ); +} + /// `rate` over a float-only series takes the batch path. The per-window path produces the same /// output, so a test comparing values alone could not tell which one ran. #[test]