mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-11-14 14:56:54 +00:00
feat(promql): execute an instant selection as a tree node (#9798)
* feat(promql): execute a bound native tree node by node, over sparse evaluated series Ports the native runtime onto the sparse evaluated-series API: a chunk is addressed by a `TimeAxis` that may hold several runs, so a consumer reads positions through the axis instead of assuming `start + row`. The driver and both node kinds move to that API: - `chunk.range().len()` becomes `chunk.series.time.len()`, the compact row count, in every metric and byte estimate. - `occupied_runs` walks `time.positions()` with one field cursor, so a chunk of several runs records each run where it belongs. A run open across a hole now ends where the hole begins; leaving it open to the last row made two runs look like one and refused a series that occupies the hole. - A window keeps the pieces its evaluator decided in a queue: one push can decide several runs, and reporting only what the push returned stranded the rest until the next push. Queued pieces are charged by `retained_bytes`, and the charge is recomputed when one leaves, so a released piece is not held. - `map_chunk` answers `Option`: an operation that empties every position (`clamp` with `max < min`) yields no chunk at all, unless the chunk closes its series -- a closing chunk holding no row is handed over so a consumer can tell "nothing more here" from "no more input". `RangeEvaluator::retained_bytes` is restored with the three byte helpers it reads: a field's buffers, a series' identity and the pending batch's vectors, all by capacity, and still counted as storage rather than as a peak over a push. Tests read absolute positions from the axis (`decided` no longer adds a row number to a start), and two sparse cases cover what the old reading got wrong: a `Window -> Map -> Window` chain over a chunk of two runs, and a series that occupies the hole between another's runs. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * refactor(promql): drop the window queue the sparse evaluator never needed `RangeEvaluator::push` and `push_series` each return `Result<Option<EvaluatedChunk>>`: the evaluator gathers every run it decided into one chunk and hands back at most one. So `pending.extend(decided)` inserted at most one chunk and `take_pending` removed it on the spot, leaving the queue empty after every successful push. That makes the queue dead weight: a `VecDeque` allocation per productive window, an accounting term that is always zero, a second drain site in `pull`, and doc comments claiming a problem the API does not have. The driver's "ask the input again when a window decides nothing" loop was already the whole story. Back to returning the evaluator's own answer, and drop `pending`, `take_pending` and `take_window_output`. Covers the empty-map protocol this port introduces while it is here: an operation that empties every position hands on nothing while its series is open, and one empty chunk when it closes. A non-closing empty chunk would be rejected downstream and a dropped closing one would leave the series unclosed. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): pin the empty-map output count at one The test checked that every output was empty and that exactly one closed, which a stray extra non-closing empty output would also satisfy. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * 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> * fix(promql): keep the source queue's charge incremental The queue's reservation was recounted by walking every queued chunk on each enqueue and each pop. `push_raw` and `push_evaluated` are public and put no bound on how deep the queue gets, so a caller that queues a whole series before pulling it paid `N(N+1)/2` chunk visits to hold `N` chunks -- the same shape of defect this branch fixes in `absorb`. `QueuedCharge` keeps a reference count and the bytes behind each shared allocation: a table or buffer is charged when the first queued chunk reaches it and given back when the last one leaves, so an operation walks the one chunk it names. The count is per chunk rather than per field, because a pop re-walks the same chunk with the same dedupe and must remove exactly the references that were added. A chunk's own struct bytes are counted per chunk (`own_bytes`); only the allocations behind its fields are counted by address, since a struct has no address to dedupe on. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * perf(promql): append a forward-moving chunk to the emitted history `absorb` rebuilt the whole stored run array on every non-empty chunk. Legal input made that quadratic across chunks with no label collision: one series in `K` chunks, chunk `i` occupying only grid position `2*i`, stores one more run per chunk and moves `K(K+1)/2` of them -- about five billion for a hundred thousand rows. A chunk that starts at or past the end of the last stored run only ever adds to the end, so its runs are appended and the boundary run is joined where it touches. One that can interleave or reach backwards still takes the full one-pass merge, which can split a run. Both leave the runs sorted, disjoint and joined, which is what `Emitted::occupy` looks a position up by. The test records the chunks through `Emitted::occupy` itself and reads the history back from it, covering the forward case, the touching boundary, a hole that has to join both neighbours, and a position before everything stored. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): report a truncated input from a tree with no window `close` only asked windows whether a series was still open, so a tree of an input alone, or of a map over one, accepted a final chunk without `is_last` and handed the consumer a series that never closed. The input node now follows the series of the chunk it last handed over, and closing such a tree reports the truncation against it, which is the same refusal a window makes for the input it reads. `occupied_runs` read a row the axis names with `expect`, which panics on a `Series` whose fields are shorter than its axis; the fields are public, so that chunk is constructible by a caller. It now returns the misalignment as an error naming the node, the field and both lengths. The chunk protocol says one series closes once, which reads as one close per label set. State that it is per input series: a node that rewrites labels can hand two series over under one label set and both of them close. Also say in `Emitted`'s doc that the table is one node's own, so a label set costs its bytes again in every node that keeps one. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * feat(promql): execute an instant selection as a tree node The tree could not run `NativeExpr::Instant`: the state builder recorded `not_implemented` and refused the whole tree before reading a chunk, while the compiler already produced the node for a bare selector. A query the plan and the compiler both accept therefore had no runtime path at all. `InstantRuntime` buffers one series to its `is_last` chunk -- the selection kernel reads a complete axis -- then calls the shared `select_instants` once and gathers the rows it picked onto the binding's own grid. Positions come from the grid lattice the kernel was asked for, and a timestamp off that lattice is an error rather than a skipped row. A series that selects nothing still closes, with an empty chunk on the grid; an input that ends inside a series is reported on `finish()`. The node does not record label-set occupancy: it keeps the metric verbatim, so it cannot map two series onto one label set, and the occupancy key drops empty-valued tags -- two genuinely distinct series would be refused as a collision. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): refuse an interleaved input, and close a label set only for its owner Two series-level identity holes, both reported against `42ea525`. The truncation flag followed the last chunk handed over, which is the same as "is a series open" only while at most one series can be open. A tree with no window still accepted `Interleaved`, and `a` open, `b` closed, end of input then passed while `a` never closed. Tracking the open set would need an identity that a raw chunk in bound-label mode does not have (`metric` is `None`, the key holds it), so the ordering is refused when the state is built instead, for every tree shape and not only under a window. The flag's doc now says what it means and why that is the whole answer. `History` closed on any series' `is_last`, so with `abs` dropping the metric name a series that contributed nothing could close a label set another series was still filling: `a{job="x"}` value, `b{job="x"}` empty close, `a{job="x"}` continuation was refused with "already closed the label set". Contiguous input hands over exactly that sequence, so it was not an interleaving artefact. A history now remembers the series that filled it and closes only for that series' closing chunk; a series that produced nothing under an open set is passed over, as it already was under a closed one. The owner is charged as part of the entry. The ordering check also reads every input node rather than only the first: a tree with two inputs would otherwise have its second declaration read and dropped. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): reserve the instant gather's workspace and keep the histogram start Two findings from the review of the instant node. The gather allocates after the last chunk arrives: the concatenated axis, the descriptors that read it, the two arrays the selection returns, and the rows the builder keeps. Nothing asked the pool for any of it, so with one sample selected by every point of a million-point grid the node took tens of megabytes while its reservation fell to the buffered chunks' size -- nothing at all, on the chunk that ends the series. `gather_bytes` now derives a bound from the values in hand (rows, chunks, segments, fields, points, and the widest histogram row per field, since one row can be selected by every point), and the reservation grows by it before the gather runs. It shrinks back on both the success and the error path, so the workspace never outlives the call. A selection reads an existing sample, so it has to carry that sample's payload. `FloatHistogram` had no `start_timestamp` and `push_to` wrote `None`, which lost it for any histogram that had one: `detect_start_timestamp_reset` reads that field to spot a counter reset, so a window over this output would mis-detect resets. The field is now carried through a view and written back; the kernels that compute a histogram instead of reading one leave it `None`, which is what they produced before. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * docs(promql): describe the gather test by what this branch does The doc on `an_oversized_selection_is_refused_by_the_pool` said the kernel has no point limit any more. It does on this branch: the limit is removed on its own branch, which this one does not carry, so the comment described a state that has not arrived. What is true here is the order: the estimate now applies to every gather, so the pool is asked before the kernel is, and the pool's refusal is the one the caller sees. That is worth saying on its own -- a bound in bytes on the path that decides the points, rather than a count that says nothing about what they cost -- and it stays true after the limit goes. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * refactor(promql): simplify native execution memory estimates Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): merge rewritten logical series before window evaluation Accept disjoint closed-source fragments under one output label set. Retain occupancy checks and merge affected window inputs once at EOF using shared typed sample segments. Keep direct input windows and root maps streaming, and release collected fragments on finish or error. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * docs(promql): clarify approximate merge workspace estimate Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): validate series boundaries before collected input merges Keep each collecting window's open metric and last position so EOF grouping cannot hide reversed runs or identity changes before closing. Preserve source and root-map piece semantics, and test malformed input release, empty closing chunks, and EOF truncation. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): supply the default binary fill in the unimplemented-node fixture Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): reserve an instant selection for the points it decides The gather reserved the whole planned grid, while select_instants clips the range to the series' first and last sample and its lookback: a week-long query at a one-minute step reserved ~504 KB for a series that produces five points, and a small pool refused it before gathering. Reserve for the clipped candidate count, derived with the same alignment the kernel uses. Also pin the start-timestamp semantics the histogram change spread to: last_over_time and first_over_time hand back the sample they selected with its start timestamp, and limitk and limit_ratio keep it on the sample they keep, so a reader downstream can still place a counter reset. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): move the histogram_quantile doc back to its own test The start-timestamp test was inserted into the middle of another test's doc comment, leaving the earlier test without one. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * test(promql): refuse an interleaved input under a selection when the tree is bound The merged plan validates an input ordering in the arm that reads it, so a selection over an interleaved input no longer binds at all. Assert that refusal where it happens instead of asserting a runtime rejection the tree cannot reach, and keep the map-over-a-selection case: a map above a selection does not shield the check, because the check looks at the input the selection itself reads. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * docs(promql): say what the start timestamp pins, and cover every path that carries it The field is GreptimeDB sample metadata, so the comments now say that instead of arguing from what Prometheus omits: a FloatHistogram there has no such field, while its rate and reset evaluation takes start timestamps through a separate channel, and this crate's native range kernel does not read the field at all. Comparison filtering and set selection were the two paths that carry the timestamp without a test. Both keep the payload of the operand they hand back, including the right side of an or at a point the left side does not cover, and a bool comparison answers with a float so no histogram reaches the output. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): drop the start timestamp a computed histogram never had A window holding one histogram became the result as it arrived, while two or more went through an add that clears the timestamp, so whether a summed histogram carried the payload said how many samples the window held rather than what the function did. Clear it where the fold returns, leaving the bucket layout to the caller. Also correct two comments that described behaviour this branch does not have: the selection kernel carries no point limit of its own, and the interleaved map sits above the selection rather than between it and the input. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
@@ -156,7 +156,8 @@ fn aggregate_point<'a>(
|
||||
/// The Kahan sum or the mean of `histograms` in order, without compaction; `None` when there is
|
||||
/// none or they mix exponential and custom buckets. Prometheus `sum_over_time` and
|
||||
/// `avg_over_time` run the same steps as the aggregations. Histograms are consumed one at a
|
||||
/// time, so the first error ends the fold.
|
||||
/// time, so the first error ends the fold. The result is derived, so it carries no start
|
||||
/// timestamp.
|
||||
pub(crate) fn sum_or_avg(
|
||||
op: HistogramAggregate,
|
||||
histograms: impl IntoIterator<Item = Result<FloatHistogram>>,
|
||||
@@ -236,5 +237,11 @@ pub(crate) fn sum_or_avg(
|
||||
}
|
||||
},
|
||||
}
|
||||
// The result was derived from its inputs, so it carries no start timestamp of its own, like
|
||||
// every other operation that builds one. Clearing it here covers the fold that added nothing:
|
||||
// the first histogram of a window becomes the result as it arrived, and the adds above are
|
||||
// what drop the field otherwise, which would make the payload depend on the sample count.
|
||||
// The bucket layout is left as the fold built it -- compaction belongs to the caller.
|
||||
value.start_timestamp = None;
|
||||
Ok(Some(value))
|
||||
}
|
||||
|
||||
@@ -35,6 +35,22 @@ type Buckets = Vec<(i32, f64)>;
|
||||
#[derive(Debug, Clone)]
|
||||
pub(crate) struct FloatHistogram {
|
||||
pub(crate) reset_hint: CounterResetHint,
|
||||
/// The start timestamp of the sample this histogram was read from, in milliseconds.
|
||||
///
|
||||
/// It is metadata of GreptimeDB's histogram row, not part of the histogram value that
|
||||
/// Prometheus's `FloatHistogram` models. This owned form preserves it -- [`Self::try_from_view`]
|
||||
/// reads it off the sample and [`Self::push_to`] writes it back -- while every operation that
|
||||
/// derives a histogram from its input ([`Self::add`], [`Self::kahan_add`], [`Self::mul`],
|
||||
/// [`Self::div`], [`Self::compact`], [`Self::copy_to_schema`], with the constructors beside
|
||||
/// them) clears it. [`Self::equals`] does not compare it.
|
||||
///
|
||||
/// Both halves are this module's own policy and say nothing about how Prometheus propagates
|
||||
/// these timestamps: its `FloatHistogram` does not carry the field, while its rate and reset
|
||||
/// evaluation receives start timestamps through a separate channel. This crate's native range
|
||||
/// kernel does not read the field; the legacy extrapolation path does, to synthesize a zero
|
||||
/// point. So a selection that keeps the field preserves what was stored, and does not by itself
|
||||
/// make a native rate agree with the legacy one.
|
||||
pub(crate) start_timestamp: Option<i64>,
|
||||
pub(crate) schema: i32,
|
||||
pub(crate) zero_threshold: f64,
|
||||
pub(crate) zero_count: f64,
|
||||
@@ -62,6 +78,7 @@ impl FloatHistogram {
|
||||
pub(crate) fn try_from_view(view: &HistogramView<'_>) -> Result<Self> {
|
||||
Ok(Self {
|
||||
reset_hint: view.reset_hint,
|
||||
start_timestamp: view.start_timestamp,
|
||||
schema: view.schema,
|
||||
zero_threshold: view.zero_threshold,
|
||||
zero_count: view.zero_count,
|
||||
@@ -130,7 +147,7 @@ impl FloatHistogram {
|
||||
count: self.count,
|
||||
zero_count: self.zero_count,
|
||||
reset_hint: self.reset_hint,
|
||||
start_timestamp: None,
|
||||
start_timestamp: self.start_timestamp,
|
||||
custom_values: &self.custom_values,
|
||||
positive: BucketSideView {
|
||||
span_offsets: &positive_offsets,
|
||||
@@ -151,6 +168,7 @@ impl FloatHistogram {
|
||||
pub(crate) fn empty_like(&self) -> Self {
|
||||
Self {
|
||||
reset_hint: CounterResetHint::Unknown,
|
||||
start_timestamp: None,
|
||||
schema: self.schema,
|
||||
zero_threshold: 0.0,
|
||||
zero_count: 0.0,
|
||||
@@ -169,6 +187,8 @@ impl FloatHistogram {
|
||||
|
||||
/// Go `Mul`.
|
||||
pub(crate) fn mul(&mut self, factor: f64) {
|
||||
// A result this operation derived, so it carries no start timestamp of its own.
|
||||
self.start_timestamp = None;
|
||||
self.zero_count *= factor;
|
||||
self.count *= factor;
|
||||
self.sum *= factor;
|
||||
@@ -182,6 +202,8 @@ impl FloatHistogram {
|
||||
|
||||
/// Go `Div`; dividing by zero removes every bucket.
|
||||
pub(crate) fn div(&mut self, scalar: f64) {
|
||||
// A result this operation derived, so it carries no start timestamp of its own.
|
||||
self.start_timestamp = None;
|
||||
self.zero_count /= scalar;
|
||||
self.count /= scalar;
|
||||
self.sum /= scalar;
|
||||
@@ -215,6 +237,8 @@ impl FloatHistogram {
|
||||
|
||||
/// Go `Compact(0)`: drops empty buckets.
|
||||
pub(crate) fn compact(&mut self) {
|
||||
// A result this operation derived, so it carries no start timestamp of its own.
|
||||
self.start_timestamp = None;
|
||||
self.positive.retain(|&(_, count)| count != 0.0);
|
||||
self.negative.retain(|&(_, count)| count != 0.0);
|
||||
self.layout = None;
|
||||
@@ -223,13 +247,19 @@ impl FloatHistogram {
|
||||
/// Go `CopyToSchema`; `target` must not exceed the schema.
|
||||
pub(crate) fn copy_to_schema(&self, target: i32) -> Result<Self> {
|
||||
if target == self.schema {
|
||||
return Ok(self.clone());
|
||||
// The copy is a histogram this call produced, like the reduced one below: a start
|
||||
// timestamp belongs to the sample that carried it, not to a value derived from it.
|
||||
return Ok(Self {
|
||||
start_timestamp: None,
|
||||
..self.clone()
|
||||
});
|
||||
}
|
||||
if self.uses_custom_buckets() || target == CUSTOM_BUCKETS_SCHEMA || target > self.schema {
|
||||
return Err(incompatible_schema());
|
||||
}
|
||||
Ok(Self {
|
||||
reset_hint: CounterResetHint::Unknown,
|
||||
start_timestamp: None,
|
||||
schema: target,
|
||||
zero_threshold: self.zero_threshold,
|
||||
zero_count: self.zero_count,
|
||||
@@ -255,6 +285,8 @@ impl FloatHistogram {
|
||||
fn add_or_sub(&mut self, other: &Self, subtract: bool) -> Result<AddOutcome> {
|
||||
check_schemas(self, other)?;
|
||||
self.layout = None;
|
||||
// A result this operation derived, so it carries no start timestamp of its own.
|
||||
self.start_timestamp = None;
|
||||
let mut outcome = AddOutcome {
|
||||
counter_reset_collision: self.adjust_counter_reset(other),
|
||||
custom_bounds_reconciled: false,
|
||||
@@ -323,6 +355,8 @@ impl FloatHistogram {
|
||||
) -> Result<(Self, AddOutcome)> {
|
||||
check_schemas(self, other)?;
|
||||
self.layout = None;
|
||||
// A result this operation derived, so it carries no start timestamp of its own.
|
||||
self.start_timestamp = None;
|
||||
let mut outcome = AddOutcome {
|
||||
counter_reset_collision: self.adjust_counter_reset(other),
|
||||
custom_bounds_reconciled: false,
|
||||
@@ -410,6 +444,7 @@ impl FloatHistogram {
|
||||
fn new_compensation(&self) -> Self {
|
||||
Self {
|
||||
reset_hint: self.reset_hint,
|
||||
start_timestamp: None,
|
||||
schema: self.schema,
|
||||
zero_threshold: self.zero_threshold,
|
||||
zero_count: 0.0,
|
||||
|
||||
@@ -16,6 +16,9 @@ use std::collections::BTreeMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use datafusion::error::DataFusionError;
|
||||
use datatypes::series::histogram::{
|
||||
BucketSideView, CounterResetHint, HistogramBatchBuilder, HistogramView,
|
||||
};
|
||||
use datatypes::series::samples::{FieldSpec, SampleRef, Series, SeriesSchema};
|
||||
use datatypes::series::time::{EvalGrid, GridRange, GridSpec, TimeAxis};
|
||||
|
||||
@@ -383,6 +386,75 @@ fn histogram_points(series: &Series) -> BTreeMap<usize, String> {
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The start timestamp is payload of a histogram that was read, and not of one that was computed.
|
||||
///
|
||||
/// This pins GreptimeDB's own payload policy, not a Prometheus expectation: the timestamp is read
|
||||
/// off a sample ([`FloatHistogram::try_from_view`]) and written back
|
||||
/// ([`FloatHistogram::push_to`]), which is what a selection needs to hand a sample on unchanged,
|
||||
/// and every operation that derives a value from its input clears it. Nothing else depends on the
|
||||
/// field: the type derives `Debug` and `Clone` only, so it has no equality, hash or serialized form
|
||||
/// that could compare it, and [`FloatHistogram::equals`] is Go `Equals` on counts and bucket layout.
|
||||
#[test]
|
||||
fn a_start_timestamp_belongs_to_a_read_histogram_and_not_to_a_computed_one() {
|
||||
let offsets = [0];
|
||||
let lengths = [2];
|
||||
let buckets = [1.0, 2.0];
|
||||
let view = HistogramView {
|
||||
schema: 0,
|
||||
zero_threshold: 0.0,
|
||||
zero_count: 0.0,
|
||||
count: 3.0,
|
||||
sum: 3.0,
|
||||
reset_hint: CounterResetHint::Unknown,
|
||||
start_timestamp: Some(1500),
|
||||
custom_values: &[],
|
||||
positive: BucketSideView {
|
||||
span_offsets: &offsets,
|
||||
span_lengths: &lengths,
|
||||
buckets: &buckets,
|
||||
},
|
||||
negative: BucketSideView {
|
||||
span_offsets: &[],
|
||||
span_lengths: &[],
|
||||
buckets: &[],
|
||||
},
|
||||
};
|
||||
let histogram = FloatHistogram::try_from_view(&view).unwrap();
|
||||
assert_eq!(histogram.start_timestamp, Some(1500));
|
||||
|
||||
// Writing it back keeps it, which is what makes a selected sample a sample.
|
||||
let mut builder = HistogramBatchBuilder::new();
|
||||
histogram.push_to(&mut builder).unwrap();
|
||||
let table = builder.finish().unwrap();
|
||||
assert_eq!(
|
||||
table.view(0).unwrap().start_timestamp,
|
||||
Some(1500),
|
||||
"the histogram was written back without the start timestamp it was read with"
|
||||
);
|
||||
|
||||
// A value derived from it carries none, whichever operation derived it.
|
||||
let mut added = histogram.clone();
|
||||
added.add(&histogram).unwrap();
|
||||
assert_eq!(added.start_timestamp, None);
|
||||
let mut summed = histogram.clone();
|
||||
let (compensation, _) = summed.kahan_add(&histogram, None).unwrap();
|
||||
assert_eq!(summed.start_timestamp, None);
|
||||
assert_eq!(compensation.start_timestamp, None);
|
||||
let mut multiplied = histogram.clone();
|
||||
multiplied.mul(2.0);
|
||||
assert_eq!(multiplied.start_timestamp, None);
|
||||
let mut divided = histogram.clone();
|
||||
divided.div(2.0);
|
||||
assert_eq!(divided.start_timestamp, None);
|
||||
let mut compacted = histogram.clone();
|
||||
compacted.compact();
|
||||
assert_eq!(compacted.start_timestamp, None);
|
||||
assert_eq!(histogram.copy_to_schema(0).unwrap().start_timestamp, None);
|
||||
assert_eq!(histogram.empty_like().start_timestamp, None);
|
||||
// The one it was read into is still the sample it was read from.
|
||||
assert_eq!(histogram.start_timestamp, Some(1500));
|
||||
}
|
||||
|
||||
/// `histogram_quantile` drops all metadata labels; the single-histogram functions drop only
|
||||
/// `__name__`, as Prometheus does.
|
||||
#[test]
|
||||
|
||||
@@ -30,6 +30,12 @@
|
||||
//! outer window reads the inner result as the samples it is, not as something to be resampled onto
|
||||
//! the outer grid first.
|
||||
//!
|
||||
//! An instant selection owns no evaluator, because [`select_instants`] takes a complete source
|
||||
//! axis: the node buffers the chunks of the series it is selecting until the chunk that carries
|
||||
//! `is_last`, selects the last sample at or before each grid point of it, and hands the series over
|
||||
//! as one chunk addressed on its own grid. Nothing is decided for a series before it ends, which is
|
||||
//! the price of the kernel taking one axis rather than a chunk of one.
|
||||
//!
|
||||
//! A node kind this runtime does not execute yet is rejected when the execution is built, naming
|
||||
//! the kind and its position. An unsupported tree therefore never starts, rather than answering
|
||||
//! half a query or panicking halfway through one.
|
||||
@@ -68,20 +74,23 @@ use datafusion::error::{DataFusionError, Result};
|
||||
use datafusion::execution::memory_pool::{MemoryConsumer, MemoryPool, MemoryReservation};
|
||||
use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
|
||||
use datatypes::arrow::array::{Array, Float64Array};
|
||||
use datatypes::arrow::buffer::Buffer;
|
||||
use datatypes::series::histogram::HistogramBatch;
|
||||
use datatypes::series::raw::RawSeriesChunk;
|
||||
use datatypes::arrow::buffer::{Buffer, ScalarBuffer};
|
||||
use datatypes::series::histogram::{HistogramBatch, HistogramView};
|
||||
use datatypes::series::raw::{RawSeriesChunk, RawTimestamps};
|
||||
use datatypes::series::samples::{
|
||||
DenseSamplesRef, SampleRef, Samples, Series, SeriesFields, SeriesSchema,
|
||||
DenseSamplesRef, RowCursor, SampleRef, Samples, Series, SeriesFields, SeriesSchema,
|
||||
};
|
||||
use datatypes::series::time::{GridRange, TimeAxis};
|
||||
use datatypes::series::time::{EvalGrid, GridRange, GridSpec, TimeAxis};
|
||||
use datatypes::series::{MetricName, SeriesOrdering};
|
||||
|
||||
use crate::grid_series::{LabelSet, label_set, merge_same_labelset};
|
||||
use crate::grid_series::{
|
||||
GridSample, GridSeriesBuilder, LabelSet, label_set, merge_same_labelset, owned,
|
||||
};
|
||||
use crate::instant_eval::select_instants;
|
||||
use crate::map::{MathOp, drop_metric_name, map_math};
|
||||
use crate::native::batch::EvaluatedChunk;
|
||||
use crate::native::metrics::{NodeMetrics, NodePath};
|
||||
use crate::native::plan::{NativeExpr, NativeTree};
|
||||
use crate::native::plan::{BoundInstant, NativeExpr, NativeTree};
|
||||
use crate::range_eval::RangeEvaluator;
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -261,6 +270,8 @@ fn buffer_bytes(buffer: &Buffer) -> usize {
|
||||
enum NodeTag {
|
||||
/// Forwards the chunks it is given.
|
||||
Input,
|
||||
/// Buffers one series and selects the last sample at or before each of its grid points.
|
||||
Instant,
|
||||
/// Feeds its input to its own [`RangeEvaluator`].
|
||||
Window,
|
||||
/// Applies point-wise operations to its input chunks.
|
||||
@@ -272,6 +283,7 @@ impl NodeTag {
|
||||
fn describe(self) -> &'static str {
|
||||
match self {
|
||||
Self::Input => "input",
|
||||
Self::Instant => "instant selection",
|
||||
Self::Window => "window",
|
||||
Self::Map => "map",
|
||||
}
|
||||
@@ -836,6 +848,438 @@ struct MapRuntime {
|
||||
schema: Arc<SeriesSchema>,
|
||||
}
|
||||
|
||||
/// What an instant node holds: the chunks of the one series it is selecting, until that series ends.
|
||||
///
|
||||
/// [`select_instants`] takes a complete source axis, so this node cannot decide a grid point from
|
||||
/// the samples one chunk carries the way a window decides a range from what it has retained: it
|
||||
/// buffers a whole series and hands one chunk over for it. What it holds is therefore the series'
|
||||
/// own bytes, not a fixed window of them, and the reservation follows the buffer.
|
||||
struct InstantRuntime {
|
||||
/// The bound selection: how far back it reads, the offset it reads at, and the grid its values
|
||||
/// are addressed on.
|
||||
selection: BoundInstant,
|
||||
/// The layout the chunk it hands out is addressed on: its input's fields on that grid.
|
||||
schema: Arc<SeriesSchema>,
|
||||
/// Where the rows the selection picked are gathered, one series at a time.
|
||||
output: GridSeriesBuilder,
|
||||
/// The chunks of the series being selected, in arrival order; empty while no series is open.
|
||||
chunks: Vec<RawSeriesChunk>,
|
||||
/// What `chunks` hold, added as each chunk arrives and given back when the series is gathered.
|
||||
///
|
||||
/// Each chunk is measured once, with the estimate a queued chunk is charged with: an allocation
|
||||
/// two chunks of the series share is counted once per chunk, which is the over-count a figure
|
||||
/// that keeps no state across chunks accepts.
|
||||
held: usize,
|
||||
}
|
||||
|
||||
impl InstantRuntime {
|
||||
/// The run-time state of the instant selection `op`, reading the samples of `input`.
|
||||
fn try_new(input: &SeriesSchema, op: &BoundInstant) -> Result<Self> {
|
||||
// The chunk this node hands out is addressed on the selection's own grid, so the grid the
|
||||
// rows are gathered on is the binding's rather than a copy of it.
|
||||
let schema = Arc::new(SeriesSchema {
|
||||
fields: Arc::clone(&input.fields),
|
||||
eval_grid: Some(Arc::clone(&op.output_grid)),
|
||||
});
|
||||
Ok(Self {
|
||||
selection: op.clone(),
|
||||
output: GridSeriesBuilder::try_new(Arc::clone(&schema))?,
|
||||
schema,
|
||||
chunks: Vec::new(),
|
||||
held: 0,
|
||||
})
|
||||
}
|
||||
|
||||
/// The bytes the buffered chunks of the open series hold.
|
||||
fn buffered_bytes(&self) -> usize {
|
||||
self.held
|
||||
}
|
||||
|
||||
/// The bytes gathering the series `closing` ends is about to allocate beside its chunks.
|
||||
///
|
||||
/// An upper bound derived from what the selection has in hand rather than a measurement: the
|
||||
/// rows the series holds once `closing` has joined it, the fields it carries, the points the
|
||||
/// selection can decide for it, and, per field, the largest histogram a row of it can carry.
|
||||
/// Each term is one thing the gather allocates:
|
||||
///
|
||||
/// - The concatenated axis ([`buffered_time`]): one `i64` per row, reserved by the row count.
|
||||
/// - The descriptors that read it ([`buffered_fields`]): one [`Samples`] per (chunk, field) for
|
||||
/// the per-field vectors, three per segment for the vector [`Samples::try_from_chunks`]
|
||||
/// flattens them into -- the segment it came from, the vector it is flattened into and the
|
||||
/// `Arc<[Samples]>` the result is copied into -- and one `Vec<Samples>` header per field.
|
||||
/// - The two arrays [`select_instants`] returns: a `u64` source row index and an `i64`
|
||||
/// evaluation timestamp per point, reserved by the count they can reach at most.
|
||||
/// - The rows [`GridSeriesBuilder`] keeps: one `usize` grid index, one `f64` per field and a
|
||||
/// validity byte per (row, field). A field of native histograms adds one `Option<u32>` index
|
||||
/// per row and the payload of the histogram that index names, charged at the largest one the
|
||||
/// field holds and per *point* rather than per row: one source row can be selected by every
|
||||
/// point the selection decides, so the point count is what that payload is multiplied by.
|
||||
/// - The row the selection fills, one [`GridSample`] slot per field, and one [`RowCursor`] per
|
||||
/// field to read the axis with.
|
||||
/// - Doubling: every column the builder keeps is a `Vec` that grows by doubling, so each is
|
||||
/// charged twice what it ends up holding, which bounds its capacity rather than its length.
|
||||
///
|
||||
/// Left out, deliberately: the headers and Arrow buffers behind the buffered chunks (the
|
||||
/// node's own reservation already holds them, allocation by allocation), the `Vec` headers the
|
||||
/// gather drops before it returns, `Series`' own header and the runs [`GridSeriesBuilder`]
|
||||
/// hands the output over with, which the output chunk is charged for by whoever reads it, and
|
||||
/// the temporaries [`FloatHistogram::push_to`] builds for one row. None of those is
|
||||
/// proportional to the points the selection decides, which is the amplification the pool has to
|
||||
/// be asked for, and all of them are constant per gather or already counted row by row.
|
||||
///
|
||||
/// The estimate applies to every gather: nothing refuses one before it allocates, so the points
|
||||
/// the series can be decided at are reserved for whatever their number is.
|
||||
///
|
||||
/// [`FloatHistogram::push_to`]: crate::histogram_eval::float_histogram::FloatHistogram::push_to
|
||||
fn gather_bytes(&self, closing: &RawSeriesChunk) -> usize {
|
||||
// The points the selection decides, not the points the grid plans: the gather clips its
|
||||
// scan to the series it holds, so a figure over the whole grid would charge the pool for
|
||||
// rows this series can never produce.
|
||||
let points = self.selected_points(closing);
|
||||
let chunks = self.chunks.len().saturating_add(1);
|
||||
let rows = self
|
||||
.chunks
|
||||
.iter()
|
||||
.chain(std::iter::once(closing))
|
||||
.map(|chunk| chunk.time.len())
|
||||
.fold(0usize, usize::saturating_add);
|
||||
let fields = self.schema.fields.len();
|
||||
let descriptor = std::mem::size_of::<Samples>();
|
||||
let segments = self
|
||||
.chunks
|
||||
.iter()
|
||||
.chain(std::iter::once(closing))
|
||||
.flat_map(|chunk| chunk.fields.as_slice())
|
||||
.map(|field| field.segments().len())
|
||||
.fold(0usize, usize::saturating_add);
|
||||
let axis = rows.saturating_mul(std::mem::size_of::<i64>());
|
||||
let layout = chunks
|
||||
.saturating_mul(fields)
|
||||
.saturating_mul(descriptor)
|
||||
.saturating_add(segments.saturating_mul(3).saturating_mul(descriptor))
|
||||
.saturating_add(fields.saturating_mul(std::mem::size_of::<Vec<Samples>>()));
|
||||
let selection =
|
||||
points.saturating_mul(std::mem::size_of::<u64>() + std::mem::size_of::<i64>());
|
||||
let mut rows_of_points = points.saturating_mul(
|
||||
std::mem::size_of::<usize>() + fields.saturating_mul(std::mem::size_of::<f64>() + 1),
|
||||
);
|
||||
for largest in self.largest_histograms(closing).iter().flatten() {
|
||||
rows_of_points = rows_of_points.saturating_add(
|
||||
points.saturating_mul(std::mem::size_of::<Option<u32>>().saturating_add(*largest)),
|
||||
);
|
||||
}
|
||||
let row = fields.saturating_mul(
|
||||
std::mem::size_of::<Option<GridSample>>() + std::mem::size_of::<RowCursor<'_>>(),
|
||||
);
|
||||
axis.saturating_add(layout)
|
||||
.saturating_add(selection)
|
||||
.saturating_add(row)
|
||||
.saturating_add(rows_of_points.saturating_mul(2))
|
||||
}
|
||||
|
||||
/// The number of grid points [`select_instants`] clips the buffered series to.
|
||||
///
|
||||
/// The gather clips its scan to the series it holds instead of reading the whole planned grid:
|
||||
/// the candidate points run from the first sample's instant to the last sample's plus the
|
||||
/// lookback, aligned to the grid and held inside the query range. A workspace is charged per
|
||||
/// point, so the count has to be the clipped one -- charging the planned grid charges the pool
|
||||
/// for rows a short series can never produce, and a pool sized for what the series costs then
|
||||
/// refuses the gather. The clipping below is [`select_instants`]' own, repeated rather than
|
||||
/// approximated: an estimate that disagrees with the kernel is a bug here rather than a
|
||||
/// conservative bound. It is the kernel's own candidate count -- the one it sizes its arrays
|
||||
/// with -- so it bounds the rows it can report even though a candidate at the lookback edge
|
||||
/// itself selects nothing. Both ends come from chunks the node already holds, so the count
|
||||
/// reads two timestamps rather than the rows.
|
||||
fn selected_points(&self, closing: &RawSeriesChunk) -> usize {
|
||||
let spec = self.selection.output_grid.spec();
|
||||
// The selection refuses these itself, before it can decide anything.
|
||||
if spec.step_ms <= 0 || self.selection.lookback_ms < 0 {
|
||||
return 0;
|
||||
}
|
||||
let query_start = i128::from(spec.start_ms);
|
||||
let query_end = i128::from(grid_end_ms(spec));
|
||||
if query_end < query_start {
|
||||
return 0;
|
||||
}
|
||||
let series = || self.chunks.iter().chain(std::iter::once(closing));
|
||||
let (Some(first), Some(last)) = (
|
||||
series().find_map(|chunk| chunk.time.as_buffer().first().copied()),
|
||||
series()
|
||||
.rev()
|
||||
.find_map(|chunk| chunk.time.as_buffer().last().copied()),
|
||||
) else {
|
||||
// A series of no sample decides no point.
|
||||
return 0;
|
||||
};
|
||||
let offset = i128::from(self.selection.offset_ms);
|
||||
let lookback = i128::from(self.selection.lookback_ms);
|
||||
let first = i128::from(first) + offset;
|
||||
let last = i128::from(last) + offset;
|
||||
let useful_last = if lookback == 0 {
|
||||
last
|
||||
} else {
|
||||
last + lookback - 1
|
||||
};
|
||||
let max_start = first.max(query_start);
|
||||
let min_end = useful_last.min(query_end);
|
||||
if max_start > min_end {
|
||||
return 0;
|
||||
}
|
||||
let step = i128::from(spec.step_ms);
|
||||
let aligned_start = query_start + (max_start - query_start) / step * step;
|
||||
let aligned_end = query_end - (query_end - min_end) / step * step;
|
||||
let points = (aligned_end - aligned_start) / step + 1;
|
||||
usize::try_from(points).unwrap_or(usize::MAX)
|
||||
}
|
||||
|
||||
/// The largest histogram a row of each field can carry, `None` for a field of floats.
|
||||
///
|
||||
/// A row of native histograms is copied into the builder once per grid point that selects it,
|
||||
/// so what such a field costs is one histogram per point: the largest row is the one to charge,
|
||||
/// because any row can be the one every decided point picks.
|
||||
fn largest_histograms(&self, closing: &RawSeriesChunk) -> Vec<Option<usize>> {
|
||||
let mut largest = vec![None; self.schema.fields.len()];
|
||||
for chunk in self.chunks.iter().chain(std::iter::once(closing)) {
|
||||
for (slot, field) in largest.iter_mut().zip(chunk.fields.as_slice()) {
|
||||
if let Some(bytes) = histogram_row_bytes(field) {
|
||||
*slot = Some(slot.unwrap_or(0).max(bytes));
|
||||
}
|
||||
}
|
||||
}
|
||||
largest
|
||||
}
|
||||
|
||||
/// Buffers one chunk of the series being selected, and gathers that series when it ends.
|
||||
///
|
||||
/// `None` means the series is still open: the selection reads one whole axis, so nothing is
|
||||
/// decided for a series until the chunk carrying `is_last` arrives.
|
||||
fn push(&mut self, path: &NodePath, chunk: RawSeriesChunk) -> Result<Option<EvaluatedChunk>> {
|
||||
if let Some(open) = self.chunks.first()
|
||||
&& (open.primary_key.as_slice() != chunk.primary_key.as_slice()
|
||||
|| open.metric != chunk.metric)
|
||||
{
|
||||
// One axis is gathered before the kernel reads any of it, so a chunk of another series
|
||||
// has nowhere to go: it would be selected as though its samples were the open series'.
|
||||
return Err(series_changed(path));
|
||||
}
|
||||
let closing = chunk.is_last;
|
||||
self.held = self.held.saturating_add(raw_bytes(&chunk));
|
||||
self.chunks.push(chunk);
|
||||
if !closing {
|
||||
return Ok(None);
|
||||
}
|
||||
self.gather(path).map(Some)
|
||||
}
|
||||
|
||||
/// Selects the instants of the buffered series, hands it over closed, and drops the buffer.
|
||||
fn gather(&mut self, path: &NodePath) -> Result<EvaluatedChunk> {
|
||||
let chunk = self.select(path)?;
|
||||
// The selection copied what it needed out of the buffered chunks, so a series that has been
|
||||
// answered must not keep charging the pool for them.
|
||||
self.chunks.clear();
|
||||
self.chunks.shrink_to_fit();
|
||||
self.held = 0;
|
||||
Ok(chunk)
|
||||
}
|
||||
|
||||
/// The last sample at or before each grid point of the buffered series, as one chunk.
|
||||
fn select(&mut self, path: &NodePath) -> Result<EvaluatedChunk> {
|
||||
let Some(first) = self.chunks.first() else {
|
||||
return Err(DataFusionError::Execution(format!(
|
||||
"the instant selection at {path} holds no series to select"
|
||||
)));
|
||||
};
|
||||
let metric = first.metric.clone().ok_or_else(|| no_labels(path))?;
|
||||
// The buffered chunks are one series, so they are read as the one axis the selection takes.
|
||||
let time = buffered_time(&self.chunks)?;
|
||||
let fields = buffered_fields(&self.chunks, self.schema.fields.len())?;
|
||||
let grid = Arc::clone(&self.selection.output_grid);
|
||||
let spec = grid.spec();
|
||||
// The grid's own points are what the selection decides, so the range and the step it is
|
||||
// asked for are the grid's: `start..=start + (len - 1) * step`, step by step.
|
||||
let (indices, timestamps) = select_instants(
|
||||
&time,
|
||||
&fields,
|
||||
spec.start_ms..=grid_end_ms(spec),
|
||||
spec.step_ms,
|
||||
self.selection.lookback_ms,
|
||||
self.selection.offset_ms,
|
||||
)?;
|
||||
self.output.begin_series();
|
||||
// One cursor per field, so a row is read as the field holds it rather than by walking the
|
||||
// chunks of a field from its start once per row.
|
||||
let mut cursors: Vec<RowCursor<'_>> = fields.iter().map(Samples::cursor).collect();
|
||||
// `GridSample` is not `Clone`, so the row is filled slot by slot rather than from one value.
|
||||
let mut row: Vec<Option<GridSample>> = (0..fields.len()).map(|_| None).collect();
|
||||
let mut read: Option<usize> = None;
|
||||
for (index, timestamp) in indices.iter().zip(×tamps) {
|
||||
let index = *index as usize;
|
||||
// The selection steps the grid's range by its step, so every timestamp it reports is a
|
||||
// point of the grid and the position is the point's index in it.
|
||||
let position = grid_position(&grid, *timestamp)?;
|
||||
// Two grid points can select one source row, and the row is read once for both.
|
||||
if read != Some(index) {
|
||||
for (slot, cursor) in row.iter_mut().zip(&mut cursors) {
|
||||
let sample = cursor.at(index).ok_or_else(|| missing_row(path, index))?;
|
||||
*slot = owned(sample)?;
|
||||
}
|
||||
read = Some(index);
|
||||
}
|
||||
// A position no field holds a value at is not a row: the builder drops it, which is
|
||||
// what keeps a hole in the grid out of the result instead of padding it.
|
||||
self.output.push_samples(position, &row)?;
|
||||
}
|
||||
// The rows the selection picked are the rows this series holds, and the whole series is
|
||||
// decided at once, so this one chunk closes it.
|
||||
match self.output.finish(metric.clone())? {
|
||||
Some(series) => EvaluatedChunk::try_new(series, true),
|
||||
None => no_row_chunk(&self.schema, metric),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The buffered chunks as the one source axis they make up: their timestamps in arrival order.
|
||||
///
|
||||
/// The axis type validates the result as strictly increasing, which is also what refuses a series
|
||||
/// whose chunks do not continue each other.
|
||||
fn buffered_time(chunks: &[RawSeriesChunk]) -> Result<RawTimestamps> {
|
||||
let len = chunks.iter().map(|chunk| chunk.time.len()).sum();
|
||||
let mut timestamps: Vec<i64> = Vec::with_capacity(len);
|
||||
for chunk in chunks {
|
||||
timestamps.extend_from_slice(chunk.time.as_buffer());
|
||||
}
|
||||
RawTimestamps::try_new(ScalarBuffer::from(timestamps)).map_err(series_error)
|
||||
}
|
||||
|
||||
/// The buffered chunks as one axis: per field, the chunks' samples in arrival order.
|
||||
///
|
||||
/// A field retains the chunks it arrived in instead of copying them out, so the axis costs one
|
||||
/// descriptor per chunk and the selection reads the values where they already are.
|
||||
fn buffered_fields(chunks: &[RawSeriesChunk], fields: usize) -> Result<SeriesFields> {
|
||||
let mut per_field: Vec<Vec<Samples>> = vec![Vec::with_capacity(chunks.len()); fields];
|
||||
for chunk in chunks {
|
||||
for (field, slot) in chunk.fields.iter().zip(&mut per_field) {
|
||||
slot.push(field.clone());
|
||||
}
|
||||
}
|
||||
let fields = per_field
|
||||
.into_iter()
|
||||
.map(|chunks| Samples::try_from_chunks(chunks).map_err(series_error))
|
||||
.collect::<Result<Vec<_>>>()?;
|
||||
SeriesFields::try_from_vec(fields).map_err(series_error)
|
||||
}
|
||||
|
||||
/// The bytes the builder's table holds for one row of the field, `None` for a field of floats.
|
||||
///
|
||||
/// The largest row of the field rather than its total: every grid point that selects one source row
|
||||
/// copies that row, so the widest histogram the field holds is what a point can cost.
|
||||
fn histogram_row_bytes(field: &Samples) -> Option<usize> {
|
||||
let holds_histograms = field.segments().iter().any(|segment| {
|
||||
matches!(
|
||||
segment.dense(),
|
||||
Some(DenseSamplesRef::Histogram(_) | DenseSamplesRef::Mixed(_))
|
||||
)
|
||||
});
|
||||
if !holds_histograms {
|
||||
return None;
|
||||
}
|
||||
Some(
|
||||
field
|
||||
.iter()
|
||||
.filter_map(|sample| match sample {
|
||||
SampleRef::Histogram(view) => Some(histogram_bytes(&view)),
|
||||
_ => None,
|
||||
})
|
||||
.fold(0, usize::max),
|
||||
)
|
||||
}
|
||||
|
||||
/// The bytes the histogram table holds for one row of `view`.
|
||||
///
|
||||
/// The row `HistogramBatchBuilder::push` writes: its scalar columns, the offset each of the seven
|
||||
/// bucket lists pays for it, and the values those lists hold -- the custom bounds, and each side's
|
||||
/// bucket counts with the span offsets and lengths that place them. The spans are the view's own,
|
||||
/// because a histogram handed on unchanged keeps its layout through
|
||||
/// [`crate::histogram_eval::float_histogram::FloatHistogram::push_to`].
|
||||
fn histogram_bytes(view: &HistogramView<'_>) -> usize {
|
||||
// Schema, zero threshold, sum, count, zero count, reset hint and start timestamp, the narrow
|
||||
// columns counted at the width of the widest.
|
||||
let scalars = 7 * std::mem::size_of::<i64>();
|
||||
// One offset per list column: custom bounds, two span columns and one bucket column per side.
|
||||
let list_offsets = 7 * std::mem::size_of::<u32>();
|
||||
let custom = view
|
||||
.custom_values
|
||||
.len()
|
||||
.saturating_mul(std::mem::size_of::<f64>());
|
||||
let sides = [view.positive, view.negative]
|
||||
.iter()
|
||||
.map(|side| {
|
||||
side.span_offsets
|
||||
.len()
|
||||
.saturating_mul(2)
|
||||
.saturating_mul(std::mem::size_of::<i32>())
|
||||
.saturating_add(
|
||||
side.buckets
|
||||
.len()
|
||||
.saturating_mul(std::mem::size_of::<f64>()),
|
||||
)
|
||||
})
|
||||
.fold(0usize, usize::saturating_add);
|
||||
scalars
|
||||
.saturating_add(list_offsets)
|
||||
.saturating_add(custom)
|
||||
.saturating_add(sides)
|
||||
}
|
||||
|
||||
/// The last timestamp of `spec`, which is inclusive: its `len` points step from its start.
|
||||
///
|
||||
/// [`GridSpec::try_new`] checked that the point fits `i64`, so widening it here cannot overflow.
|
||||
fn grid_end_ms(spec: GridSpec) -> i64 {
|
||||
(i128::from(spec.start_ms) + i128::from(spec.step_ms) * (i128::from(spec.len) - 1)) as i64
|
||||
}
|
||||
|
||||
/// The position of the grid point at `timestamp`, which the selection reported as one of its own.
|
||||
///
|
||||
/// The selection steps the range it is given by the step it is given, so every timestamp it reports
|
||||
/// is `start_ms + position * step_ms` for one position of the grid: subtracting the start and
|
||||
/// dividing by the step gives that position back. A timestamp off that lattice, or one past the end
|
||||
/// of the grid, is a disagreement between the selection and its binding rather than a sample to
|
||||
/// skip, so it is reported instead of dropped.
|
||||
fn grid_position(grid: &EvalGrid, timestamp: i64) -> Result<usize> {
|
||||
let spec = grid.spec();
|
||||
let offset = i128::from(timestamp) - i128::from(spec.start_ms);
|
||||
let step = i128::from(spec.step_ms);
|
||||
let position = offset.div_euclid(step);
|
||||
if offset.rem_euclid(step) != 0 || position < 0 || position >= i128::from(spec.len) {
|
||||
return Err(DataFusionError::Execution(format!(
|
||||
"the instant selection reported the timestamp {timestamp}ms, which is not a point of \
|
||||
its grid {spec:?}"
|
||||
)));
|
||||
}
|
||||
Ok(position as usize)
|
||||
}
|
||||
|
||||
/// A closing chunk carrying `metric` and no row at all.
|
||||
///
|
||||
/// A series whose selection picked no source row still closes exactly once, on a chunk that holds
|
||||
/// nothing: the labels say which series closed, and the address says it closed before its first
|
||||
/// point, as the empty chunk a window hands out for such a series does.
|
||||
fn no_row_chunk(schema: &SeriesSchema, metric: MetricName) -> Result<EvaluatedChunk> {
|
||||
let grid = schema.eval_grid.clone().ok_or_else(|| {
|
||||
DataFusionError::Execution(
|
||||
"an instant selection produces an evaluated chunk, which is addressed on a grid"
|
||||
.to_string(),
|
||||
)
|
||||
})?;
|
||||
let empty = Samples::from_float(Float64Array::from(Vec::<f64>::new()));
|
||||
let fields =
|
||||
SeriesFields::try_from_vec(vec![empty; schema.fields.len()]).map_err(series_error)?;
|
||||
let time = GridRange::try_new(grid, 0, 0).map_err(series_error)?;
|
||||
let series =
|
||||
Series::try_new(schema, metric, TimeAxis::Grid(time), fields).map_err(series_error)?;
|
||||
EvaluatedChunk::try_new(series, true)
|
||||
}
|
||||
|
||||
/// The per-node run-time state of a node.
|
||||
struct NodeRuntime {
|
||||
buffer: NodeBuffer,
|
||||
@@ -846,6 +1290,9 @@ struct NodeRuntime {
|
||||
/// 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)>,
|
||||
/// 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<InstantRuntime>,
|
||||
/// A window node's evaluator, with its own input and output grid; `None` before it is built and
|
||||
/// once it is closed.
|
||||
window: Option<RangeEvaluator>,
|
||||
@@ -1029,6 +1476,95 @@ impl NodeRuntime {
|
||||
Ok(output)
|
||||
}
|
||||
|
||||
/// Feeds one chunk to this node's instant selection.
|
||||
///
|
||||
/// `None` means the series this chunk belongs to has not ended: the selection reads one whole
|
||||
/// axis, so it decides nothing until the chunk carrying `is_last` arrives.
|
||||
fn instant(&mut self, path: &NodePath, chunk: NativeChunk) -> Result<Option<EvaluatedChunk>> {
|
||||
let NativeChunk::Raw(chunk) = chunk else {
|
||||
return Err(DataFusionError::Execution(format!(
|
||||
"the instant selection at {path} consumes raw samples, not an evaluated chunk"
|
||||
)));
|
||||
};
|
||||
let (positions, closing) = (chunk.time.len(), chunk.is_last);
|
||||
// The chunk is held from here on: a selection keeps every chunk of a series until the one
|
||||
// that closes it, so what this node reserves for is the series it has buffered.
|
||||
self.buffer.grow(raw_bytes(&chunk))?;
|
||||
// The chunk that closes the series gathers it in one go, and the gather allocates a
|
||||
// workspace of its own beside the buffered chunks: the axis it concatenates, the descriptors
|
||||
// that read it, the two arrays the selection returns and the rows the builder keeps. That
|
||||
// workspace is reserved here, before it exists, so a pool too small for it refuses the
|
||||
// gather instead of learning about it afterwards. Without this the reservation would fall
|
||||
// back to the buffered chunks alone -- nothing at all, on the chunk that ends the series --
|
||||
// and a selection over a million grid points would take its tens of megabytes without
|
||||
// asking. [`InstantRuntime::gather_bytes`] derives the figure, because the chunks it counts
|
||||
// are that node's.
|
||||
let Some(instant) = self.instant.as_ref() else {
|
||||
return Err(DataFusionError::Execution(format!(
|
||||
"the node at {path} holds no instant selection"
|
||||
)));
|
||||
};
|
||||
let scratch = if closing {
|
||||
instant.gather_bytes(&chunk)
|
||||
} else {
|
||||
0
|
||||
};
|
||||
self.buffer.grow(scratch)?;
|
||||
let metrics = Arc::clone(&self.buffer.metrics);
|
||||
let outcome = {
|
||||
// The timer covers this node's own work only: it is dropped before the driver polls
|
||||
// the input again.
|
||||
let _timer = metrics.timer();
|
||||
match self.instant.as_mut() {
|
||||
Some(instant) => instant.push(path, chunk),
|
||||
// Rejected when the state was built, since a selection reads raw samples; kept as
|
||||
// an error rather than a panic so an impossible tree cannot take the process down.
|
||||
None => Err(DataFusionError::Execution(format!(
|
||||
"the node at {path} holds no instant selection"
|
||||
))),
|
||||
}
|
||||
};
|
||||
metrics.record_input(1, positions);
|
||||
if closing {
|
||||
// A series closes exactly once, on the chunk carrying `is_last`: the selection gathers
|
||||
// the whole series on it, so it is counted here rather than on the chunk handed out.
|
||||
metrics.record_completed_series(1);
|
||||
}
|
||||
// A failed gather gives its workspace back here rather than leaving it to the driver: the
|
||||
// driver releases every node when the execution stops, but the workspace is this call's and
|
||||
// not something the node keeps, so holding the pool's budget for it until the stopped
|
||||
// executor is dropped would charge the pool for memory nothing reads any more.
|
||||
let output = match outcome {
|
||||
Ok(output) => output,
|
||||
Err(error) => {
|
||||
self.settle_buffered()?;
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
if let Some(output) = &output {
|
||||
metrics.record_grid_points(output.series.time.len());
|
||||
metrics.record_output(1, valid_values(output));
|
||||
}
|
||||
// What this node still holds is the chunks of the series it is selecting: a whole series is
|
||||
// held until it ends, so the reservation is what those chunks hold rather than a figure
|
||||
// that grows with the chunk count of anything already answered. The chunk just consumed and
|
||||
// the workspace of the gather it closed are both gone, so this is what gives them back.
|
||||
self.settle_buffered()?;
|
||||
Ok(output)
|
||||
}
|
||||
|
||||
/// Sets this node's reservation to the chunks of the series it is still gathering.
|
||||
///
|
||||
/// The chunk the node just consumed and the workspace of a gather it closed are not part of it:
|
||||
/// the first has left the buffer and the second has been handed to the output.
|
||||
fn settle_buffered(&mut self) -> Result<()> {
|
||||
let buffered = self
|
||||
.instant
|
||||
.as_ref()
|
||||
.map_or(0, InstantRuntime::buffered_bytes);
|
||||
self.buffer.resize(buffered)
|
||||
}
|
||||
|
||||
/// Keeps one fragment of this node's collected input, charged for as it arrives.
|
||||
///
|
||||
/// The fragment is measured once, here, and its estimate stands until the flush that merges it
|
||||
@@ -1124,12 +1660,13 @@ impl NodeRuntime {
|
||||
|
||||
/// Returns everything this node holds, including queued chunks.
|
||||
fn release(&mut self) {
|
||||
// The window goes first: dropping the evaluator is what gives back the memory its
|
||||
// reservation was taken for. Returning the reservation while the evaluator still held the
|
||||
// samples would let the pool admit memory the process has not released, for as long as the
|
||||
// caller keeps the stopped executor alive. The label sets this node handed out and the
|
||||
// fragments it collected go with it.
|
||||
// A node's own state goes first: dropping the evaluator and the buffered series is what
|
||||
// gives back the memory their reservation was taken for. Returning the reservation while
|
||||
// they still held the samples would let the pool admit memory the process has not released,
|
||||
// for as long as the caller keeps the stopped executor alive. The label sets this node
|
||||
// handed out and the fragments it collected go with it.
|
||||
self.window = None;
|
||||
self.instant = None;
|
||||
self.emitted = None;
|
||||
self.merge = None;
|
||||
// A node that holds no chunk holds no series either, so nothing it handed over is still
|
||||
@@ -1213,6 +1750,7 @@ impl NativeExecutionState {
|
||||
source = Some((index, samples));
|
||||
(NodeTag::Input, None)
|
||||
}
|
||||
NativeExpr::Instant { input, .. } => (NodeTag::Instant, Some(input.as_ref())),
|
||||
NativeExpr::Window { input, .. } => (NodeTag::Window, Some(input.as_ref())),
|
||||
NativeExpr::Map { input, .. } => (NodeTag::Map, Some(input.as_ref())),
|
||||
// The kinds below are legal in a bound tree, and `validate` stays the authority on
|
||||
@@ -1220,7 +1758,6 @@ impl NativeExecutionState {
|
||||
// chunk is read: a tree the runtime cannot execute must not start, because a
|
||||
// partial answer would be one the caller never asked for.
|
||||
NativeExpr::Scalar(_) => return Err(not_implemented("Scalar", &path)),
|
||||
NativeExpr::Instant { .. } => return Err(not_implemented("Instant", &path)),
|
||||
NativeExpr::Aggregate { .. } => return Err(not_implemented("Aggregate", &path)),
|
||||
NativeExpr::Binary { .. } => return Err(not_implemented("Binary", &path)),
|
||||
};
|
||||
@@ -1228,6 +1765,7 @@ impl NativeExecutionState {
|
||||
let mut runtime = NodeRuntime {
|
||||
buffer,
|
||||
queued: VecDeque::new(),
|
||||
instant: None,
|
||||
window: None,
|
||||
map: None,
|
||||
emitted: None,
|
||||
@@ -1236,6 +1774,19 @@ impl NativeExecutionState {
|
||||
};
|
||||
match &node.expr {
|
||||
NativeExpr::Input(_) => {}
|
||||
NativeExpr::Instant { input, op } => {
|
||||
let input_schema = input.output.samples().ok_or_else(|| {
|
||||
DataFusionError::Execution(format!(
|
||||
"the instant selection at {path} consumes samples, but its input \
|
||||
produces none"
|
||||
))
|
||||
})?;
|
||||
runtime.instant = Some(InstantRuntime::try_new(input_schema, op)?);
|
||||
// A selection keeps the labels of the series it reads, empty-valued ones
|
||||
// included, so it is not one of the nodes that can turn two series into one
|
||||
// label set and it keeps no table of what it handed out: the node that drops a
|
||||
// label is the one that can collide, and it records its own.
|
||||
}
|
||||
NativeExpr::Map { op, .. } => {
|
||||
let schema = node.output.samples().cloned().ok_or_else(|| {
|
||||
DataFusionError::Execution(format!(
|
||||
@@ -1271,7 +1822,6 @@ impl NativeExecutionState {
|
||||
runtime.emitted = Some(Emitted::default());
|
||||
}
|
||||
NativeExpr::Scalar(_)
|
||||
| NativeExpr::Instant { .. }
|
||||
| NativeExpr::Aggregate { .. }
|
||||
| NativeExpr::Binary { .. } => {
|
||||
// Rejected above, where the pass has the position to name.
|
||||
@@ -1321,6 +1871,17 @@ impl NativeExecutionState {
|
||||
nodes[index].input.is_some_and(|child| holds_shared[child]);
|
||||
match nodes[index].tag {
|
||||
NodeTag::Input => {}
|
||||
NodeTag::Instant => {
|
||||
// A selection keeps the labels it read, so it is not where two series
|
||||
// become one label set and its own output carries its input's flag. Its
|
||||
// input is the source -- `validate` binds a selection to raw samples, and
|
||||
// no node above a source produces them -- and a source hands each series'
|
||||
// chunks over together, so the flag is false there in every tree this
|
||||
// runtime runs. Should the pass ever compute it true, the selection still
|
||||
// needs no collected input: it reads one series at a time and refuses a
|
||||
// chunk of another before it is buffered.
|
||||
holds_shared[index] = input_holds_shared;
|
||||
}
|
||||
NodeTag::Map => holds_shared[index] = true,
|
||||
NodeTag::Window => {
|
||||
let (drops_metric_name, input_schema) = match expr {
|
||||
@@ -1414,6 +1975,19 @@ impl NativeExecutionState {
|
||||
.count()
|
||||
}
|
||||
|
||||
/// How many nodes still hold the chunks of a series an instant selection was gathering.
|
||||
///
|
||||
/// Test-only, and for the same reason as [`Self::nodes_holding_windows`]: the buffered chunks
|
||||
/// have to be dropped, not only the bytes they were reserved for, and that is not observable
|
||||
/// from the pool.
|
||||
#[cfg(test)]
|
||||
pub(crate) fn nodes_holding_instants(&self) -> usize {
|
||||
self.runtimes
|
||||
.iter()
|
||||
.filter(|runtime| runtime.instant.is_some())
|
||||
.count()
|
||||
}
|
||||
|
||||
/// The input that ended before its last chunk, if there is one.
|
||||
///
|
||||
/// It is read before anything is merged: a window that collects its input closes every series
|
||||
@@ -1729,6 +2303,19 @@ impl NativeExecutor {
|
||||
}
|
||||
}
|
||||
}
|
||||
NodeTag::Instant => {
|
||||
let child = input.ok_or_else(|| missing_input(tag))?;
|
||||
// A selection decides a whole series at once, so every chunk before the one that
|
||||
// closes it yields nothing and the selection asks its input again.
|
||||
loop {
|
||||
let Some(chunk) = self.pull(child)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
if let Some(output) = self.instant_node(index, chunk)? {
|
||||
return Ok(Some(NativeChunk::Evaluated(output)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1752,6 +2339,13 @@ impl NativeExecutor {
|
||||
runtime.window(path, chunk)
|
||||
}
|
||||
|
||||
/// Feeds one chunk to the instant selection at `index`.
|
||||
fn instant_node(&mut self, index: usize, chunk: NativeChunk) -> Result<Option<EvaluatedChunk>> {
|
||||
let path = &self.state.paths[index];
|
||||
let runtime = &mut self.state.runtimes[index];
|
||||
runtime.instant(path, chunk)
|
||||
}
|
||||
|
||||
/// Queues a source chunk at the tree's input node.
|
||||
fn push_source(&mut self, chunk: NativeChunk) -> Result<()> {
|
||||
self.check_running()?;
|
||||
@@ -1881,6 +2475,39 @@ fn closed(path: &NodePath) -> DataFusionError {
|
||||
DataFusionError::Execution(format!("the window at {path} is already closed"))
|
||||
}
|
||||
|
||||
/// The error for a chunk of another series arriving while one is being selected.
|
||||
///
|
||||
/// [`select_instants`] reads one whole axis, so the buffered chunks have to be one series: a chunk
|
||||
/// of another series has nowhere to go, and appending it would select its samples as though they
|
||||
/// were the open series'.
|
||||
fn series_changed(path: &NodePath) -> DataFusionError {
|
||||
DataFusionError::Execution(format!(
|
||||
"the instant selection at {path} combines one series into one axis, and a chunk of another \
|
||||
series arrived before the first one ended"
|
||||
))
|
||||
}
|
||||
|
||||
/// The error for a series an instant selection has no labels to report it under.
|
||||
///
|
||||
/// A source bound in label mode carries its identity encoded and no decoded labels, and a chunk
|
||||
/// that is handed on has to name the series it holds.
|
||||
fn no_labels(path: &NodePath) -> DataFusionError {
|
||||
DataFusionError::Execution(format!(
|
||||
"the instant selection at {path} requires materialized series labels"
|
||||
))
|
||||
}
|
||||
|
||||
/// The error for a source row the selection named but the series does not hold.
|
||||
///
|
||||
/// The selection returns indices into the axis it was given, so a row outside it means the two
|
||||
/// disagree rather than that a sample is missing.
|
||||
fn missing_row(path: &NodePath, index: usize) -> DataFusionError {
|
||||
DataFusionError::Execution(format!(
|
||||
"the instant selection at {path} selected the source row {index}, which the series it read \
|
||||
does not hold"
|
||||
))
|
||||
}
|
||||
|
||||
/// The rejection of an input ordering this runtime cannot execute.
|
||||
///
|
||||
/// It names the input that declares the ordering and the ordering found, so a reader can tell a
|
||||
|
||||
@@ -35,10 +35,11 @@ use datatypes::series::{Label, MetricName, SeriesOrdering};
|
||||
|
||||
use super::{
|
||||
Emitted, NativeChunk, NativeExecutionState, NativeExecutor, NodeBuffer, NodeRuntime,
|
||||
absorb_rebuilds, evaluated_bytes, map_chunk, raw_bytes,
|
||||
absorb_rebuilds, evaluated_bytes, grid_end_ms, map_chunk, raw_bytes,
|
||||
};
|
||||
use crate::evaluated::EvaluatedChunk;
|
||||
use crate::grid_series::merge_same_labelset;
|
||||
use crate::instant_eval::select_instants;
|
||||
use crate::map::MathOp;
|
||||
use crate::native::metrics::{NodeMetric, NodeMetrics, NodePath};
|
||||
use crate::native::plan::{
|
||||
@@ -93,11 +94,16 @@ fn metric() -> MetricName {
|
||||
|
||||
/// The raw source every supported tree starts from.
|
||||
fn input() -> NativeNode {
|
||||
raw_input_with_ordering(SeriesOrdering::KeyThenTime)
|
||||
}
|
||||
|
||||
/// The same raw source, declaring `ordering`.
|
||||
fn raw_input_with_ordering(ordering: SeriesOrdering) -> NativeNode {
|
||||
let output = NativeType::Samples(raw_schema());
|
||||
NativeNode {
|
||||
expr: NativeExpr::Input(BoundInput {
|
||||
output: output.clone(),
|
||||
ordering: SeriesOrdering::KeyThenTime,
|
||||
ordering,
|
||||
}),
|
||||
output,
|
||||
}
|
||||
@@ -193,6 +199,39 @@ fn window_over_fields(count: usize, output_grid: Arc<EvalGrid>) -> NativeNode {
|
||||
}
|
||||
}
|
||||
|
||||
/// A bare selector over `inner`: the last sample at or before each grid point, read within
|
||||
/// `lookback_ms` and at `offset_ms`.
|
||||
///
|
||||
/// The selection reports one value per input field, so its layout is the input's, just as the
|
||||
/// compiler builds it.
|
||||
fn instant(
|
||||
inner: NativeNode,
|
||||
lookback_ms: i64,
|
||||
offset_ms: i64,
|
||||
output_grid: Arc<EvalGrid>,
|
||||
) -> NativeNode {
|
||||
let fields = inner
|
||||
.output
|
||||
.samples()
|
||||
.expect("a selection reads samples")
|
||||
.fields
|
||||
.clone();
|
||||
NativeNode {
|
||||
expr: NativeExpr::Instant {
|
||||
input: Box::new(inner),
|
||||
op: BoundInstant {
|
||||
lookback_ms,
|
||||
offset_ms,
|
||||
output_grid: Arc::clone(&output_grid),
|
||||
},
|
||||
},
|
||||
output: NativeType::Samples(Arc::new(SeriesSchema {
|
||||
fields,
|
||||
eval_grid: Some(output_grid),
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
fn map(inner: NativeNode, ops: Vec<MathOp>) -> NativeNode {
|
||||
let output = inner.output.grid().expect("a map needs a grid").clone();
|
||||
NativeNode {
|
||||
@@ -265,6 +304,34 @@ fn raw_chunk_repeated(
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// One raw chunk of `series-a` whose fields carry `columns`, one column per field.
|
||||
///
|
||||
/// A `None` is a position that field holds no sample at, which is what a partially filled field
|
||||
/// looks like at the source.
|
||||
fn raw_chunk_columns(
|
||||
schema: &SeriesSchema,
|
||||
timestamps: &[i64],
|
||||
columns: Vec<Vec<Option<f64>>>,
|
||||
is_last: bool,
|
||||
) -> RawSeriesChunk {
|
||||
let fields = SeriesFields::try_from_vec(
|
||||
columns
|
||||
.into_iter()
|
||||
.map(|values| Samples::from_float(Float64Array::from(values)))
|
||||
.collect(),
|
||||
)
|
||||
.unwrap();
|
||||
RawSeriesChunk::try_new(
|
||||
schema,
|
||||
Buffer::from_slice_ref(b"series-a"),
|
||||
Some(metric()),
|
||||
RawTimestamps::try_new(ScalarBuffer::from(timestamps.to_vec())).unwrap(),
|
||||
fields,
|
||||
is_last,
|
||||
)
|
||||
.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 {
|
||||
@@ -320,6 +387,59 @@ fn raw_histogram_chunk(
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// One raw chunk of `series-a` whose one field holds a four-bucket native histogram at each of
|
||||
/// `timestamps`, each carrying `start_timestamp` as the payload it was read from reports it.
|
||||
fn raw_histogram_chunk_with_start(
|
||||
schema: &SeriesSchema,
|
||||
timestamps: &[i64],
|
||||
start_timestamp: Option<i64>,
|
||||
is_last: bool,
|
||||
) -> RawSeriesChunk {
|
||||
let mut builder = HistogramBatchBuilder::new();
|
||||
let offsets = [0];
|
||||
let lengths = [4];
|
||||
let counts = vec![1.0; 4];
|
||||
let indices: UInt32Array = timestamps
|
||||
.iter()
|
||||
.map(|_| {
|
||||
Some(
|
||||
builder
|
||||
.push(HistogramView {
|
||||
schema: 0,
|
||||
zero_threshold: 0.0,
|
||||
zero_count: 0.0,
|
||||
count: 4.0,
|
||||
sum: 4.0,
|
||||
reset_hint: CounterResetHint::Unknown,
|
||||
start_timestamp,
|
||||
custom_values: &[],
|
||||
positive: BucketSideView {
|
||||
span_offsets: &offsets,
|
||||
span_lengths: &lengths,
|
||||
buckets: &counts,
|
||||
},
|
||||
negative: BucketSideView {
|
||||
span_offsets: &[],
|
||||
span_lengths: &[],
|
||||
buckets: &[],
|
||||
},
|
||||
})
|
||||
.unwrap(),
|
||||
)
|
||||
})
|
||||
.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 raw chunk of `series-a` whose one field holds the histogram at `index` of `table`.
|
||||
///
|
||||
/// Two chunks built this way share one `Arc<HistogramBatch>`, which is what makes the allocation
|
||||
@@ -354,6 +474,7 @@ fn source_runtime(pool: &Arc<dyn MemoryPool>) -> NodeRuntime {
|
||||
pool,
|
||||
),
|
||||
queued: VecDeque::new(),
|
||||
instant: None,
|
||||
window: None,
|
||||
map: None,
|
||||
emitted: None,
|
||||
@@ -458,11 +579,16 @@ fn run(executor: &mut NativeExecutor) -> Vec<EvaluatedChunk> {
|
||||
}
|
||||
|
||||
fn sample_values(chunk: &EvaluatedChunk) -> Vec<Option<f64>> {
|
||||
field_values(chunk, 0)
|
||||
}
|
||||
|
||||
/// The values field `index` carries, one per row the chunk holds.
|
||||
fn field_values(chunk: &EvaluatedChunk, index: usize) -> Vec<Option<f64>> {
|
||||
let field = chunk
|
||||
.series
|
||||
.fields
|
||||
.get(0)
|
||||
.expect("these series have one field");
|
||||
.get(index)
|
||||
.unwrap_or_else(|| panic!("the chunk carries no field {index}"));
|
||||
field
|
||||
.iter()
|
||||
.map(|sample| match sample {
|
||||
@@ -472,6 +598,25 @@ fn sample_values(chunk: &EvaluatedChunk) -> Vec<Option<f64>> {
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The start timestamp of the histogram at every position of a chunk, in grid order.
|
||||
///
|
||||
/// It is read off the output's own histogram views rather than off a converted value, because the
|
||||
/// figure under test is the one a later reader sees.
|
||||
fn histogram_starts(chunk: &EvaluatedChunk) -> Vec<Option<i64>> {
|
||||
let mut cursor = chunk.series.fields.as_slice()[0].cursor();
|
||||
chunk
|
||||
.series
|
||||
.time
|
||||
.positions()
|
||||
.filter_map(
|
||||
|(_, row)| match cursor.at(row).expect("the axis and the fields are aligned") {
|
||||
SampleRef::Histogram(view) => Some(view.start_timestamp),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn series_start(chunk: &EvaluatedChunk) -> usize {
|
||||
chunk
|
||||
.series
|
||||
@@ -828,23 +973,6 @@ fn an_unimplemented_node_is_reported_not_panicked() {
|
||||
let statement = || grid(0, 120_000, 2);
|
||||
|
||||
let scalar_tree = scalar(1.0, statement());
|
||||
let instant_tree = {
|
||||
let out = statement();
|
||||
map(
|
||||
NativeNode {
|
||||
expr: NativeExpr::Instant {
|
||||
input: Box::new(input()),
|
||||
op: BoundInstant {
|
||||
lookback_ms: 300_000,
|
||||
offset_ms: 0,
|
||||
output_grid: Arc::clone(&out),
|
||||
},
|
||||
},
|
||||
output: NativeType::Samples(evaluated_schema(out)),
|
||||
},
|
||||
vec![MathOp::Abs],
|
||||
)
|
||||
};
|
||||
let aggregate_tree = {
|
||||
let out = statement();
|
||||
NativeNode {
|
||||
@@ -879,7 +1007,6 @@ fn an_unimplemented_node_is_reported_not_panicked() {
|
||||
|
||||
let cases = [
|
||||
("Scalar", "root", scalar_tree),
|
||||
("Instant", "root.input", instant_tree),
|
||||
("Aggregate", "root", aggregate_tree),
|
||||
("Binary", "root", binary_tree),
|
||||
];
|
||||
@@ -2464,6 +2591,68 @@ fn an_interleaved_input_is_refused_without_a_window() {
|
||||
assert!(message.contains("root.input"), "{message}");
|
||||
}
|
||||
|
||||
/// A selection over an interleaved input is refused when the tree is bound, before it can run.
|
||||
///
|
||||
/// The binding is where the ordering is known, and the plan's `Instant` arm refuses an input that
|
||||
/// declares `Interleaved` for the reason the window arm does: a selection's consumer collects one
|
||||
/// series' chunks into one buffer, so it reads a series to its last chunk before the next. Nothing
|
||||
/// about an interleaved input is therefore left to the runtime along a selection, and the tree
|
||||
/// never reaches [`NativeExecutionState::try_new`]. The ordering check there stays for the shape
|
||||
/// the binding cannot see through: it only looks at a window's own input, so a window over a map
|
||||
/// over the source still reads an interleaved stream, which is
|
||||
/// [`a_window_over_an_interleaved_input_is_rejected`]. The shape with a map above the selection is
|
||||
/// [`an_interleaved_input_is_refused_under_a_map_over_a_selection`].
|
||||
#[test]
|
||||
fn an_interleaved_input_is_refused_under_a_selection() {
|
||||
let out_grid = grid(0, 60_000, 2);
|
||||
let error = NativeTree::try_new(instant(
|
||||
raw_input_with_ordering(SeriesOrdering::Interleaved),
|
||||
60_000,
|
||||
0,
|
||||
Arc::clone(&out_grid),
|
||||
))
|
||||
.expect_err("a selection over an interleaved input must not bind");
|
||||
// A rejected tree is a compiler defect rather than a query error, so the binding reports it as
|
||||
// `Internal`; what this pins is that the refusal names the ordering it came from, since the
|
||||
// same check refuses a window over the same input.
|
||||
assert!(matches!(error, DataFusionError::Internal(_)), "{error}");
|
||||
let message = error.to_string();
|
||||
assert!(
|
||||
message.contains("interleaved chunks cannot be fed to it"),
|
||||
"{message}"
|
||||
);
|
||||
}
|
||||
|
||||
/// The same refusal with a map above the selection.
|
||||
///
|
||||
/// The tree here is `Input -> Instant -> Map`: the map sits above the selection, and the selection
|
||||
/// reads the interleaved input directly. A map between the input and the selection is not a tree
|
||||
/// this compiler can build at all -- a `Map` produces an evaluated vector while an instant
|
||||
/// selection requires raw samples, and the plan's `Instant` arm rejects evaluated values -- so what
|
||||
/// this test pins is the recursion: `validate` descends through the outer map to the selection
|
||||
/// below it, whose own arm then refuses the input it reads, and the tree is refused at the same
|
||||
/// point as [`an_interleaved_input_is_refused_under_a_selection`].
|
||||
#[test]
|
||||
fn an_interleaved_input_is_refused_under_a_map_over_a_selection() {
|
||||
let out_grid = grid(0, 60_000, 2);
|
||||
let error = NativeTree::try_new(map(
|
||||
instant(
|
||||
raw_input_with_ordering(SeriesOrdering::Interleaved),
|
||||
60_000,
|
||||
0,
|
||||
Arc::clone(&out_grid),
|
||||
),
|
||||
vec![MathOp::Abs],
|
||||
))
|
||||
.expect_err("a map over a selection over an interleaved input must not bind");
|
||||
assert!(matches!(error, DataFusionError::Internal(_)), "{error}");
|
||||
let message = error.to_string();
|
||||
assert!(
|
||||
message.contains("interleaved chunks cannot be fed to it"),
|
||||
"{message}"
|
||||
);
|
||||
}
|
||||
|
||||
/// A series that produced values under a label set, followed by one that produced none under the
|
||||
/// same label set, is not a collision: the second contributes no sample, so the answer is just the
|
||||
/// first one's values.
|
||||
@@ -3186,3 +3375,651 @@ fn a_sparse_series_that_moves_forward_is_appended_to_its_history() {
|
||||
"the merged history is not sorted, disjoint and joined: {runs:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// A bare selector stamps the last sample at or before each grid point onto its own grid.
|
||||
///
|
||||
/// The chunk is addressed on the node's output grid, not on the source's timestamps; a grid point
|
||||
/// may select a row an earlier point already did (the point after the last sample repeats it); and
|
||||
/// the same samples delivered as one chunk and as three answer the same, because the selection reads
|
||||
/// a whole axis rather than a chunk.
|
||||
#[test]
|
||||
fn a_bare_selector_selects_instants_onto_its_own_grid() {
|
||||
let out_grid = grid(0, 60_000, 4);
|
||||
let tree = NativeTree::try_new(instant(input(), 120_000, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let schema = raw_schema();
|
||||
let whole = raw_chunk(
|
||||
&schema,
|
||||
&[0, 30_000, 90_000, 120_000, 200_000],
|
||||
&[1.0, 2.0, 3.0, 4.0, 5.0],
|
||||
true,
|
||||
);
|
||||
// The same series as three chunks, so nothing about it is decided by a chunk boundary.
|
||||
let split = [
|
||||
raw_chunk(&schema, &[0, 30_000], &[1.0, 2.0], false),
|
||||
raw_chunk(&schema, &[90_000], &[3.0], false),
|
||||
raw_chunk(&schema, &[120_000, 200_000], &[4.0, 5.0], true),
|
||||
];
|
||||
|
||||
let mut once = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
once.push_raw(whole).unwrap();
|
||||
let out = run(&mut once);
|
||||
|
||||
let mut piecewise = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
let mut in_pieces = Vec::new();
|
||||
for chunk in split {
|
||||
let closing = chunk.is_last;
|
||||
piecewise.push_raw(chunk).unwrap();
|
||||
// Until the series ends nothing is decided for it, so only the closing chunk reports.
|
||||
let pulled = piecewise.next_chunk().unwrap();
|
||||
assert_eq!(
|
||||
pulled.is_some(),
|
||||
closing,
|
||||
"a series reported a chunk before its last one arrived"
|
||||
);
|
||||
in_pieces.extend(pulled);
|
||||
}
|
||||
in_pieces.extend(piecewise.finish().unwrap());
|
||||
|
||||
assert_eq!(out.len(), 1, "one series is one chunk, however it arrived");
|
||||
let chunk = &out[0];
|
||||
assert!(chunk.is_last, "the only chunk of a series closes it");
|
||||
assert!(
|
||||
Arc::ptr_eq(chunk.series.time.grid().expect("a grid chunk"), &out_grid),
|
||||
"the chunk is addressed on a grid that is not the selection's output grid"
|
||||
);
|
||||
// 0 -> 0, 60s -> 30s, 120s -> 120s, 180s -> 120s: the row at 90s is never the last one before a
|
||||
// grid point, and the point past the last sample repeats the row before it.
|
||||
assert_eq!(
|
||||
decided(&out),
|
||||
vec![
|
||||
(0, Some(1.0)),
|
||||
(1, Some(2.0)),
|
||||
(2, Some(4.0)),
|
||||
(3, Some(4.0)),
|
||||
]
|
||||
);
|
||||
// A selection keeps the labels it read, the metric name included: unlike a range function it
|
||||
// does not rewrite what the selector named.
|
||||
assert_eq!(chunk.series.metric, metric());
|
||||
assert_eq!(
|
||||
decided(&in_pieces),
|
||||
decided(&out),
|
||||
"the same series split into chunks answered differently"
|
||||
);
|
||||
}
|
||||
|
||||
/// The lookback edge is exclusive: a sample exactly `lookback_ms` before a grid point is not
|
||||
/// selected, one millisecond later is.
|
||||
///
|
||||
/// This pins what the selection kernel does rather than what Prometheus' documentation suggests: a
|
||||
/// sample at the point itself is always selected, and the window is `(point - lookback, point]`.
|
||||
#[test]
|
||||
fn a_sample_at_the_lookback_edge_is_not_selected() {
|
||||
// The statement's grid is the single point 1000ms, and the lookback is 200ms.
|
||||
let out_grid = grid(1000, 100, 1);
|
||||
let schema = raw_schema();
|
||||
let selected = |timestamp: i64| {
|
||||
let tree = NativeTree::try_new(instant(input(), 200, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[timestamp], &[1.0], true))
|
||||
.unwrap();
|
||||
run(&mut execution)
|
||||
};
|
||||
|
||||
// 800 is exactly `1000 - 200`, so the edge itself selects nothing.
|
||||
let edge = selected(800);
|
||||
assert_eq!(edge.len(), 1, "the series still closes on one chunk");
|
||||
assert!(edge[0].is_last);
|
||||
assert!(
|
||||
decided(&edge).is_empty(),
|
||||
"a sample exactly at the lookback edge was selected: {:?}",
|
||||
decided(&edge)
|
||||
);
|
||||
// One millisecond inside the window is selected; a sample after the grid point never is.
|
||||
assert_eq!(decided(&selected(801)), vec![(0, Some(1.0))]);
|
||||
assert!(decided(&selected(1001)).is_empty());
|
||||
}
|
||||
|
||||
/// A positive offset reads earlier samples, and a negative one later samples.
|
||||
#[test]
|
||||
fn an_offset_reads_a_different_sample_at_each_grid_point() {
|
||||
let out_grid = grid(0, 60_000, 2);
|
||||
let schema = raw_schema();
|
||||
let values = |offset_ms: i64| {
|
||||
let tree = NativeTree::try_new(instant(input(), 600_000, offset_ms, Arc::clone(&out_grid)))
|
||||
.unwrap();
|
||||
let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
execution
|
||||
.push_raw(raw_chunk(
|
||||
&schema,
|
||||
&[-60_000, 0, 60_000],
|
||||
&[1.0, 2.0, 3.0],
|
||||
true,
|
||||
))
|
||||
.unwrap();
|
||||
decided(&run(&mut execution))
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
values(0),
|
||||
vec![(0, Some(2.0)), (1, Some(3.0))],
|
||||
"without an offset a point reads the sample at or before it"
|
||||
);
|
||||
assert_eq!(
|
||||
values(60_000),
|
||||
vec![(0, Some(1.0)), (1, Some(2.0))],
|
||||
"a minute of offset did not read the samples a minute earlier"
|
||||
);
|
||||
// A negative offset reads samples the statement's own range does not reach: both points read
|
||||
// the sample one minute after them.
|
||||
assert_eq!(
|
||||
values(-60_000),
|
||||
vec![(0, Some(3.0)), (1, Some(3.0))],
|
||||
"a negative offset did not read the samples a minute later"
|
||||
);
|
||||
}
|
||||
|
||||
/// A series that selects nothing still closes, exactly once, on a chunk that holds no row.
|
||||
#[test]
|
||||
fn a_series_with_no_selectable_sample_closes_once() {
|
||||
// Every sample carries the stale marker, which suppresses the selection at every point.
|
||||
let stale = f64::from_bits(common_query::prometheus::PROMETHEUS_STALE_NAN_BITS);
|
||||
let out_grid = grid(0, 60_000, 3);
|
||||
let tree = NativeTree::try_new(instant(input(), 300_000, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let set = ExecutionPlanMetricsSet::new();
|
||||
let mut execution = executor(&tree, pool(), &set);
|
||||
let schema = raw_schema();
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[0, 60_000], &[stale, stale], true))
|
||||
.unwrap();
|
||||
let out = run(&mut execution);
|
||||
|
||||
assert_eq!(out.len(), 1, "the series has to close exactly once");
|
||||
let chunk = &out[0];
|
||||
assert!(chunk.is_last);
|
||||
assert!(
|
||||
chunk.series.time.is_empty(),
|
||||
"the closing chunk of a series that selected nothing holds a row"
|
||||
);
|
||||
assert_eq!(
|
||||
series_start(chunk),
|
||||
0,
|
||||
"the empty chunk is not addressed on the grid"
|
||||
);
|
||||
assert!(
|
||||
Arc::ptr_eq(chunk.series.time.grid().expect("a grid chunk"), &out_grid),
|
||||
"the empty chunk is addressed on a grid that is not the selection's output grid"
|
||||
);
|
||||
assert_eq!(
|
||||
chunk.series.metric,
|
||||
metric(),
|
||||
"the series lost its identity"
|
||||
);
|
||||
assert_eq!(
|
||||
count(&set, &NodePath::root(), NodeMetric::COMPLETED_SERIES),
|
||||
1,
|
||||
"the series did not close exactly once"
|
||||
);
|
||||
}
|
||||
|
||||
/// A row one field fills and another misses stays one row, with a null where the field misses it.
|
||||
#[test]
|
||||
fn a_selected_row_keeps_the_fields_that_miss_it_null() {
|
||||
let out_grid = grid(0, 60_000, 2);
|
||||
let tree =
|
||||
NativeTree::try_new(instant(raw_input_of(2), 60_000, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let schema = raw_schema_of(field_count(2));
|
||||
// Each field holds a value at one of the two rows and none at the other.
|
||||
let chunk = raw_chunk_columns(
|
||||
&schema,
|
||||
&[0, 60_000],
|
||||
vec![vec![Some(1.0), None], vec![None, Some(4.0)]],
|
||||
true,
|
||||
);
|
||||
let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
execution.push_raw(chunk).unwrap();
|
||||
let out = run(&mut execution);
|
||||
|
||||
assert_eq!(out.len(), 1);
|
||||
let chunk = &out[0];
|
||||
assert_eq!(
|
||||
chunk.series.time.len(),
|
||||
2,
|
||||
"a row no field of which holds a value was written down"
|
||||
);
|
||||
assert_eq!(field_values(chunk, 0), vec![Some(1.0), None]);
|
||||
assert_eq!(field_values(chunk, 1), vec![None, Some(4.0)]);
|
||||
assert_eq!(
|
||||
chunk
|
||||
.series
|
||||
.time
|
||||
.positions()
|
||||
.map(|(position, _)| position)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![0, 1]
|
||||
);
|
||||
}
|
||||
|
||||
/// A grid position nobody fills is not a row, and a hole does not close the series early.
|
||||
///
|
||||
/// The samples lie at the first and the last point of the grid with nothing in between, and the
|
||||
/// lookback is too short to carry either one across: two rows, two runs, and no row per position
|
||||
/// between them.
|
||||
#[test]
|
||||
fn a_position_no_field_fills_is_not_a_row() {
|
||||
let out_grid = grid(0, 60_000, 5);
|
||||
let tree = NativeTree::try_new(instant(input(), 1_000, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let schema = raw_schema();
|
||||
let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[0], &[1.0], false))
|
||||
.unwrap();
|
||||
// The hole between the two samples decides nothing, and it does not end the series.
|
||||
assert!(
|
||||
execution.next_chunk().unwrap().is_none(),
|
||||
"a hole in the grid closed the series"
|
||||
);
|
||||
assert!(
|
||||
execution.state().reserved_bytes() > 0,
|
||||
"the open series is not reserved for"
|
||||
);
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[240_000], &[5.0], true))
|
||||
.unwrap();
|
||||
let out = run(&mut execution);
|
||||
|
||||
assert_eq!(out.len(), 1);
|
||||
let chunk = &out[0];
|
||||
assert!(chunk.is_last);
|
||||
assert_eq!(
|
||||
chunk
|
||||
.series
|
||||
.time
|
||||
.positions()
|
||||
.map(|(position, _)| position)
|
||||
.collect::<Vec<_>>(),
|
||||
vec![0, 4],
|
||||
"the positions nobody fills were written down as rows"
|
||||
);
|
||||
assert_eq!(decided(&out), vec![(0, Some(1.0)), (4, Some(5.0))]);
|
||||
assert!(
|
||||
chunk.range().is_none(),
|
||||
"a chunk of two runs of positions was addressed as one range"
|
||||
);
|
||||
}
|
||||
|
||||
/// An instant selection selects: the histogram it picks keeps its start timestamp.
|
||||
///
|
||||
/// The sample is read, converted into the histogram type the builder writes back, and written out
|
||||
/// again, and that type used to have no start timestamp to write: a sample carrying
|
||||
/// `start_timestamp = Some(1500)` at 2000ms came out `None`. It matters downstream -- a reader
|
||||
/// detects a counter reset from a start timestamp that falls inside the range it reads -- and it is
|
||||
/// the difference between selecting a sample and computing a value.
|
||||
///
|
||||
/// One source sample is selected by three grid points here, which is the case a per-row conversion
|
||||
/// can get wrong twice: applied once and forgotten on the second use, or applied to a value that no
|
||||
/// longer carries the timestamp.
|
||||
#[test]
|
||||
fn a_selected_histogram_keeps_its_start_timestamp() {
|
||||
let out_grid = grid(2000, 1, 3);
|
||||
let tree = NativeTree::try_new(instant(input(), 60_000, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let schema = raw_schema();
|
||||
let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
execution
|
||||
.push_raw(raw_histogram_chunk_with_start(
|
||||
&schema,
|
||||
&[2000],
|
||||
Some(1500),
|
||||
true,
|
||||
))
|
||||
.unwrap();
|
||||
let out = run(&mut execution);
|
||||
|
||||
assert_eq!(out.len(), 1);
|
||||
assert_eq!(
|
||||
histogram_starts(&out[0]),
|
||||
vec![Some(1500), Some(1500), Some(1500)],
|
||||
"the selected sample lost the start timestamp it was read with"
|
||||
);
|
||||
}
|
||||
|
||||
/// Two series that differ only in an empty-valued label are two series, and both are answered.
|
||||
///
|
||||
/// The label set the runtime's occupancy table compares drops empty values, so a node that recorded
|
||||
/// what it handed out under one would refuse the second series. A selection keeps the labels it
|
||||
/// read, empty values included, so it keeps them apart and records nothing.
|
||||
#[test]
|
||||
fn two_series_differing_only_in_an_empty_label_are_both_answered() {
|
||||
let out_grid = grid(0, 60_000, 1);
|
||||
let tree = NativeTree::try_new(instant(input(), 300_000, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let schema = raw_schema();
|
||||
let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
// `zone=""` against an absent `zone`: one label set to everything that drops empty values.
|
||||
execution
|
||||
.push_raw(raw_chunk_of(
|
||||
&schema,
|
||||
b"a",
|
||||
host_metric(Some("")),
|
||||
&[0],
|
||||
&[1.0],
|
||||
true,
|
||||
))
|
||||
.unwrap();
|
||||
execution
|
||||
.push_raw(raw_chunk_of(
|
||||
&schema,
|
||||
b"b",
|
||||
host_metric(None),
|
||||
&[0],
|
||||
&[2.0],
|
||||
true,
|
||||
))
|
||||
.unwrap();
|
||||
let out = run(&mut execution);
|
||||
|
||||
assert_eq!(out.len(), 2, "one of the two series was not answered");
|
||||
assert_eq!(out[0].series.metric, host_metric(Some("")));
|
||||
assert_eq!(out[1].series.metric, host_metric(None));
|
||||
assert_eq!(decided(&out[..1]), vec![(0, Some(1.0))]);
|
||||
assert_eq!(decided(&out[1..]), vec![(0, Some(2.0))]);
|
||||
}
|
||||
|
||||
/// What a selection reserves is the series it has buffered, measured chunk by chunk.
|
||||
///
|
||||
/// The node holds a whole series, so its reservation grows with the chunks that series arrived in
|
||||
/// and goes back when the series is answered. Each chunk is measured once, with the same estimate a
|
||||
/// queued chunk is charged with: an allocation two chunks share is counted once per chunk, which is
|
||||
/// the over-count a figure that keeps no state across chunks accepts.
|
||||
///
|
||||
/// What this checks is the buffer alone. The workspace the gather of that buffer allocates -- the
|
||||
/// axis, the selection's two arrays and the rows the builder keeps -- is reserved for while it is
|
||||
/// live and is not part of this figure, so this test cannot see it, however many bytes it is:
|
||||
/// [`an_instant_selection_reserves_the_workspace_it_gathers`] is the one that does.
|
||||
#[test]
|
||||
fn an_instant_selection_reserves_the_series_it_buffers() {
|
||||
let out_grid = grid(0, 60_000, 4);
|
||||
let tree = NativeTree::try_new(instant(input(), 300_000, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let schema = raw_schema();
|
||||
let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
let mut previous = 0;
|
||||
for timestamps in [vec![0, 60_000], vec![120_000], vec![180_000]] {
|
||||
let values = vec![1.0; timestamps.len()];
|
||||
let chunk = raw_chunk(&schema, ×tamps, &values, false);
|
||||
let bytes = raw_bytes(&chunk);
|
||||
execution.push_raw(chunk).unwrap();
|
||||
assert!(
|
||||
execution.next_chunk().unwrap().is_none(),
|
||||
"a series was reported before its last chunk arrived"
|
||||
);
|
||||
let held = execution.state().reserved_bytes();
|
||||
assert_eq!(
|
||||
held,
|
||||
previous + bytes,
|
||||
"the buffered series is not what is reserved for: {held} bytes held after {previous} \
|
||||
and a chunk of {bytes}"
|
||||
);
|
||||
previous = held;
|
||||
}
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[240_000], &[2.0], true))
|
||||
.unwrap();
|
||||
let out = run(&mut execution);
|
||||
assert_eq!(out.len(), 1);
|
||||
assert_eq!(
|
||||
execution.state().reserved_bytes(),
|
||||
0,
|
||||
"the series that was answered is still reserved for"
|
||||
);
|
||||
}
|
||||
|
||||
/// The workspace the gather allocates is reserved for before it exists.
|
||||
///
|
||||
/// A selection decides the points of the grid it can be selected at, so its workspace is
|
||||
/// proportional to those points rather than to the samples of the series: the axis it concatenates,
|
||||
/// the source row and evaluation timestamp it reports per point, and the rows the builder keeps.
|
||||
/// The lookback here reaches from the sample to the last point, so this series is decididable at
|
||||
/// every one of the million points -- the clipped count equals the grid's -- and the two selection
|
||||
/// arrays alone are 16 MB while the built rows are more, against a series the node holds of one
|
||||
/// sample of a few hundred bytes. [`a_selection_reserves_the_points_it_decides`] is the other side
|
||||
/// of the same figure: a series whose samples cover only the start of a long grid reserves for the
|
||||
/// few points it can decide and is admitted by a pool this one is refused by.
|
||||
///
|
||||
/// The pool here holds 8 KiB, which is enough for the buffered chunk and nothing near the
|
||||
/// workspace, so the gather is refused instead of being performed. Nothing else refuses a million
|
||||
/// points, so this is the reservation and not a point limit answering: without it the gather simply
|
||||
/// runs.
|
||||
#[test]
|
||||
fn an_instant_selection_reserves_the_workspace_it_gathers() {
|
||||
let out_grid = grid(0, 1, 1_000_000);
|
||||
let tree = NativeTree::try_new(instant(input(), 1_000_000, 0, out_grid)).unwrap();
|
||||
let pool = bounded_pool(8 * 1024);
|
||||
let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
|
||||
let schema = raw_schema();
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[0], &[1.0], true))
|
||||
.unwrap();
|
||||
|
||||
let error = execution.next_chunk().unwrap_err();
|
||||
assert!(
|
||||
matches!(error, DataFusionError::ResourcesExhausted(_)),
|
||||
"the gather's workspace was taken without asking the pool: {error}"
|
||||
);
|
||||
assert!(
|
||||
!error.to_string().contains("exceed limit"),
|
||||
"a point limit answered in the pool's place: {error}"
|
||||
);
|
||||
assert_eq!(
|
||||
pool.reserved(),
|
||||
0,
|
||||
"the stopped execution still holds bytes"
|
||||
);
|
||||
assert_eq!(
|
||||
execution.state().reserved_bytes(),
|
||||
0,
|
||||
"the workspace of the refused gather is still reserved for"
|
||||
);
|
||||
assert_eq!(
|
||||
execution.state().nodes_holding_instants(),
|
||||
0,
|
||||
"the stopped execution still holds the buffered series"
|
||||
);
|
||||
// A refused gather stops the execution rather than leaving it to be resumed.
|
||||
let again = execution.next_chunk().unwrap_err();
|
||||
assert!(
|
||||
again.to_string().contains("stopped after an earlier error"),
|
||||
"{again}"
|
||||
);
|
||||
}
|
||||
|
||||
/// A selection reserves the points it can decide, not the points the grid plans.
|
||||
///
|
||||
/// The review's shape: a seven-day grid at a one-minute step is 10,081 points, and one sample at
|
||||
/// the start of the series read with a five-minute lookback is decided at five of them. The
|
||||
/// grid-scaled terms of `InstantRuntime::gather_bytes` at the planned grid -- the selection's two
|
||||
/// arrays and the doubled builder columns -- are about 504 KB, which the 64 KiB pool here is far
|
||||
/// below; the handful of points really decided need a few hundred bytes, so this gather has to be
|
||||
/// admitted. What it hands out is the rows `select_instants` reports for the same series, which is
|
||||
/// the count the reservation has to agree with.
|
||||
///
|
||||
/// The second half is the other direction: the same grid, the same pool, and a series that spans
|
||||
/// the grid with a lookback long enough to cover it, so all 10,081 points can be decided. The
|
||||
/// figure then exceeds the pool and the gather is refused, which is what keeps this from being a
|
||||
/// reservation of nothing.
|
||||
#[test]
|
||||
fn a_selection_reserves_the_points_it_decides() {
|
||||
let out_grid = grid(0, 60_000, 10_081);
|
||||
let schema = raw_schema();
|
||||
let budget = 64 << 10;
|
||||
// The two grid-scaled terms of `gather_bytes` at the planned grid: 10,081 points x 16 bytes for
|
||||
// the selection's arrays, and 10,081 points x (8 + 8 + 1) x 2 for the doubled builder columns.
|
||||
let full_grid = 10_081 * (8 + 8) + 10_081 * (8 + 8 + 1) * 2;
|
||||
assert!(
|
||||
full_grid > budget,
|
||||
"the pool is not below what the full grid would be reserved: {full_grid} bytes"
|
||||
);
|
||||
|
||||
// One sample at the start, read within five minutes: the kernel reports 0s..240s, and the
|
||||
// candidate bound it sizes its arrays with is 0s..300s.
|
||||
let tree = NativeTree::try_new(instant(input(), 300_000, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let chunk = raw_chunk(&schema, &[0], &[1.0], true);
|
||||
let (_, timestamps) = select_instants(
|
||||
&chunk.time,
|
||||
&chunk.fields,
|
||||
out_grid.spec().start_ms..=grid_end_ms(out_grid.spec()),
|
||||
out_grid.spec().step_ms,
|
||||
300_000,
|
||||
0,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
timestamps.len(),
|
||||
5,
|
||||
"the clipping kernel decides another count"
|
||||
);
|
||||
let mut execution = executor(&tree, bounded_pool(budget), &ExecutionPlanMetricsSet::new());
|
||||
execution.push_raw(chunk).unwrap();
|
||||
let out = run(&mut execution);
|
||||
assert_eq!(out.len(), 1);
|
||||
let expected: Vec<(usize, Option<f64>)> = timestamps
|
||||
.iter()
|
||||
.map(|timestamp| {
|
||||
(
|
||||
usize::try_from((timestamp - out_grid.spec().start_ms) / out_grid.spec().step_ms)
|
||||
.unwrap(),
|
||||
Some(1.0),
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(
|
||||
decided(&out),
|
||||
expected,
|
||||
"the gather handed out another set of points than the clipping decides"
|
||||
);
|
||||
|
||||
// The same series spans the grid with a lookback that covers it, so every point can be decided
|
||||
// and the workspace no longer fits the same pool.
|
||||
let tree =
|
||||
NativeTree::try_new(instant(input(), 604_800_001, 0, Arc::clone(&out_grid))).unwrap();
|
||||
let pool = bounded_pool(budget);
|
||||
let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[0, 604_800_000], &[1.0, 2.0], true))
|
||||
.unwrap();
|
||||
let error = execution.next_chunk().unwrap_err();
|
||||
assert!(
|
||||
matches!(error, DataFusionError::ResourcesExhausted(_)),
|
||||
"a series covering the grid was gathered without asking the pool: {error}"
|
||||
);
|
||||
assert_eq!(pool.reserved(), 0);
|
||||
assert_eq!(execution.state().reserved_bytes(), 0);
|
||||
}
|
||||
|
||||
/// A grid longer than any point limit is charged for, and the pool is what refuses it.
|
||||
///
|
||||
/// `gather_bytes` used to return zero for a grid this long, so that a point limit inside the
|
||||
/// kernel stayed the error reported and the workspace of a long gather was never reserved. The
|
||||
/// estimate now applies to every gather, so the pool is asked before the kernel is and its refusal
|
||||
/// is the one the caller sees — a bound in bytes, on the path that decides the points, rather than
|
||||
/// a count that says nothing about what they cost.
|
||||
#[test]
|
||||
fn an_oversized_selection_is_refused_by_the_pool() {
|
||||
// One million and two points over two samples: longer than the short circuit used to allow,
|
||||
// so the estimate has to cover it rather than return zero.
|
||||
let out_grid = grid(0, 1, 1_000_002);
|
||||
let tree = NativeTree::try_new(instant(input(), 1_000_000, 0, out_grid)).unwrap();
|
||||
let pool = bounded_pool(8 * 1024);
|
||||
let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
|
||||
let schema = raw_schema();
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[0, 1_000_001], &[1.0, 2.0], true))
|
||||
.unwrap();
|
||||
|
||||
let error = execution.next_chunk().unwrap_err();
|
||||
assert!(
|
||||
matches!(error, DataFusionError::ResourcesExhausted(_)),
|
||||
"the gather's workspace was taken without asking the pool: {error}"
|
||||
);
|
||||
assert!(
|
||||
!error.to_string().contains("exceed limit"),
|
||||
"a point limit answered in the pool's place: {error}"
|
||||
);
|
||||
// Nothing is dropped silently: the execution stops and gives its reservation back.
|
||||
assert_eq!(pool.reserved(), 0);
|
||||
assert_eq!(execution.state().reserved_bytes(), 0);
|
||||
assert_eq!(execution.state().nodes_holding_instants(), 0);
|
||||
let again = execution.next_chunk().unwrap_err();
|
||||
assert!(
|
||||
again.to_string().contains("stopped after an earlier error"),
|
||||
"{again}"
|
||||
);
|
||||
}
|
||||
|
||||
/// An input that ends inside a series is an error, not an empty answer.
|
||||
///
|
||||
/// The selection holds the series and gathers it only on the chunk that closes it, so nothing is
|
||||
/// answered for it. The node that notices is the input: the chunk it handed over did not say
|
||||
/// `is_last`, and the tree's close reports that before any collected input is merged.
|
||||
#[test]
|
||||
fn an_input_that_ends_inside_a_series_is_reported() {
|
||||
let tree = NativeTree::try_new(instant(input(), 300_000, 0, grid(0, 60_000, 2))).unwrap();
|
||||
let pool = pool();
|
||||
let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
|
||||
let schema = raw_schema();
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[0], &[1.0], false))
|
||||
.unwrap();
|
||||
assert!(execution.next_chunk().unwrap().is_none());
|
||||
|
||||
let error = execution.finish().unwrap_err();
|
||||
assert!(
|
||||
error.to_string().contains("ended before the last chunk"),
|
||||
"{error}"
|
||||
);
|
||||
assert_eq!(pool.reserved(), 0);
|
||||
assert_eq!(execution.state().reserved_bytes(), 0);
|
||||
assert_eq!(
|
||||
execution.state().nodes_holding_instants(),
|
||||
0,
|
||||
"the stopped execution still holds the buffered series"
|
||||
);
|
||||
}
|
||||
|
||||
/// A selection combines one series into one axis, so a chunk of another series is refused.
|
||||
///
|
||||
/// The buffered chunks are read as one axis, so a chunk of a second series has nowhere to go: it
|
||||
/// would be selected as though its samples were the open series'.
|
||||
#[test]
|
||||
fn a_chunk_of_another_series_is_refused() {
|
||||
let tree = NativeTree::try_new(instant(input(), 300_000, 0, grid(0, 60_000, 2))).unwrap();
|
||||
let schema = raw_schema();
|
||||
let mut execution = executor(&tree, pool(), &ExecutionPlanMetricsSet::new());
|
||||
execution
|
||||
.push_raw(raw_chunk(&schema, &[0], &[1.0], false))
|
||||
.unwrap();
|
||||
assert!(execution.next_chunk().unwrap().is_none());
|
||||
// The second chunk carries another series' key.
|
||||
execution
|
||||
.push_raw(raw_chunk_of(
|
||||
&schema,
|
||||
b"series-b",
|
||||
metric(),
|
||||
&[60_000],
|
||||
&[2.0],
|
||||
true,
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
let error = execution.next_chunk().unwrap_err();
|
||||
assert!(
|
||||
error
|
||||
.to_string()
|
||||
.contains("a chunk of another series arrived"),
|
||||
"{error}"
|
||||
);
|
||||
assert_eq!(
|
||||
execution.state().reserved_bytes(),
|
||||
0,
|
||||
"the refused series is still reserved for"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1010,6 +1010,180 @@ fn histogram_chunk(schema: &SeriesSchema, buckets: i32, ts: i64) -> RawSeriesChu
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// The single histogram sample of [`histogram_chunk`], read with `start_timestamp`.
|
||||
///
|
||||
/// A sample carries the start timestamp a scan read it with, and it is payload rather than a
|
||||
/// property of the value.
|
||||
fn histogram_chunk_from(
|
||||
schema: &SeriesSchema,
|
||||
buckets: i32,
|
||||
ts: i64,
|
||||
start_timestamp: Option<i64>,
|
||||
is_last: bool,
|
||||
) -> 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,
|
||||
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(),
|
||||
is_last,
|
||||
)
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
/// A function that hands back the sample it selected hands back that sample's payload.
|
||||
///
|
||||
/// `last_over_time` and `first_over_time` select a sample instead of computing one, so the
|
||||
/// histogram that leaves them is the histogram that arrived, start timestamp included. The window
|
||||
/// here holds two samples that differ in bucket count and in start timestamp, so the count and the
|
||||
/// timestamp together say which one was selected rather than only that some timestamp survived:
|
||||
/// the last sample for `last_over_time`, the first one for `first_over_time`. What this pins is
|
||||
/// that the selection passes the stored payload through; it does not pin a Prometheus result, whose
|
||||
/// histogram values carry no such field, and it does not claim that a native rate reads the field
|
||||
/// afterwards. A histogram computed from samples carries none, as
|
||||
/// `histogram_eval::tests::a_start_timestamp_belongs_to_a_read_histogram_and_not_to_a_computed_one`
|
||||
/// pins for the operations that build one.
|
||||
#[test]
|
||||
fn a_selected_histogram_keeps_its_start_timestamp() {
|
||||
let schema = raw_schema(1);
|
||||
let (first_start, last_start): (i64, i64) = (1_500, 2_500);
|
||||
// One bucket at 0ms and two at 5000ms, each read with its own start timestamp. Both samples
|
||||
// are inside the window of the second grid point; the first point is decided on the first
|
||||
// chunk alone, before the second sample has arrived.
|
||||
let chunks = [
|
||||
histogram_chunk_from(&schema, 1, 0, Some(first_start), false),
|
||||
histogram_chunk_from(&schema, 2, 5_000, Some(last_start), true),
|
||||
];
|
||||
// The histogram of every row the function hands out, read off the output's own views: the
|
||||
// bucket count tells the selected sample apart from any other, the start timestamp is the
|
||||
// payload this test is about.
|
||||
let selected = |function: RangeFunction| -> Vec<(usize, f64, Option<i64>)> {
|
||||
let mut evaluator = RangeEvaluator::try_new(
|
||||
&schema,
|
||||
grid(0, 10_000, 2),
|
||||
RangeWindow {
|
||||
range_ms: 20_000,
|
||||
offset_ms: 0,
|
||||
anchor_ms: None,
|
||||
},
|
||||
function,
|
||||
)
|
||||
.unwrap();
|
||||
let out = collect_chunks(&mut evaluator, &chunks);
|
||||
assert!(!out.is_empty(), "{function:?} answered no chunk");
|
||||
let mut picked = Vec::new();
|
||||
for chunk in &out {
|
||||
let mut cursor = chunk.series.fields.as_slice()[0].cursor();
|
||||
for (position, row) in chunk.series.time.positions() {
|
||||
match cursor.at(row).expect("the axis and the fields are aligned") {
|
||||
SampleRef::Histogram(view) => {
|
||||
picked.push((position, view.count, view.start_timestamp));
|
||||
}
|
||||
SampleRef::Float(_) => panic!("{function:?} answered a float for a histogram"),
|
||||
SampleRef::Missing => {
|
||||
panic!("{function:?} handed out a row of its own axis without a sample")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
picked
|
||||
};
|
||||
|
||||
assert_eq!(
|
||||
selected(RangeFunction::LastOverTime),
|
||||
[(0, 1.0, Some(first_start)), (1, 2.0, Some(last_start))],
|
||||
"last_over_time did not hand back the sample it selected"
|
||||
);
|
||||
assert_eq!(
|
||||
selected(RangeFunction::FirstOverTime),
|
||||
[(0, 1.0, Some(first_start)), (1, 1.0, Some(first_start))],
|
||||
"first_over_time did not hand back the sample it selected"
|
||||
);
|
||||
}
|
||||
|
||||
/// A histogram computed from a window carries no start timestamp, even when the window holds one.
|
||||
///
|
||||
/// `sum_over_time` computes a histogram rather than selecting one, so the timestamp the sample was
|
||||
/// read with is not part of the result -- the rule the histogram operations themselves follow, as
|
||||
/// `histogram_eval::tests::a_start_timestamp_belongs_to_a_read_histogram_and_not_to_a_computed_one`
|
||||
/// pins. The fold clears the field while adding, so a window holding two histograms already loses
|
||||
/// it; a window holding one must not depend on that, or whether a result carries the field would
|
||||
/// say how many samples the window held rather than what the function did.
|
||||
#[test]
|
||||
fn a_computed_histogram_drops_its_start_timestamp() {
|
||||
let schema = raw_schema(1);
|
||||
let start: i64 = 1_500;
|
||||
// One histogram sample inside the window of both grid points, so the fold adds nothing to it
|
||||
// and the single input is the whole result.
|
||||
let chunks = [histogram_chunk_from(&schema, 1, 0, Some(start), true)];
|
||||
let mut evaluator = RangeEvaluator::try_new(
|
||||
&schema,
|
||||
grid(0, 10_000, 2),
|
||||
RangeWindow {
|
||||
range_ms: 20_000,
|
||||
offset_ms: 0,
|
||||
anchor_ms: None,
|
||||
},
|
||||
RangeFunction::SumOverTime,
|
||||
)
|
||||
.unwrap();
|
||||
let out = collect_chunks(&mut evaluator, &chunks);
|
||||
let mut computed = Vec::new();
|
||||
for chunk in &out {
|
||||
let mut cursor = chunk.series.fields.as_slice()[0].cursor();
|
||||
for (position, row) in chunk.series.time.positions() {
|
||||
match cursor.at(row).expect("the axis and the fields are aligned") {
|
||||
SampleRef::Histogram(view) => {
|
||||
computed.push((position, view.count, view.start_timestamp));
|
||||
}
|
||||
SampleRef::Float(_) => panic!("sum_over_time answered a float for a histogram"),
|
||||
SampleRef::Missing => {
|
||||
panic!("sum_over_time handed out a row of its own axis without a sample")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// The sum of one histogram is that histogram, so the count says the sample was summed rather
|
||||
// than passed over; only the start timestamp has to be gone.
|
||||
assert_eq!(
|
||||
computed,
|
||||
[(0, 1.0, None), (1, 1.0, None)],
|
||||
"sum_over_time kept the start timestamp of the histogram it summed"
|
||||
);
|
||||
}
|
||||
|
||||
/// A window retaining a native histogram is charged for the payload it owns, not only the slot.
|
||||
///
|
||||
/// The histogram sample below stays ahead of the second grid point's window, so the evaluator keeps
|
||||
|
||||
@@ -269,6 +269,67 @@ fn limit_operators_keep_histogram_samples() {
|
||||
assert_eq!(kinds(&kept[0], 0), [(1, "histogram")]);
|
||||
}
|
||||
|
||||
/// A selection hands back the sample it selected, start timestamp included.
|
||||
///
|
||||
/// `limitk` and `limit_ratio` choose samples rather than compute them, so a histogram that arrives
|
||||
/// with a start timestamp leaves with it. The count and the sum are compared too, so a computed
|
||||
/// value that happened to carry a timestamp would not pass for the sample selected. What this pins
|
||||
/// is that the selection passes the stored payload through, not a Prometheus result and not a
|
||||
/// downstream rate that reads the field. A histogram computed from samples carries none, as
|
||||
/// `histogram_eval::tests::a_start_timestamp_belongs_to_a_read_histogram_and_not_to_a_computed_one`
|
||||
/// pins for the operations that build one.
|
||||
#[test]
|
||||
fn limit_operators_keep_the_start_timestamp_of_a_selected_sample() {
|
||||
let histogram =
|
||||
OwnedHistogram::parse("0/0/0/3ff0000000000000/3ff0000000000000/0/1_1/3ff0000000000000///");
|
||||
let start = 1_500;
|
||||
// The sample a scan read, which is what a selection hands on.
|
||||
let stamped = || {
|
||||
let mut read = FloatHistogram::try_from_view(&histogram.view()).unwrap();
|
||||
read.start_timestamp = Some(start);
|
||||
GridSample::Histogram(Box::new(read))
|
||||
};
|
||||
// What a limit operator keeps out of one series: the count and sum of the histogram it kept,
|
||||
// so the sample selected is named, and the start timestamp it kept with it.
|
||||
let kept = |series: &Series| -> Vec<(usize, f64, f64, Option<i64>)> {
|
||||
let samples = &series.fields.as_slice()[0];
|
||||
series
|
||||
.time
|
||||
.positions()
|
||||
.filter_map(|(point, row)| match samples.sample(row) {
|
||||
SampleRef::Histogram(view) => {
|
||||
Some((point, view.count, view.sum, view.start_timestamp))
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
let source = histogram.view();
|
||||
let expected = |point: usize| (point, source.count, source.sum, Some(start));
|
||||
|
||||
let schema = schema(3, 1);
|
||||
let mut builder = GridSeriesBuilder::try_new(Arc::new(schema.clone())).unwrap();
|
||||
builder.push_samples(0, &[Some(stamped())]).unwrap();
|
||||
builder.push_samples(2, &[Some(stamped())]).unwrap();
|
||||
let series = builder.finish(metric("__name__=a")).unwrap().unwrap();
|
||||
|
||||
let by_ratio = limit_ratio(1.0, &schema, std::slice::from_ref(&series)).unwrap();
|
||||
assert_eq!(by_ratio.len(), 1);
|
||||
assert_eq!(
|
||||
kept(&by_ratio[0]),
|
||||
[expected(0), expected(2)],
|
||||
"limit_ratio did not hand back the samples it selected"
|
||||
);
|
||||
|
||||
let by_count = limitk(1.0, &Grouping::By(Vec::new()), &schema, &[series]).unwrap();
|
||||
assert_eq!(by_count.len(), 1);
|
||||
assert_eq!(
|
||||
kept(&by_count[0]),
|
||||
[expected(0), expected(2)],
|
||||
"limitk did not hand back the samples it selected"
|
||||
);
|
||||
}
|
||||
|
||||
fn grouping(kind: &str, labels: &str) -> Grouping {
|
||||
match kind {
|
||||
"without" => Grouping::Without(names(labels)),
|
||||
|
||||
@@ -26,11 +26,11 @@ use super::{
|
||||
BinaryMatching, BinaryOp, Cardinality, FillValues, MatchingLabels, SetOp, binary_vectors,
|
||||
set_vectors,
|
||||
};
|
||||
use crate::grid_series::GridSeriesBuilder;
|
||||
use crate::grid_series::reference::{
|
||||
Dense, OwnedHistogram, Points, assert_histogram_close, assert_result_eq, golden_cases,
|
||||
mixed_series, piece_ranges, random_dense, run_ranges, sparse_series,
|
||||
};
|
||||
use crate::grid_series::{GridSample, GridSeriesBuilder};
|
||||
use crate::histogram_eval::float_histogram::FloatHistogram;
|
||||
|
||||
/// Labels with `__name__` as an ordinary entry, the way Prometheus holds them.
|
||||
@@ -1487,3 +1487,248 @@ fn set_operators_of_sparse_operands_keep_their_own_positions() {
|
||||
expected[9] = vec![Some(90.0)];
|
||||
assert_result_eq(Some(&result[0]), &expected, &lhs.metric);
|
||||
}
|
||||
|
||||
/// A comparison filters, so the histogram it keeps is the left operand's, payload included.
|
||||
///
|
||||
/// A non-`bool` `==` or `!=` between two histograms keeps the left sample where the comparison
|
||||
/// holds, so the histogram that leaves is the one that arrived on the left, start timestamp
|
||||
/// included, and never the right operand's timestamp. `==` holds only for equal values, so that
|
||||
/// pair differs in reset hint and start timestamp alone; `!=` holds for differing values, so its
|
||||
/// pair differs in count and sum as well. What this pins is that a comparison passes the stored
|
||||
/// payload through, not a Prometheus result. A `bool` comparison answers in floats, so nothing it
|
||||
/// hands out is a histogram and no start timestamp leaves it.
|
||||
#[test]
|
||||
fn a_histogram_comparison_filter_keeps_the_payload_of_the_operand_it_keeps() {
|
||||
let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 1_000, 3).unwrap()).unwrap());
|
||||
let schema = Arc::new(schema(1, &grid));
|
||||
// The start timestamp each operand was read with: written into its samples by `operand` below
|
||||
// and repeated in every expectation, never read back through the operators under test.
|
||||
const LEFT_START: i64 = 1_500;
|
||||
const RIGHT_START: i64 = 2_500;
|
||||
// count 4, sum 6, one span of two buckets holding 1 and 3, reset hint 2 (`NotCounterReset`).
|
||||
let left = "0/0/0/4010000000000000/4018000000000000/2/0_2/3ff0000000000000,4008000000000000///";
|
||||
// The same value with reset hint 1 (`CounterReset`): `==` ignores the hint, so the pair is kept
|
||||
// and only the hint and the start timestamp say which side survived.
|
||||
let equal =
|
||||
"0/0/0/4010000000000000/4018000000000000/1/0_2/3ff0000000000000,4008000000000000///";
|
||||
// count 5, sum 8: a value that differs, which is what `!=` keeps.
|
||||
let other =
|
||||
"0/0/0/4014000000000000/4020000000000000/2/0_2/4000000000000000,4008000000000000///";
|
||||
let operand = |name: &str, job: &str, text: &str, start: i64| -> Series {
|
||||
let labels = Labels::from([
|
||||
("__name__".to_string(), name.to_string()),
|
||||
("job".to_string(), job.to_string()),
|
||||
]);
|
||||
let histogram = OwnedHistogram::parse(text);
|
||||
let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap();
|
||||
for point in [0, 2] {
|
||||
let mut read = FloatHistogram::try_from_view(&histogram.view()).unwrap();
|
||||
read.start_timestamp = Some(start);
|
||||
builder
|
||||
.push_samples(point, &[Some(GridSample::Histogram(Box::new(read)))])
|
||||
.unwrap();
|
||||
}
|
||||
builder.finish(metric(&labels)).unwrap().unwrap()
|
||||
};
|
||||
// Two pairs, one match group each: `job=eq` holds histograms `==` keeps, `job=ne` ones `!=`
|
||||
// keeps, so each comparison passes exactly one pair and its result is named by that pair.
|
||||
let lhs = [
|
||||
operand("eq", "eq", left, LEFT_START),
|
||||
operand("ne", "ne", left, LEFT_START),
|
||||
];
|
||||
let rhs = [
|
||||
operand("eq_right", "eq", equal, RIGHT_START),
|
||||
operand("ne_right", "ne", other, RIGHT_START),
|
||||
];
|
||||
// Every histogram a comparison handed out, by the series and point it sits at, with the reset
|
||||
// hint as the number the fixtures wrote.
|
||||
let kept = |series: &[Series]| -> Vec<(String, usize, f64, f64, Option<i64>, i32)> {
|
||||
let mut kept: Vec<(String, usize, f64, f64, Option<i64>, i32)> = Vec::new();
|
||||
for series in series {
|
||||
let name = series.metric.name().unwrap_or_default().to_string();
|
||||
let mut cursor = series.fields.as_slice()[0].cursor();
|
||||
for (point, row) in series.time.positions() {
|
||||
match cursor.at(row).expect("the axis and the fields are aligned") {
|
||||
SampleRef::Histogram(view) => kept.push((
|
||||
name.clone(),
|
||||
point,
|
||||
view.count,
|
||||
view.sum,
|
||||
view.start_timestamp,
|
||||
i32::from(view.reset_hint),
|
||||
)),
|
||||
_ => panic!("a comparison handed out a sample that is not a histogram"),
|
||||
}
|
||||
}
|
||||
}
|
||||
kept.sort_by(|a, b| (a.0.as_str(), a.1).cmp(&(b.0.as_str(), b.1)));
|
||||
kept
|
||||
};
|
||||
let matching = |op: BinaryOp| BinaryMatching {
|
||||
op,
|
||||
return_bool: false,
|
||||
labels: MatchingLabels::Ignoring(Vec::new()),
|
||||
cardinality: Cardinality::OneToOne,
|
||||
fill: FillValues::default(),
|
||||
};
|
||||
|
||||
let eq = binary_vectors(&matching(BinaryOp::Eq), &schema, &lhs, &schema, &rhs).unwrap();
|
||||
assert_eq!(
|
||||
kept(&eq),
|
||||
[
|
||||
("eq".to_string(), 0, 4.0, 6.0, Some(LEFT_START), 2),
|
||||
("eq".to_string(), 2, 4.0, 6.0, Some(LEFT_START), 2),
|
||||
],
|
||||
"`==` did not hand back the left histogram it kept: {eq:?}"
|
||||
);
|
||||
|
||||
let ne = binary_vectors(&matching(BinaryOp::Ne), &schema, &lhs, &schema, &rhs).unwrap();
|
||||
assert_eq!(
|
||||
kept(&ne),
|
||||
[
|
||||
("ne".to_string(), 0, 4.0, 6.0, Some(LEFT_START), 2),
|
||||
("ne".to_string(), 2, 4.0, 6.0, Some(LEFT_START), 2),
|
||||
],
|
||||
"`!=` did not hand back the left histogram it kept: {ne:?}"
|
||||
);
|
||||
|
||||
// A `bool` comparison answers in floats, so nothing it hands out is a histogram and no start
|
||||
// timestamp leaves it: `job=eq` holds equal histograms, so it answers 1, and `job=ne` differs
|
||||
// in count and sum, so it answers 0.
|
||||
let bool_matching = BinaryMatching {
|
||||
return_bool: true,
|
||||
..matching(BinaryOp::Eq)
|
||||
};
|
||||
let compared = binary_vectors(&bool_matching, &schema, &lhs, &schema, &rhs).unwrap();
|
||||
let answers = |series: &[Series]| -> Vec<(String, usize, f64)> {
|
||||
let mut answers: Vec<(String, usize, f64)> = Vec::new();
|
||||
for series in series {
|
||||
let job = series.metric.get("job").unwrap_or_default().to_string();
|
||||
let mut cursor = series.fields.as_slice()[0].cursor();
|
||||
for (point, row) in series.time.positions() {
|
||||
match cursor.at(row).expect("the axis and the fields are aligned") {
|
||||
SampleRef::Float(value) => answers.push((job.clone(), point, value)),
|
||||
_ => panic!("a `bool` comparison handed out a sample that is not a float"),
|
||||
}
|
||||
}
|
||||
}
|
||||
answers.sort_by(|a, b| (a.0.as_str(), a.1).cmp(&(b.0.as_str(), b.1)));
|
||||
answers
|
||||
};
|
||||
assert_eq!(
|
||||
answers(&compared),
|
||||
[
|
||||
("eq".to_string(), 0, 1.0),
|
||||
("eq".to_string(), 2, 1.0),
|
||||
("ne".to_string(), 0, 0.0),
|
||||
("ne".to_string(), 2, 0.0),
|
||||
],
|
||||
"a `bool` comparison of histograms did not answer in floats: {compared:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// A set operation hands back the sample it kept, start timestamp included.
|
||||
///
|
||||
/// `and`, `unless` and `or` choose samples rather than compute them, so each histogram that leaves
|
||||
/// them is the histogram that arrived, payload included. The two operands were read with different
|
||||
/// start timestamps and hold different counts and sums, so the count, the sum and the timestamp
|
||||
/// together name the side and the point a kept sample came from. The one exception is `or` at a
|
||||
/// point where the left side holds no sample: the sample kept is the right side's, and it carries
|
||||
/// the right side's own start timestamp. What this pins is that the operators pass the stored
|
||||
/// payload through, not a Prometheus result.
|
||||
#[test]
|
||||
fn a_set_operation_keeps_the_payload_of_the_sample_it_keeps() {
|
||||
let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(0, 1_000, 4).unwrap()).unwrap());
|
||||
let schema = Arc::new(schema(1, &grid));
|
||||
// The start timestamp each operand was read with: written into its samples by `operand` below
|
||||
// and repeated in every expectation, never read back through the operators under test.
|
||||
const LHS_START: i64 = 1_500;
|
||||
const RHS_START: i64 = 2_500;
|
||||
// count 4, sum 6, one span of two buckets holding 1 and 3.
|
||||
let lhs_text =
|
||||
"0/0/0/4010000000000000/4018000000000000/2/0_2/3ff0000000000000,4008000000000000///";
|
||||
// count 5, sum 8, so the count and the sum name the side a kept sample came from as well.
|
||||
let rhs_text =
|
||||
"0/0/0/4014000000000000/4020000000000000/2/0_2/4000000000000000,4008000000000000///";
|
||||
let operand = |name: &str, text: &str, start: i64, points: &[usize]| -> Series {
|
||||
let labels = Labels::from([
|
||||
("__name__".to_string(), name.to_string()),
|
||||
("job".to_string(), "x".to_string()),
|
||||
]);
|
||||
let histogram = OwnedHistogram::parse(text);
|
||||
let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap();
|
||||
for &point in points {
|
||||
let mut read = FloatHistogram::try_from_view(&histogram.view()).unwrap();
|
||||
read.start_timestamp = Some(start);
|
||||
builder
|
||||
.push_samples(point, &[Some(GridSample::Histogram(Box::new(read)))])
|
||||
.unwrap();
|
||||
}
|
||||
builder.finish(metric(&labels)).unwrap().unwrap()
|
||||
};
|
||||
// The left side holds 0 and 1, the right side 0 and 2, so each operator keeps a different set:
|
||||
// the shared point for `and`, the left-only point for `unless`, and the union for `or`.
|
||||
let lhs = operand("l", lhs_text, LHS_START, &[0, 1]);
|
||||
let rhs = operand("r", rhs_text, RHS_START, &[0, 2]);
|
||||
// Every histogram an operator handed out, by the series and point it sits at.
|
||||
let kept = |series: &[Series]| -> Vec<(String, usize, f64, f64, Option<i64>)> {
|
||||
let mut kept: Vec<(String, usize, f64, f64, Option<i64>)> = Vec::new();
|
||||
for series in series {
|
||||
let name = series.metric.name().unwrap_or_default().to_string();
|
||||
let mut cursor = series.fields.as_slice()[0].cursor();
|
||||
for (point, row) in series.time.positions() {
|
||||
match cursor.at(row).expect("the axis and the fields are aligned") {
|
||||
SampleRef::Histogram(view) => kept.push((
|
||||
name.clone(),
|
||||
point,
|
||||
view.count,
|
||||
view.sum,
|
||||
view.start_timestamp,
|
||||
)),
|
||||
_ => panic!("a set operator handed out a sample that is not a histogram"),
|
||||
}
|
||||
}
|
||||
}
|
||||
kept.sort_by(|a, b| (a.0.as_str(), a.1).cmp(&(b.0.as_str(), b.1)));
|
||||
kept
|
||||
};
|
||||
let labels = MatchingLabels::Ignoring(Vec::new());
|
||||
let eval = |op: SetOp| {
|
||||
set_vectors(
|
||||
op,
|
||||
&labels,
|
||||
&schema,
|
||||
std::slice::from_ref(&lhs),
|
||||
&schema,
|
||||
std::slice::from_ref(&rhs),
|
||||
)
|
||||
.unwrap()
|
||||
};
|
||||
|
||||
let and = eval(SetOp::And);
|
||||
assert_eq!(
|
||||
kept(&and),
|
||||
[("l".to_string(), 0, 4.0, 6.0, Some(LHS_START))],
|
||||
"`and` did not hand back the left sample it kept: {and:?}"
|
||||
);
|
||||
|
||||
let unless = eval(SetOp::Unless);
|
||||
assert_eq!(
|
||||
kept(&unless),
|
||||
[("l".to_string(), 1, 4.0, 6.0, Some(LHS_START))],
|
||||
"`unless` did not hand back the left sample it kept: {unless:?}"
|
||||
);
|
||||
|
||||
// The right side comes back at point 2, where the left side holds no sample, and carries the
|
||||
// right side's own start timestamp -- the one case where the payload kept is not the left one.
|
||||
let or = eval(SetOp::Or);
|
||||
assert_eq!(
|
||||
kept(&or),
|
||||
[
|
||||
("l".to_string(), 0, 4.0, 6.0, Some(LHS_START)),
|
||||
("l".to_string(), 1, 4.0, 6.0, Some(LHS_START)),
|
||||
("r".to_string(), 2, 5.0, 8.0, Some(RHS_START)),
|
||||
],
|
||||
"`or` did not hand back each side's own sample: {or:?}"
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user