rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-14 14:56:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'narthex-binary-runtime' into feat/narthex-fused-tree

Browse Source 
Bring in #9813's later budget fixes. The public `NativeExecutor` API this
branch uses (`try_new`, `push_raw`, `next_chunk`, `finish`) is unchanged, so
nothing here needed adjusting: the merge is clean and the tree still builds.

Review asked for this: the branch was written against `859b549e70f8` and the
runtime has moved on since.

Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
2026-10-21 14:12:15 +08:00
parent 9b015bc57d 2f3a34dc6f
commit d44445fb29
3 changed files with 775 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/src/native/execute/tests.rs
+154
View File
@@ -4939,3 +4939,157 @@ fn a_binary_counts_each_closed_result_once() {
assert_eq!(count(&set, &root, NodeMetric::INPUT_BATCHES), 2);
assert_eq!(count(&set, &root, NodeMetric::OUTPUT_BATCHES), 1);
}
/// A legitimate empty result over a wide operand finishes in a 4 MiB pool and holds nothing after.
///
/// A match group of one member collects no points in the duplicate scan, so charging a point per
/// row of an operand reserves memory the kernel never allocates.
#[test]
fn an_empty_result_of_a_wide_operand_fits_its_pool() {
let points = 10_000;
let (tree, grid, left, right) = binary_tree(BinaryOp::Add, points as u64);
let schema = evaluated_schema(grid);
let pool = bounded_pool(4 << 20);
let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
let series = |job: &str| pod_metric("requests", job, "p");
execution
.push_evaluated_at(
&left,
evaluated_chunk_of(&schema, series("left"), 0, vec![Some(1.0); points], true),
)
.unwrap();
assert!(execution.next_chunk().unwrap().is_none());
for index in 0..32 {
execution
.push_evaluated_at(
&right,
evaluated_chunk_of(
&schema,
series(&format!("r{index:02}")),
0,
vec![Some(2.0); points],
true,
),
)
.unwrap();
assert!(execution.next_chunk().unwrap().is_none());
}
let finished = execution.finish();
assert!(
finished.is_ok(),
"an empty result over {points} points does not fit a 4 MiB pool: {:?}",
finished.as_ref().err()
);
assert!(
finished.unwrap().is_empty(),
"nothing matches, so nothing is answered"
);
assert_eq!(pool.reserved(), 0);
assert_eq!(execution.state().reserved_bytes(), 0);
}
/// A legitimate wide query finishes in a 32 MiB pool, keeping every field's sum and nothing else.
///
/// A thousand points pair all sixty-four fields of both operands, so the workspace is the rows
/// filed and the tables beside them, not a figure far past either.
#[test]
fn a_dense_wide_result_fits_a_32_mib_pool() {
let (points, width) = (1_000usize, 64usize);
let grid = grid(0, 60_000, points as u64);
let (left_node, right_node) = (
evaluated_input_of(width, Arc::clone(&grid)),
evaluated_input_of(width, Arc::clone(&grid)),
);
let schema = left_node
.output
.samples()
.expect("a source of evaluated samples")
.clone();
let root = binary_of(left_node, right_node, binary_matching(BinaryOp::Add));
let tree = NativeTree::try_new(root).unwrap();
let pool = bounded_pool(32 << 20);
let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
let columns = |value: f64| -> Vec<Vec<Option<f64>>> {
(0..width).map(|_| vec![Some(value); points]).collect()
};
let (left, right) = (NodePath::root().left(), NodePath::root().right());
execution
.push_evaluated_at(
&left,
evaluated_columns(&schema, metric(), 0, columns(1.0), true),
)
.unwrap();
execution
.push_evaluated_at(
&right,
evaluated_columns(&schema, metric(), 0, columns(2.0), true),
)
.unwrap();
assert!(execution.next_chunk().unwrap().is_none());
let finished = execution.finish();
assert!(
finished.is_ok(),
"{width} fields over {points} points do not fit a 32 MiB pool: {:?}",
finished.as_ref().err()
);
let out = finished.unwrap();
assert_eq!(out.len(), 1, "one matched label set answers one chunk");
let chunk = &out[0];
assert!(chunk.is_last, "a result series is handed over closed");
assert_eq!(chunk.series.time.len(), points);
assert_eq!(series_start(chunk), 0);
assert_eq!(chunk.series.fields.len(), width);
for field in 0..width {
assert_eq!(
field_values(chunk, field),
vec![Some(3.0); points],
"field {field}"
);
}
assert_eq!(pool.reserved(), 0);
assert_eq!(execution.state().reserved_bytes(), 0);
}
/// An all-missing match group of two right series finishes in a 4 MiB pool and answers nothing.
///
/// No member carries a sample, so the duplicate scan needs no point buffer.
#[test]
fn an_all_missing_match_group_fits_its_pool() {
let points = 100_000;
let (tree, grid, left, right) = binary_tree(BinaryOp::Add, points as u64);
let schema = evaluated_schema(grid);
let pool = bounded_pool(4 << 20);
let mut execution = executor(&tree, Arc::clone(&pool), &ExecutionPlanMetricsSet::new());
execution
.push_evaluated_at(
&left,
evaluated_chunk_of(&schema, metric(), 0, vec![Some(1.0); points], true),
)
.unwrap();
assert!(execution.next_chunk().unwrap().is_none());
// Both right chunks close the one label set, so its match group has two members.
for _ in 0..2 {
execution
.push_evaluated_at(
&right,
evaluated_chunk_of(&schema, metric(), 0, vec![None; points], true),
)
.unwrap();
assert!(execution.next_chunk().unwrap().is_none());
}
let finished = execution.finish();
assert!(
finished.is_ok(),
"an all-missing match group over {points} points does not fit a 4 MiB pool: {:?}",
finished.as_ref().err()
);
assert!(
finished.unwrap().is_empty(),
"no point holds a value on both sides, so nothing is answered"
);
assert_eq!(pool.reserved(), 0);
assert_eq!(execution.state().reserved_bytes(), 0);
}
src/promql/src/vector_match.rs
+127 -34
View File
@@ -217,7 +217,7 @@ fn binary_vectors_inner(
earliest_error(errors)?;
// The merge holds every field's results at once: the triples it files, the table it files them
// in and the builder writing the series out are the last workspace of the operation.
if let Some(budget) = budget {
if let Some(budget) = budget.as_deref_mut() {
let sets = fields
.iter()
.map(|results| results.series.len())
@@ -231,9 +231,12 @@ fn binary_vectors_inner(
);
budget.step(merge)?;
}
merge_fields(fields, output)
merge_fields(fields, output, budget)
}
/// The capacity a fresh `Vec` holds: a first push allocates four entries, and a full one doubles.
const VEC_MIN_CAPACITY: usize = 4;
/// One label of a match key, charged twice because `group_key` grows its `Vec` by doubling.
const KEY_LABEL_BYTES: usize = 2 * std::mem::size_of::<(Arc<str>, Arc<str>)>();
@@ -377,7 +380,8 @@ impl<'a> Budget<'a> {
/// What one field's tables, flags and messages cost before it pairs anything.
#[derive(Default)]
struct FieldMetadata {
/// The presence flags, the one side's match groups and the duplicate scan's points.
/// The presence flags and the one side's match groups: what the field holds beside the pairing
/// it walks, and the figure a step of that walk is charged on top of.
tables: usize,
/// The allowance for the messages the field can report, which it reserves for its sort before
/// it knows which of them it appends; only the ones it writes are kept.
@@ -390,11 +394,12 @@ struct FieldMetadata {
///
/// The presence of each side and of the points both are present at are three flags per grid point;
/// the one side's groups are a key, the values `on(...)` matching copies and a member list per
/// series with the table holding them and the references indexing them; the duplicate scan
/// collects that side's present points; one series of the many side is looked up by a key of its
/// own; and a field reports at most three messages, each a fixed frame around one rendering.
/// Nothing here counts the pairing the field walks: which series match, where their axes overlap
/// and which rows the operator keeps is only known as the walk reaches them.
/// series with the table holding them and the references indexing them; one series of the many side
/// is looked up by a key of its own; and a field reports at most three messages, each a fixed frame
/// around one rendering. Nothing here counts the pairing the field walks: which series match, where
/// their axes overlap and which rows the operator keeps is only known as the walk reaches them.
/// The duplicate scan's list of present points is not part of these tables either: a group of one
/// member collects nothing, and a wider group's list is charged by [`overlap`] as it is pushed to.
fn field_metadata_bytes(
matching: &BinaryMatching,
include: &[Arc<str>],
@@ -410,9 +415,7 @@ fn field_metadata_bytes(
.saturating_add(2 * std::mem::size_of::<usize>())
.saturating_add(GROUP_SLOT_BYTES)
.saturating_add(2 * std::mem::size_of::<&GroupKey>())
.saturating_add(2 * std::mem::size_of::<(&GroupKey, usize)>())
// The duplicate scan's point list, one point per row the side carries at most.
.saturating_add(2 * std::mem::size_of::<usize>() * series.time.len());
.saturating_add(2 * std::mem::size_of::<(&GroupKey, usize)>());
largest_key = largest_key.max(result_key_bytes(&series.metric));
}
let mut lookup = 0usize;
@@ -539,11 +542,12 @@ fn result_set_bytes(metric: &MetricName, one: &MetricName, include: &[Arc<str>])
///
/// The merge files every kept triple under its label set and hands one series per label set to a
/// builder, so the triples, the table, the row it fills, the output vector and the builder's own
/// buffers are alive together: one index per retained row, one value and validity bit per field and
/// row, the span a run may start with, a row of the histogram table per histogram and the heap its
/// buckets are copied to. Every term counts a value a field actually kept or a label set it filed,
/// so a pair the operator filtered costs its label set -- the table holds one without a value --
/// and no row at all.
/// buffers are alive together: one triple of a label set's samples, the span a run may start with, a
/// row of the histogram table per histogram and the heap its buckets are copied to. Every term counts
/// a value a field actually kept or a label set it filed, so a pair the operator filtered costs its
/// label set -- the table holds one without a value -- and no row at all. What grows with the output
/// rows is charged by [`merge_fields`] as they are pushed: the value and validity column of every
/// field, and the index column of a field whose first histogram backfills the rows before it.
fn merge_bytes(
kept: usize,
sets: usize,
@@ -551,22 +555,20 @@ fn merge_bytes(
histograms: usize,
histogram_heap: usize,
) -> usize {
// One kept triple, charged twice for the capacity its vector ends at, and one row of the
// builder's columns: its index, its value and validity bit per field, the span it may start
// and its row of the histogram table's index column.
let values = 2 * fields.saturating_mul(std::mem::size_of::<f64>() + 1);
// One kept triple, charged twice for the capacity its vector ends at, one slot of the row the
// builder reads and the span it may start.
let row = (2 * std::mem::size_of::<(usize, usize, GridSample)>())
.saturating_add(2 * std::mem::size_of::<usize>())
.saturating_add(2 * std::mem::size_of::<GridSpan>())
.saturating_add(values)
.saturating_add(std::mem::size_of::<Option<u32>>());
.saturating_add(2 * std::mem::size_of::<GridSpan>());
kept.saturating_mul(row)
// One label set's entry in the merge table and the series it is written out as.
.saturating_add(
sets.saturating_mul(MERGE_SLOT_BYTES.saturating_add(2 * std::mem::size_of::<Series>())),
)
// The row the builder reads its pushes from, one slot per field.
// The row the builder reads its pushes from and the per-field ledger of the capacity each
// histogram index column has reached, one slot per field.
.saturating_add(2 * fields.saturating_mul(std::mem::size_of::<Option<GridSample>>()))
.saturating_add(fields.saturating_mul(std::mem::size_of::<usize>()))
// The histogram table of every field that carries one: the builders of its columns, the
// scalars and list offsets of its rows, and the buckets the batch copies beside the value
// the results still hold.
@@ -575,6 +577,16 @@ fn merge_bytes(
.saturating_add(2 * histogram_heap)
}
/// The bytes one output row of the merge adds to the builder's value columns: the value and the
/// validity bit of every field of the schema, charged twice for the capacity each column grows to.
///
/// A row holds a value only in the fields the results filed at its point, and the builder writes a
/// null into every other column, so this figure counts the schema's fields rather than the row's
/// values: fields whose samples sit at different positions still write a column each.
fn merge_row_bytes(fields: usize) -> usize {
2 * fields.saturating_mul(std::mem::size_of::<f64>() + 1)
}
/// The bytes one match key holds: its label pairs and, for `on(...)`, a copy of each value.
///
/// `group_key` copies an `on(...)` value with `Arc::from` and shares the `Arc` of an
@@ -735,7 +747,16 @@ fn binary_field(
let one_groups = group(one, &matching.labels);
let mut one_duplicate: Option<(usize, &GroupKey)> = None;
for (key, members) in &one_groups {
if let Some(point) = overlap(one, members, |point| both_present[point])
// The group's list of present points is charged as this scan grows it, beside the tables
// the field keeps from here on.
let duplicate = overlap(
one,
members,
|point| both_present[point],
metadata.tables,
budget.as_deref_mut(),
)?;
if let Some(point) = duplicate
&& one_duplicate.is_none_or(|(first, _)| point < first)
{
one_duplicate = Some((point, key));
@@ -1024,7 +1045,8 @@ pub fn set_vectors(
fields.push(results);
}
earliest_error(errors)?;
merge_fields(fields, output)
// A set operator accounts for no workspace, so its merge is charged nothing either.
merge_fields(fields, output, None)
}
fn set_field(
@@ -1084,8 +1106,30 @@ fn set_field(
Ok(results)
}
/// The capacity a field's histogram index column reaches when `pushed` output rows have been pushed
/// and the row being pushed holds `histogram`; `None` when the column does not grow.
///
/// [`GridSeriesBuilder::push_samples`] creates the column at the first histogram with one entry per
/// row before it and then pushes, so that push allocates `max(2 * pushed, 4)`; every later row adds
/// an entry, and a full column doubles.
fn index_column_growth(pushed: usize, capacity: usize, histogram: bool) -> Option<usize> {
if capacity == 0 {
return histogram.then(|| (2 * pushed).max(VEC_MIN_CAPACITY));
}
(pushed == capacity).then(|| 2 * capacity)
}
/// Joins the per-field results into series with all fields, by label set.
fn merge_fields(fields: Vec<Results>, schema: Arc<SeriesSchema>) -> Result<Vec<Series>> {
///
/// The builder's columns grow with the output rows, which only exist once the triples are being
/// grouped, so they are reserved per row -- one row's worth before the builder takes it -- when a
/// caller accounts for the workspace: the value and validity column of every field, and the
/// histogram index column of a field that holds one.
fn merge_fields(
fields: Vec<Results>,
schema: Arc<SeriesSchema>,
mut budget: Option<&mut Budget<'_>>,
) -> Result<Vec<Series>> {
let width = schema.fields.len();
// Per label set: the metric and its `(point, field, value)` samples.
type Merged = (MetricName, Vec<(usize, usize, GridSample)>);
@@ -1108,8 +1152,18 @@ fn merge_fields(fields: Vec<Results>, schema: Arc<SeriesSchema>) -> Result<Vec<S
for (metric, mut samples) in merged.into_values() {
// One builder serves every label set, so each result series starts explicitly.
builder.begin_series();
// The capacity each field's histogram index column has reached, `0` until its first
// histogram. The columns are handed out inside the series, so what they are charged stays
// charged; only the growth of one series is charged at a time. Only a caller accounting for
// the workspace reads the ledger, so an unaccounted merge allocates nothing for it.
let mut index_capacity = if budget.is_some() {
vec![0usize; width]
} else {
Vec::new()
};
samples.sort_unstable_by_key(|(point, field, _)| (*point, *field));
let mut samples = samples.into_iter().peekable();
let mut pushed = 0usize;
while let Some((point, field, value)) = samples.next() {
row.clear();
row.resize_with(width, || None);
@@ -1117,7 +1171,25 @@ fn merge_fields(fields: Vec<Results>, schema: Arc<SeriesSchema>) -> Result<Vec<S
while let Some((_, field, value)) = samples.next_if(|next| next.0 == point) {
row[field] = Some(value);
}
// The columns of this row, admitted before the builder grows them.
if let Some(budget) = budget.as_deref_mut() {
budget.retain(merge_row_bytes(width))?;
let mut index = 0usize;
for (capacity, value) in index_capacity.iter_mut().zip(&row) {
let histogram = matches!(value, Some(GridSample::Histogram(_)));
if let Some(grown) = index_column_growth(pushed, *capacity, histogram) {
index = index.saturating_add(
grown
.saturating_sub(*capacity)
.saturating_mul(std::mem::size_of::<Option<u32>>()),
);
*capacity = grown;
}
}
budget.retain(index)?;
}
builder.push_samples(point, &row)?;
pushed += 1;
}
output.extend(builder.finish(metric)?);
}
@@ -1187,13 +1259,20 @@ fn group(operand: &Operand<'_>, labels: &MatchingLabels) -> HashMap<GroupKey, Ve
}
/// The first point where two `members` are present and `relevant` holds.
///
/// The list of the group's present points is the only thing this scan allocates, and it is charged
/// as a slot is pushed: the figure follows the list's capacity, so a group of one member and a group
/// whose members carry no sample both cost nothing, however long their axes are. The list is given
/// back once it is dropped.
fn overlap(
operand: &Operand<'_>,
members: &[usize],
relevant: impl Fn(usize) -> bool,
) -> Option<usize> {
tables: usize,
mut budget: Option<&mut Budget<'_>>,
) -> Result<Option<usize>> {
if members.len() < 2 {
return None;
return Ok(None);
}
// Sorting the present points keeps the cost proportional to them, not to the span.
let mut points = Vec::new();
@@ -1204,17 +1283,31 @@ fn overlap(
cursor.at(row).expect("a row never goes back"),
SampleRef::Missing
) {
// The push that starts a new allocation: the capacity the list reaches is what is
// live beside the field's tables, charged twice for the buffer it replaces.
if let Some(budget) = budget.as_deref_mut()
&& points.len() == points.capacity()
{
let capacity = points.capacity().saturating_mul(2).max(VEC_MIN_CAPACITY);
let list = capacity.saturating_mul(2 * std::mem::size_of::<usize>());
budget.step(tables.saturating_add(list))?;
}
points.push(point);
}
}
}
points.sort_unstable();
for pair in points.windows(2) {
if pair[0] == pair[1] && relevant(pair[0]) {
return Some(pair[0]);
}
let duplicate = points
.windows(2)
.find(|pair| pair[0] == pair[1] && relevant(pair[0]))
.map(|pair| pair[0]);
// The list belongs to this scan alone: it is dropped -- a duplicate found or not -- before its
// bytes are given back.
drop(points);
if let Some(budget) = budget {
budget.step(tables)?;
}
None
Ok(duplicate)
}
fn result_metric(
src/promql/src/vector_match/tests.rs
+494
View File
@@ -2563,3 +2563,497 @@ fn a_budgeted_run_reports_the_public_rejection() {
assert_eq!(output.unwrap_err().to_string(), public.to_string());
assert_eq!(seen.len(), 1, "{seen:?}");
}
/// A match group of one member cannot hold two rows at a point, so its duplicate scan collects
/// nothing and the figure charges it no point list: many such groups over a long grid cost their
/// presence flags and their table entries, not one sixteen-byte slot per row of every one-side
/// series.
#[test]
fn singleton_groups_are_charged_no_duplicate_point_list() {
let points = 10_000;
let grid = budget_grid(points);
let schema = Arc::new(schema(1, &grid));
let at: Vec<usize> = (0..points).collect();
// One left-hand series under a `job` no right-hand series carries, and `groups` right-hand
// series that each carry their own `job` and a value at every point: every group has one
// member, nothing matches and no row is filed.
let lhs = vec![jobbed_operand(&schema, "a", "absent", &at)];
let one = |groups: usize| -> Vec<Series> {
(0..groups)
.map(|group| jobbed_operand(&schema, "b", &format!("j{group}"), &at))
.collect()
};
let matching = budget_matching(BinaryOp::Add, MatchingLabels::On(vec![Arc::from("job")]));
let measure = |groups: usize| -> usize {
let (output, seen) = budgeted(&matching, &schema, &lhs, &schema, &one(groups), usize::MAX);
assert!(
output.expect("no group matches").is_empty(),
"{groups}: no `job` is shared"
);
*seen
.iter()
.max()
.expect("the operation asks for its workspace")
};
let single = measure(1);
let many = measure(16);
// What a run of this shape holds: the two presence flags and the intersection of them, one
// byte per grid point, and nothing proportional to the rows the one side carries.
let presence = 3 * points;
assert!(
single <= presence + 64 * 1024,
"single {single} for {presence} presence bytes"
);
assert!(
many <= presence + 64 * 1024,
"sixteen singleton groups {many} for {presence} presence bytes: the scan collected the \
rows of every group"
);
// A limit in the same order as the presence flags alone runs the wider operand: the sixteen
// groups' rows are sixteen times the grid, and a figure that charged their points as held
// storage would need megabytes of them.
let (output, _) = budgeted(&matching, &schema, &lhs, &schema, &one(16), 512 * 1024);
assert!(
output
.expect("no group matches under a small budget")
.is_empty(),
"no `job` is shared"
);
}
/// The duplicate scan's list of present points is one group's workspace: the next group's scan
/// replaces it instead of adding to it, and the figure charges the points the group's members
/// carry rather than one slot per row of every series of the field.
#[test]
fn a_groups_duplicate_scan_is_replaced_by_the_one_after_it() {
// Two members of a group hold the even and the odd points of at most the first half, so the
// scan collects every point of both and no point of the group is present twice: no group is
// reported as a duplicate whatever the other side carries.
let rows_max = 512;
let points = 2 * rows_max;
let grid = budget_grid(points);
let schema = Arc::new(schema(1, &grid));
// The left side holds one value at the first point, whose `job` no right-hand series carries,
// so no group matches and no row is filed.
let lhs = vec![jobbed_operand(&schema, "a", "absent", &[0])];
let one = |groups: usize, rows: usize| -> Vec<Series> {
// The schema is borrowed, so the iterator below captures the reference and not the `Arc`.
let schema = &schema;
(0..groups)
.flat_map(|group| {
let job = format!("j{group}");
(0..2).map(move |member| {
let at: Vec<usize> = (0..rows).map(|row| 2 * row + member).collect();
jobbed_operand(schema, "b", &job, &at)
})
})
.collect()
};
let matching = budget_matching(BinaryOp::Add, MatchingLabels::On(vec![Arc::from("job")]));
let measure = |groups: usize, rows: usize| -> usize {
let (output, seen) = budgeted(
&matching,
&schema,
&lhs,
&schema,
&one(groups, rows),
usize::MAX,
);
assert!(
output
.expect("no point is present on both sides")
.is_empty(),
"{groups} groups of {rows} rows file no row"
);
*seen
.iter()
.max()
.expect("the operation asks for its workspace")
};
// The list is charged: a group whose two members carry `rows_max` points each costs the slots
// of those points, at least one per point.
let short = measure(1, 8);
let long = measure(1, rows_max);
assert!(
long >= short + 2 * (rows_max - 16) * std::mem::size_of::<usize>(),
"short {short}, long {long}: the scan's list is not charged"
);
// And it is replaced, not accumulated: eight groups hold the same list as one group does, so
// what the eight add is their table entries, never their sequences of points.
let eight = measure(8, rows_max);
assert!(
eight <= long + 16 * 1024,
"one group {long}, eight {eight}: the scans' lists added up"
);
}
/// A dense run of `fields` fields writes one value column and one validity bit per field at every
/// output row, and the figure charges those columns once per row: a run four times as wide costs
/// about four times as much, never the width's square.
#[test]
fn a_dense_run_charges_its_output_columns_once_per_row() {
let points = 32;
let grid = budget_grid(points);
let matching = budget_matching(BinaryOp::Add, MatchingLabels::Ignoring(Vec::new()));
// Both operands hold every field at every point, so every field of the single result series
// keeps every point.
let operands = |fields: usize| -> (Arc<SeriesSchema>, Vec<Series>, Vec<Series>) {
let schema = Arc::new(schema(fields, &grid));
let side = |name: &str, value: f64| -> Vec<Series> {
let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap();
builder.begin_series();
let row = vec![Some(value); fields];
for point in 0..points {
builder.push(point, &row).unwrap();
}
builder
.finish(metric(&Labels::from([
("__name__".to_string(), name.to_string()),
("job".to_string(), "x".to_string()),
])))
.unwrap()
.into_iter()
.collect()
};
let a = side("a", 1.0);
let b = side("b", 2.0);
(schema, a, b)
};
let run = |fields: usize, limit: usize| {
let (schema, lhs, rhs) = operands(fields);
budgeted(&matching, &schema, &lhs, &schema, &rhs, limit)
};
let measure = |fields: usize| -> usize {
let (output, seen) = run(fields, usize::MAX);
let output = output.expect("the fields pair");
assert_eq!(output.len(), 1, "{fields} fields: {output:?}");
assert_eq!(
run_ranges(&output[0]),
vec![(0, points)],
"{fields} fields answer every point"
);
// Every field of the single result series holds the sum of both operands at every point.
for field in 0..fields {
let mut cursor = output[0].fields.as_slice()[field].cursor();
let mut rows = 0;
for (_, row) in output[0].time.positions() {
match cursor.at(row).expect("a row never goes back") {
SampleRef::Float(value) => assert_eq!(value, 3.0, "{fields} fields"),
_ => panic!("a field of the sum holds no float"),
}
rows += 1;
}
assert_eq!(rows, points, "{fields} fields answer every point");
}
*seen
.iter()
.max()
.expect("the operation asks for its workspace")
};
// What one field of this run costs: its rows, its slices of the triples, its columns and its
// share of the builder. Every term grows with the width and none with its square, so the cost
// per field is what stays put as the run widens.
let narrow = measure(4);
let wide = measure(64);
assert!(
wide <= 2 * (64 / 4) * narrow,
"sixteen times the fields cost {wide} beside the {narrow} four cost: a figure that charged \
every field's columns to every field's sample multiplies the cost per field by the width"
);
// The columns are part of the figure, so it is not trivially small.
assert!(
narrow >= points * 4 * (std::mem::size_of::<f64>() + 1),
"four fields cost {narrow}, less than their columns"
);
// A run of every width is admitted by a limit in the order of its columns, not of the width's
// square: the widest dense result is not a phantom megabyte per field.
for fields in [1, 4, 16, 64] {
assert!(
run(fields, 2 * 1024 * 1024).0.is_ok(),
"{fields} fields are refused under two megabytes, {wide} bytes for sixty-four"
);
}
}
/// Fields that hold different points of one label set still write every field's column at every
/// output row: the figure charges the schema's fields per row, so a run whose fields never share a
/// point is covered as fully as a dense one, and each field keeps its own values.
#[test]
fn fields_holding_disjoint_points_are_charged_their_columns() {
let points = 64;
let fields = 4;
let grid = budget_grid(points);
let schema = Arc::new(schema(fields, &grid));
// Field `f` holds a value only at the points `f` apart, so the four fields share no point and
// each output row fills one field's column and leaves the other three null.
let side = |name: &str, offset: f64| -> Vec<Series> {
let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap();
builder.begin_series();
for point in 0..points {
let mut row = vec![None; fields];
row[point % fields] = Some(point as f64 + offset);
builder.push(point, &row).unwrap();
}
builder
.finish(metric(&Labels::from([
("__name__".to_string(), name.to_string()),
("job".to_string(), "x".to_string()),
])))
.unwrap()
.into_iter()
.collect()
};
let lhs = side("a", 1.0);
let rhs = side("b", 2.0);
let matching = budget_matching(BinaryOp::Add, MatchingLabels::Ignoring(Vec::new()));
let (output, seen) = budgeted(&matching, &schema, &lhs, &schema, &rhs, usize::MAX);
let output = output.expect("every point of every field is kept");
assert_eq!(output.len(), 1, "{output:?}");
assert_eq!(
run_ranges(&output[0]),
vec![(0, points)],
"one series over every point"
);
// Each point holds the sum of both operands in the field it belongs to and nothing elsewhere.
let mut cursors: Vec<RowCursor<'_>> = output[0]
.fields
.iter()
.map(|field| field.cursor())
.collect();
for (point, row) in output[0].time.positions() {
for (field, cursor) in cursors.iter_mut().enumerate() {
let value = match cursor.at(row).expect("a row never goes back") {
SampleRef::Missing => None,
SampleRef::Float(value) => Some(value),
SampleRef::Histogram(_) => panic!("a float field holds no histogram"),
};
// The lhs holds `point + 1` in its field and the rhs `point + 2`, so their sum is
// `2 * point + 3`, in the field the point belongs to and nowhere else.
let expected = (field == point % fields).then_some(2.0 * point as f64 + 3.0);
assert_eq!(value, expected, "point {point}, field {field}");
}
}
// The figure covers the composite a row costs: the rows the fields filed, the triple of each
// kept sample and the value and validity columns every field writes at every output row --
// the columns of the fields a row leaves empty included, since the builder writes them too. No
// row filling more than one field lets the columns be divided among the fields.
let peak = *seen
.iter()
.max()
.expect("the operation asks for its workspace");
let filed = points * RESULT_ROW_BYTES;
let triples = points * 2 * std::mem::size_of::<(usize, usize, GridSample)>();
let columns = points * fields * (std::mem::size_of::<f64>() + 1);
assert!(
peak >= filed + triples + columns,
"peak {peak} below the {filed} rows, {triples} triples and {columns} columns it holds"
);
assert!(
peak <= points * (fields * 2 * (std::mem::size_of::<f64>() + 1) + 4 * RESULT_ROW_BYTES)
+ 128 * 1024,
"peak {peak} for {points} rows of {fields} fields"
);
}
/// A duplicate scan is charged the points it collects: a group whose members carry no sample
/// collects nothing, however long their axes are, while the points a group does carry are its own
/// slots.
#[test]
fn a_duplicate_scan_is_charged_the_points_it_collects() {
let points = 100_000;
let grid = budget_grid(points);
let schema = Arc::new(schema(1, &grid));
// A member of the group with no value at any point: its axis is the whole grid.
let null_member = |name: &str| -> Series {
Series {
metric: metric(&Labels::from([
("__name__".to_string(), name.to_string()),
("job".to_string(), "x".to_string()),
])),
time: TimeAxis::Grid(GridRange::try_new(grid.clone(), 0, points).unwrap()),
fields: SeriesFields::one(Samples::from_float(Float64Array::new_null(points))),
}
};
let sparse_member = |name: &str, at: &[usize]| -> Series {
let labels = Labels::from([
("__name__".to_string(), name.to_string()),
("job".to_string(), "x".to_string()),
]);
let values: Points = at
.iter()
.map(|&position| (position, f64::to_bits(position as f64)))
.collect();
sparse_series(&schema, &labels, &values).expect("the operand holds a value")
};
// The left side carries one value, at a point no member carries, so the group is never a
// duplicate and no row is filed.
let lhs = vec![sparse_member("a", &[90_000])];
let matching = budget_matching(BinaryOp::Add, MatchingLabels::On(vec![Arc::from("job")]));
// What the run holds beside the list: three presence flags per grid point.
let floor = 3 * points;
let run = |rhs: &[Series], limit: usize| -> Result<Vec<Series>, DataFusionError> {
budgeted(&matching, &schema, &lhs, &schema, rhs, limit).0
};
// Two members of one group, each with a value at no point: their axes are the whole grid and
// the list stays empty.
let empty_group = vec![null_member("b"), null_member("c")];
assert!(
binary_vectors(&matching, &schema, &lhs, &schema, &empty_group)
.expect("no member carries a sample")
.is_empty()
);
let output = run(&empty_group, floor + 64 * 1024).expect("an empty list costs nothing");
assert!(output.is_empty(), "{output:?}");
// A few points of each member are the list's own slots, and still fit beside the flags.
let few = vec![
sparse_member("b", &[10, 20, 30, 40]),
sparse_member("c", &[50, 60, 70, 80]),
];
assert!(
run(&few, floor + 64 * 1024)
.expect("a few points cost a few slots")
.is_empty()
);
// Thousands of points are charged as they are collected: the same limit refuses them, since
// their slots alone are tens of kilobytes.
let many = vec![
sparse_member("b", &(0..4_000).collect::<Vec<_>>()),
sparse_member("c", &(4_000..8_000).collect::<Vec<_>>()),
];
let error = run(&many, floor + 64 * 1024)
.expect_err("the collected points are charged as they are pushed");
assert!(
matches!(error, DataFusionError::ResourcesExhausted(_)),
"{error}"
);
}
/// A field's histogram index column is charged as it grows: the builder creates it over the rows
/// before the field's first histogram and grows it with every row after, so a wide result whose
/// fields meet their histograms late holds those columns beside its value columns.
#[test]
fn sparse_wide_results_are_charged_their_index_columns() {
let points = 1_025;
let fields = 64;
let grid = budget_grid(points);
let schema = Arc::new(schema(fields, &grid));
// The value of a field at its one histogram point.
let histogram = || {
GridSample::Histogram(Box::new(
FloatHistogram::try_from_view(&OwnedHistogram::parse("0/0/0/0/0/0/////").view())
.expect("an empty histogram is readable"),
))
};
// Field `f` holds a value only at the points `f` apart: a float at every one of them but the
// last, and an empty histogram at its single point of the last `fields` rows. Every field meets
// its first histogram late and keeps the column to the end.
let side = |name: &str, offset: f64| -> Vec<Series> {
let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap();
builder.begin_series();
for point in 0..points {
let mut row: Vec<Option<GridSample>> = Vec::with_capacity(fields);
row.resize_with(fields, || None);
row[point % fields] = Some(if point >= points - fields {
histogram()
} else {
GridSample::Float(point as f64 + offset)
});
builder.push_samples(point, &row).unwrap();
}
builder
.finish(metric(&Labels::from([
("__name__".to_string(), name.to_string()),
("job".to_string(), "x".to_string()),
])))
.unwrap()
.into_iter()
.collect()
};
let matching = budget_matching(BinaryOp::Add, MatchingLabels::Ignoring(Vec::new()));
let (output, seen) = budgeted(
&matching,
&schema,
&side("a", 1.0),
&schema,
&side("b", 2.0),
usize::MAX,
);
let output = output.expect("every field adds up");
assert_eq!(output.len(), 1, "{output:?}");
// Every point keeps the field it belongs to and nothing in the others, the floats add up and
// the empty histograms still hold an empty histogram.
let mut cursors: Vec<RowCursor<'_>> = output[0]
.fields
.iter()
.map(|field| field.cursor())
.collect();
let mut rows = 0;
for (point, row) in output[0].time.positions() {
rows += 1;
for (field, cursor) in cursors.iter_mut().enumerate() {
let owned = field == point % fields;
match cursor.at(row).expect("a row never goes back") {
SampleRef::Float(value) => {
assert!(owned, "point {point} keeps field {field}");
assert!(point < points - fields, "point {point} is not a histogram");
assert_eq!(value, 2.0 * point as f64 + 3.0, "point {point}");
}
SampleRef::Histogram(view) => {
assert!(owned, "point {point} keeps field {field}");
assert!(point >= points - fields, "point {point} is a histogram");
assert_eq!(view.count, 0.0, "empty histograms add up empty");
assert!(
view.positive.buckets.is_empty() && view.negative.buckets.is_empty(),
"point {point} keeps an empty histogram"
);
}
SampleRef::Missing => assert!(!owned, "point {point} holds field {field}"),
}
}
}
assert_eq!(rows, points, "every point is a row");
// The capacity Rust gives these columns for this fixture, from its growth rule alone: a push
// into a full list allocates at least four entries and otherwise doubles. One value column per
// field, pushed once per row, and one index column per field, created over the rows before its
// first histogram and pushed once per row after.
let grow = |capacity: usize, len: usize| -> usize {
if len < capacity {
capacity
} else {
(2 * capacity).max(4)
}
};
let mut values = 0usize;
let mut index = vec![0usize; fields];
for pushed in 0..points {
values = grow(values, pushed);
let late = pushed >= points - fields;
for (field, capacity) in index.iter_mut().enumerate() {
if *capacity == 0 {
if late && pushed % fields == field {
// Created over the `pushed` rows before the field's first histogram, then
// pushed to, so its length is `pushed + 1` like every column made before it.
*capacity = grow(pushed, pushed);
}
} else {
*capacity = grow(*capacity, pushed);
}
}
}
let columns = fields * values * std::mem::size_of::<f64>()
+ index.iter().fold(0usize, |bytes, capacity| {
bytes.saturating_add(capacity.saturating_mul(std::mem::size_of::<Option<u32>>()))
});
let peak = *seen
.iter()
.max()
.expect("the operation asks for its workspace");
assert!(
peak >= columns,
"peak {peak} below the {columns} bytes of value and index columns the result holds"
);
assert!(
peak <= 2 * columns,
"peak {peak} is more than the {columns} bytes of columns the result holds"
);
}