diff --git a/src/promql/src/native/execute/tests.rs b/src/promql/src/native/execute/tests.rs index f18a42f46d0..3b197d251ef 100644 --- a/src/promql/src/native/execute/tests.rs +++ b/src/promql/src/native/execute/tests.rs @@ -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>> { + (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); +} diff --git a/src/promql/src/vector_match.rs b/src/promql/src/vector_match.rs index 02cee43811f..8f4e49ab250 100644 --- a/src/promql/src/vector_match.rs +++ b/src/promql/src/vector_match.rs @@ -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, Arc)>(); @@ -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], @@ -410,9 +415,7 @@ fn field_metadata_bytes( .saturating_add(2 * std::mem::size_of::()) .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::() * 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]) /// /// 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::() + 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::()) - .saturating_add(2 * std::mem::size_of::()) - .saturating_add(values) - .saturating_add(std::mem::size_of::>()); + .saturating_add(2 * std::mem::size_of::()); 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::())), ) - // 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::>())) + .saturating_add(fields.saturating_mul(std::mem::size_of::())) // 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::() + 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 { + 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, schema: Arc) -> Result> { +/// +/// 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, + schema: Arc, + mut budget: Option<&mut Budget<'_>>, +) -> Result> { 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, schema: Arc) -> Result, schema: Arc) -> Result>()), + ); + *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, members: &[usize], relevant: impl Fn(usize) -> bool, -) -> Option { + tables: usize, + mut budget: Option<&mut Budget<'_>>, +) -> Result> { 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::()); + 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( diff --git a/src/promql/src/vector_match/tests.rs b/src/promql/src/vector_match/tests.rs index d3ce3ef74ee..d2c35a47710 100644 --- a/src/promql/src/vector_match/tests.rs +++ b/src/promql/src/vector_match/tests.rs @@ -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 = (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 { + (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 { + // 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 = (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::(), + "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, Vec, Vec) { + let schema = Arc::new(schema(fields, &grid)); + let side = |name: &str, value: f64| -> Vec { + 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::() + 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 { + 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> = 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::() + 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::() + 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, 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::>()), + sparse_member("c", &(4_000..8_000).collect::>()), + ]; + 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 { + let mut builder = GridSeriesBuilder::try_new(schema.clone()).unwrap(); + builder.begin_series(); + for point in 0..points { + let mut row: Vec> = 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> = 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::() + + index.iter().fold(0usize, |bytes, capacity| { + bytes.saturating_add(capacity.saturating_mul(std::mem::size_of::>())) + }); + 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" + ); +}