fix(promql): charge the memory the native evaluator actually retains

Four omissions made the reservation blind to most of what a window over native
histograms keeps, and the existing incremental-equals-recomputed test could not see
any of them because both sides used the same formula.

- `FloatHistogram` owns four heap allocations -- both bucket lists, the custom
  bounds, and the span layout -- that `FieldSamples::bytes` never looked at. A
  retained 4096-bucket sample is tens of kilobytes against a ledger that saw a
  struct header. `heap_bytes` counts them, and both the retained sample slice and
  the histograms a row boxes are charged for what they own.
- Queued chunks were estimated as one float per field per row, which is not what a
  histogram field holds: it holds indices into a shared `Arc<HistogramBatch>` and
  the buffers behind the Arrow arrays. The figure now walks the samples, and counts
  a shared table or a shared buffer slice once per pass rather than once per chunk
  that references it.
- `history_bytes` included `size_of::<History>()` while the map slot term charged a
  `History` per entry as well, so every entry paid two headers. The per-entry figure
  now covers only the heap the entry owns.
- The primary key is charged for the allocation its `Buffer` keeps alive rather than
  the bytes it shows: a scan slices the key out of a decoded dictionary-values
  buffer, so a short key can pin a batch-sized allocation.

Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-10-20 21:29:42 +08:00
parent 9bd5c85a0f
commit 922c354b98
5 changed files with 564 additions and 61 deletions
@@ -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::<FloatHistogram>()` 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::<f64>()),
)
.saturating_add(layout)
}
/// The spans of each side.
fn spans(&self) -> [Vec<(i32, i32)>; 2] {
self.layout
+177 -55
View File
@@ -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::<RawSeriesChunk>()
+ chunk.primary_key.len()
+ chunk.time.len()
* (std::mem::size_of::<i64>() + chunk.fields.len() * std::mem::size_of::<f64>())
/// 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<HistogramBatch>` 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<HistogramBatch>) -> 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::<HistogramBatch>()
.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::<RawSeriesChunk>()
.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::<EvaluatedChunk>()
+ chunk.series.time.len() * chunk.series.fields.len() * std::mem::size_of::<f64>()
.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<NodeMetrics>,
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<Vec<Run>>,
/// 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::<History>()` 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>()
+ history.fields.capacity() * std::mem::size_of::<Vec<Run>>()
history.fields.capacity() * std::mem::size_of::<Vec<Run>>()
+ 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<Option<NativeChunk>> {
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<Option<EvaluatedChunk>> {
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<Option<EvaluatedChunk>> {
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
+215 -2
View File
@@ -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::<History>()` 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::<Vec<Run>>())
.saturating_add(
history
.fields
.iter()
.map(|runs| runs.capacity() * std::mem::size_of::<Run>())
.sum::<usize>(),
);
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::<History>(),
);
assert_eq!(
emitted.bytes(),
entry.saturating_add(
emitted.sets.capacity()
* (std::mem::size_of::<LabelSet>() + std::mem::size_of::<History>())
),
"the table's capacity is where a label-set slot and one `History` per entry are charged",
);
}
+51 -4
View File
@@ -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::<i64>())
@@ -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<GridSample>`: a pointer and a tag, which is all
/// `size_of::<Option<GridSample>>()` 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<GridSample>]) -> usize {
row.iter()
.flatten()
.filter_map(|sample| match sample {
GridSample::Histogram(histogram) => {
Some(std::mem::size_of::<FloatHistogram>().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::<Option<GridSample>>();
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())
}
+87
View File
@@ -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]