rudb_native/stats.rs
1//! Building a table's statistics sections from the table's own columns.
2//!
3//! The same meeting place `graph` is, for the other document. `rudb-stats` at rank 5 knows what a
4//! column summary says and knows nothing about a file; the rest of this crate knows how to put an
5//! opaque payload in a file and nothing about what one means. Building a summary for a real table
6//! means reading the column back, so it happens here, in the crate allowed to see both.
7//!
8//! Everything here obeys `spec/stats/03-the-file-format.md` section 3.1, which is the graph
9//! document's section 3.1 applied to a second kind of payload: delete every statistics section and
10//! no query changes its answer, only the time. That is why [`summary`] and [`sketches`] answer with
11//! an [`Option`] and not a [`Result`]. There is no failure they could report that is not answered
12//! by planning the query the way it was planned before the section existed.
13//!
14//! # The invariant has teeth here that it does not have in the graph layer
15//!
16//! A key map can only make a join faster. A summary can answer a query: a `COUNT(DISTINCT c)` comes
17//! out of one without the column being touched. So the thing that has to survive is not only *is
18//! the section there* but *is the number in it exact*, and [`Summary::distinct_class`] is where that
19//! lives. This module's job is to never write [`Class::Exact`] onto a number that is not, which in
20//! practice means one rule: the sketch says whether it overflowed, and everything else follows from
21//! that answer rather than from what the writer hoped.
22//!
23//! # One pass, and what that costs
24//!
25//! Section 3.7 gives the statistics build ten percent of the native write time, and the way to stay
26//! inside it is not to be clever but to read the column once. [`build_summary`] takes one scan and
27//! computes every field of the summary and the sketch from it, so the cost of statistics on a write
28//! is the cost of one more read of each column asked for, and no column is read twice.
29//!
30//! # The per stripe rule
31//!
32//! Section 3.8 says per stripe structures are written only for the columns that get read, and the
33//! arithmetic behind that is not close: sixteen `lineitem` columns at SF100, sketched per stripe
34//! even at the small k a stripe sketch keeps, come to several hundred megabytes against a budget of
35//! two percent. So the default is a merged sketch and nothing else, and [`Sketches::stripes`] being
36//! empty is the state the rule says most columns are in rather than a degraded one.
37//!
38//! [`read_columns`] is what this build promotes a column with. It reads the promoted set off the
39//! file, which today means the columns that already carry a key map or a forward link, because
40//! those are the columns something has declared a relationship or a key over and section 3.8 names
41//! them directly. Document 06's observation log is the other source the spec names and it is not
42//! built yet, so when it arrives it adds columns to this list and changes nothing else here.
43//!
44//! Promotion costs no extra hashing. The column is read once and hashed once either way, and what
45//! changes is where the counting is reset. [`build_summary_for`] has the argument.
46
47use std::cmp::Ordering;
48use std::path::Path;
49use std::sync::Arc;
50use std::time::{Duration, Instant};
51
52use rudb_common::bounds::{self, Bound};
53use rudb_common::stat::Class;
54use rudb_common::{LogicalType, Result, Value};
55use rudb_encoding::sketch::{DEFAULT_K, Sketch};
56use rudb_stats::{Order, STRIPE_K, Sketches, Summary, sketches::HEADER_BYTES as SKETCH_HEADER};
57use rudb_storage::count::{Counts, countable};
58use rudb_vector::{Data, Form, Validity, Vector};
59
60use crate::section::{self, Attachment};
61use crate::{Catalog, Reader, invalid};
62
63/// The share of a table's stored column bytes its statistics sections are allowed to cost together.
64///
65/// Two percent, per section 3.8, and kept apart from the graph layer's ten percent rather than
66/// pooled with it. Two budgets that share a pot are two budgets where the one that runs first wins,
67/// and a table whose key maps happened to be built before its summaries would then have no
68/// summaries for a reason that has nothing to do with summaries. They are counted separately for the
69/// same reason they are two documents.
70pub const BUDGET_SHARE: u64 = 2;
71
72/// The size below which a table's statistics sections always fit, whatever the share works out to.
73///
74/// The same floor and the same argument as `graph::BUDGET_FLOOR`. A summary is a few hundred bytes
75/// on a table of any size and two percent of a small, well compressed column is less than that, so
76/// the pure rule would throw away the cheapest structure in the system for being expensive.
77pub const BUDGET_FLOOR: u64 = 64 * 1024;
78
79/// What one column's statistics cost and what they say.
80#[derive(Debug, Clone)]
81pub struct Built {
82 /// Which column was summarized.
83 pub column: usize,
84 /// Rows in the column, nulls included.
85 pub rows: u64,
86 /// Distinct non-null values, as the summary reports them.
87 pub distinct: u64,
88 /// Whether that distinct count is exact rather than a sketch estimate.
89 pub exact: bool,
90 /// Which way the values run.
91 pub order: Order,
92 /// What the summary section takes in the file.
93 pub summary_bytes: usize,
94 /// What the sketches section takes in the file.
95 pub sketch_bytes: usize,
96 /// How many per stripe sketches went in it, which is zero for a column the per stripe rule did
97 /// not promote and is most of them.
98 pub stripes: usize,
99 /// What the column takes in the file, which is what the budget is a share of.
100 pub column_bytes: u64,
101 /// Whether the sections were kept. False means they were built, measured, and found to cost more
102 /// than section 3.8 allows, so the file does not have them and every query plans as though
103 /// statistics had never been implemented.
104 pub built: bool,
105 /// How long the build took, the reading of the column included.
106 pub build: Duration,
107}
108
109impl Built {
110 /// Both sections together, which is what the budget spends.
111 #[must_use]
112 pub fn bytes(&self) -> usize {
113 self.summary_bytes + self.sketch_bytes
114 }
115}
116
117/// A column's summary and its sketches, which are built together because they are one pass.
118#[derive(Debug, Clone)]
119pub struct Stats {
120 /// What the column says about itself.
121 pub summary: Summary,
122 /// The sketch the distinct count came out of.
123 pub sketches: Sketches,
124}
125
126/// Builds the summary and the sketches for one column of a committed table.
127///
128/// # Errors
129///
130/// If the column cannot be read, is past the end of the table, or is of a type with no hash rule.
131/// The last one is refused by name rather than approximated: the types without a rule are the
132/// interval and the nested ones, a summary of one would carry a distinct count of zero that nothing
133/// could tell from a column of nulls, and none of TPC-H or ClickBench has one.
134pub fn build_summary(reader: &Reader, column: usize) -> Result<Stats> {
135 build_summary_for(reader, column, false)
136}
137
138/// The same, keeping a sketch per stripe as well as the merged one when `per_stripe` is set.
139///
140/// Whether to set it is section 3.8's rule and not a caller's taste: per stripe structures are
141/// written only for the columns that get read, because sixteen `lineitem` columns at SF100 come to
142/// several hundred megabytes of them against a budget of two percent. [`read_columns`] is what this
143/// build answers that question with.
144///
145/// The extra sketches cost no extra hashing. Each stripe is counted into its own [`Counts`] at the
146/// column's k, the merged sketch is the union of those, which is exact because they are all at the
147/// same k, and each one is written down at [`rudb_stats::STRIPE_K`] through [`Sketch::narrowed`],
148/// which is exact because a bottom-k of a bottom-k is a bottom-k. So the column is read once and
149/// hashed once either way, and the difference between a promoted column and an ordinary one is
150/// where the counting is reset and how much of it is written.
151///
152/// # Errors
153///
154/// If the column cannot be read, is past the end of the table, or is of a type with no hash rule.
155/// The last one is refused by name rather than approximated: the types without a rule are the
156/// interval and the nested ones, a summary of one would carry a distinct count of zero that nothing
157/// could tell from a column of nulls, and none of TPC-H or ClickBench has one.
158pub fn build_summary_for(reader: &Reader, column: usize, per_stripe: bool) -> Result<Stats> {
159 let fields = reader.table().fields();
160 let Some(field) = fields.get(column) else {
161 return Err(invalid(&format!(
162 "column {column} is past the {} of table {}",
163 fields.len(),
164 reader.table().name()
165 )));
166 };
167 if !countable(&field.ty) {
168 return Err(invalid(&format!(
169 "a summary of {} needs a hash rule, and {} has none",
170 field.name, field.ty
171 )));
172 }
173 let blind = || {
174 // A blind column: a form `rudb_storage::count` has no arm for turned up, so its sketch is
175 // missing rows and says nothing about which. A distinct count that is too low is the one
176 // error an estimator has no defence against, so the column gets no summary at all rather
177 // than a summary with a number in it nothing can check.
178 invalid(&format!(
179 "column {} of {} holds a form with no hash rule, so it has no sketch",
180 field.name,
181 reader.table().name()
182 ))
183 };
184
185 let mut whole = Counts::new(1);
186 let mut stripes = Vec::new();
187 let mut pass = Pass::new(&field.ty, reader.table().generation());
188 for (at, stripe) in reader.stripe_parts().into_iter().enumerate() {
189 pass.open_stripe((at as u64, 0));
190 let mut counted = per_stripe.then(|| Counts::new(1));
191 for part in stripe {
192 let chunk = reader.read(part, &[column])?;
193 match counted.as_mut() {
194 Some(counted) => counted.add(&chunk),
195 None => whole.add(&chunk),
196 }
197 pass.scan(chunk.column(0)?);
198 }
199 pass.close_stripe();
200 if let Some(counted) = counted {
201 stripes.push(counted.sketch(0).ok_or_else(blind)?);
202 }
203 }
204 if !per_stripe {
205 return Ok(pass.finish(whole.sketch(0).ok_or_else(blind)?, Vec::new()));
206 }
207 let mut merged = Sketch::new(DEFAULT_K)?;
208 for stripe in &stripes {
209 merged = merged.union(stripe)?;
210 }
211 let narrowed =
212 stripes.iter().map(|stripe| stripe.narrowed(STRIPE_K)).collect::<Result<Vec<_>>>()?;
213 Ok(pass.finish(merged, narrowed))
214}
215
216/// What one stripe says about the order of its rows, kept until every stripe is in.
217#[derive(Debug)]
218struct Piece {
219 key: (u64, u64),
220 first: Option<Bound>,
221 last: Option<Bound>,
222 ascending: bool,
223 descending: bool,
224 runs: u64,
225}
226
227/// One scan of one column, in `rid` order, for everything the sketch does not answer.
228///
229/// In `rid` order because the order fields depend on it. A pass that read the parts in any other
230/// order would report a column as unordered that is sorted, which costs a plan and not an answer,
231/// and would report the run count of a shuffle, which is worse because it is a number rather than a
232/// flag and looks like it was measured.
233///
234/// The distinct count is not here. That is `rudb_storage::count::Counts`, which walks a vector by
235/// its form rather than a row at a time and which a column of a million runs costs one hash. Doing
236/// it twice would double the expensive half of the build and the budget is ten percent of the write.
237#[derive(Debug)]
238struct Pass {
239 rows: u64,
240 nulls: u64,
241 low: Option<Bound>,
242 high: Option<Bound>,
243 /// False once a non-null value turned up that has no ordered bound, which makes both ends
244 /// unusable rather than merely absent.
245 bounded: bool,
246 ascending: bool,
247 descending: bool,
248 runs: u64,
249 previous: Option<Bound>,
250 bytes: u64,
251 widest: u64,
252 generation: u64,
253 /// The two ends of the stripe being read, kept apart rather than as a pair so that each one can
254 /// be compared against and refilled on its own. A pair would have to be taken out and put back
255 /// whole, which is the move that made this pass allocate.
256 stripe_low: Option<Bound>,
257 stripe_high: Option<Bound>,
258 stripes: Vec<(Bound, Bound)>,
259 /// Where the stripe being read sits in the table, and the first value it held.
260 ///
261 /// A writer fed by several pipeline instances gets its stripes in the order they finished
262 /// rather than the order they sit in, and sorts them by this key when it commits. The order
263 /// fields are about adjacent rows, so they are read a stripe at a time into [`Piece`]s and put
264 /// together in key order at the end, which is the rid order the reader will see.
265 key: (u64, u64),
266 first: Option<Bound>,
267 pieces: Vec<Piece>,
268 /// What one value of this column takes, when every value takes the same.
269 ///
270 /// Read off the type once rather than off each value, because for every fixed width column it is
271 /// a constant and asking a value for it is a branch a hundred million times to hear the same
272 /// number. `None` is a variable width type and those are measured per value.
273 fixed: Option<u64>,
274 /// The scale of a decimal column, so that an integer read out of a vector becomes the bound the
275 /// column's other writers would have written for the same value.
276 scale: Option<u8>,
277 /// The last dictionary this pass read, so that a column whose vectors share one reads it once.
278 coded: Option<Coded>,
279}
280
281impl Pass {
282 fn new(ty: &LogicalType, generation: u64) -> Self {
283 Self {
284 rows: 0,
285 nulls: 0,
286 low: None,
287 high: None,
288 bounded: true,
289 ascending: true,
290 descending: true,
291 runs: 0,
292 previous: None,
293 bytes: 0,
294 widest: 0,
295 generation,
296 stripe_low: None,
297 stripe_high: None,
298 stripes: Vec::new(),
299 key: (0, 0),
300 first: None,
301 pieces: Vec::new(),
302 fixed: fixed_width(ty),
303 scale: bounds::scale_of(ty),
304 coded: None,
305 }
306 }
307
308 /// One vector of the column, a vector at a time where the layout allows it and a row at a time
309 /// where it does not.
310 fn scan(&mut self, vector: &Vector) {
311 if self.scan_flat(vector) || self.scan_dictionary(vector) {
312 return;
313 }
314 self.scan_rows(vector);
315 }
316
317 /// One vector of a flat signed column, with the layout matched on once instead of once a row.
318 ///
319 /// `false` if the vector is not one of those, and the caller falls back to [`Self::scan_rows`].
320 ///
321 /// This is where the build's time went. [`Self::scan_rows`] asks `Vector::signed_at` for every
322 /// row, and that is a validity test, a match over the body forms and a second match over the
323 /// dozen layouts, and then the answer is wrapped in a [`Bound`] and compared through
324 /// [`Bound::order`], which is another match, four times. Measured on TPC-H SF1 that came to
325 /// about 355 instructions for a value whose whole job is three comparisons: 53.4 G instructions
326 /// of the 60.8 G the statistics added to the write, against 7.9 G for the sketch that hashes
327 /// every one of the same values. The sketch was never the expensive half.
328 ///
329 /// Matched once, the loop underneath is a validity bit and three integer compares. The layouts
330 /// are the signed group and not the unsigned one, because `Vector::signed_at` reads the signed
331 /// group and this has to agree with the path it is replacing rather than be better than it.
332 fn scan_flat(&mut self, vector: &Vector) -> bool {
333 if vector.form() != Form::Flat {
334 return false;
335 }
336 let Some(data) = vector.data() else { return false };
337 let rows = vector.len();
338 let validity = vector.validity();
339 macro_rules! signed {
340 ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
341 match data {
342 $(Data::$variant(held) => {
343 let held: &[$native] = held;
344 if held.len() < rows {
345 return false;
346 }
347 let spread = spread(rows, validity, |row| i128::from(held[row]));
348 let width = self.fixed.unwrap_or(8);
349 self.fold(Reduced {
350 rows: spread.rows,
351 nulls: spread.nulls,
352 values: spread.values,
353 bytes: width.saturating_mul(spread.values),
354 widest: if spread.values > 0 { width } else { 0 },
355 ascents: spread.ascents,
356 descents: spread.descents,
357 ends: (spread.values > 0).then(|| Ends {
358 low: self.bound(spread.low),
359 high: self.bound(spread.high),
360 first: self.bound(spread.first),
361 last: self.bound(spread.last),
362 }),
363 });
364 return true;
365 })+
366 _ => false,
367 }
368 };
369 }
370 rudb_vector::for_each_layout!(signed, signed)
371 }
372
373 /// One vector of a dictionary column, with the values compared once each instead of once a row.
374 ///
375 /// `false` if the vector is not one, or if the dictionary is too big for this to be worth it, or
376 /// if its entries turn out not to be orderable against each other.
377 ///
378 /// A dictionary vector is where the rest of the build's time went, and it is most of what a load
379 /// hands the writer: on a TPC-H SF1 `lineitem` about seven vectors in ten arrive dictionary
380 /// coded, the five string columns among them. Reading one a row at a time costs a code lookup
381 /// and then all the work the flat path was doing, and for a string column it costs a byte
382 /// comparison against the value before it, for a column whose whole point is that it holds a few
383 /// dozen distinct values.
384 ///
385 /// So the dictionary is read once and then the rows are read against what it came to. See
386 /// [`Coded`] for what that is and [`Self::read_dictionary`] for how it is built.
387 fn scan_dictionary(&mut self, vector: &Vector) -> bool {
388 let Some((codes, values)) = vector.shared_dictionary_parts() else { return false };
389 let rows = vector.len();
390 if codes.len() < rows {
391 return false;
392 }
393 let held = match self.coded.take() {
394 Some(held) if Arc::ptr_eq(&held.values, values) => held,
395 // A dictionary this pass has not read. Wider than the vector it codes means reading it
396 // costs more than the rows it is about are worth, so that one goes back to the row at a
397 // time pass rather than being read at all.
398 _ => {
399 if values.len() > rows {
400 return false;
401 }
402 match self.read_dictionary(values) {
403 Some(read) => read,
404 None => return false,
405 }
406 }
407 };
408 let out = held.reduce(codes, rows, vector.validity());
409 self.coded = Some(held);
410 self.fold(out);
411 true
412 }
413
414 /// Reads a dictionary into the positions and the widths its codes stand for.
415 ///
416 /// `None` for a dictionary holding a value with no ordered bound, or a pair this build cannot
417 /// order against each other. Either way the vector goes back to [`Self::scan_rows`], which has
418 /// the rule for what a value like that does to a column's ends and is the one place it lives.
419 ///
420 /// Every entry is ordered against every other, which is one sort of a few dozen things, and then
421 /// each code carries the position its value holds in that order. Entries that order equal share
422 /// a position, so a dictionary that happens to hold one value twice says what the row at a time
423 /// pass says rather than seeing a step between the two copies of it.
424 fn read_dictionary(&self, values: &Arc<Vector>) -> Option<Coded> {
425 let mut entries = Vec::with_capacity(values.len());
426 // row at a time: these are a dictionary's entries rather than a column's rows, and there are
427 // a few dozen of them behind the thousands of rows that code against them. The third arm
428 // builds a `Value` and is the one the checker is looking for, and it runs for a float
429 // dictionary and for nothing else.
430 for at in 0..values.len() {
431 if values.is_null_at(at) {
432 entries.push(None);
433 continue;
434 }
435 // Derived the way `scan_rows` derives it, arm for arm, because the two have to agree on
436 // what bound a value has. A date read as a signed integer and a date read through
437 // `Bound::of_value` are not required to be the same bound, and a column whose vectors
438 // took different paths would be comparing one against the other.
439 let entry = match values.signed_at(at) {
440 Some(signed) => Some((self.bound(signed), self.fixed.unwrap_or(8))),
441 None => match values.bytes_at(at) {
442 Some(bytes) => Some((Bound::Bytes(bytes.to_vec()), bytes.len() as u64)),
443 None => {
444 let value = values.value_at(at);
445 let wide = self.fixed.unwrap_or_else(|| width(&value));
446 Bound::of_value(&value).map(|bound| (bound, wide))
447 }
448 },
449 };
450 entries.push(Some(entry?));
451 }
452 let mut order = (0..entries.len()).filter(|&at| entries[at].is_some()).collect::<Vec<_>>();
453 order.sort_by(|&one, &other| {
454 bound_of(&entries, one).order(bound_of(&entries, other)).unwrap_or(Ordering::Equal)
455 });
456 // The walk that hands out the positions is also what checks the sort meant anything: a pair
457 // this build cannot order sorted to wherever it happened to sit, so an unordered pair here
458 // is the whole dictionary going back to the row at a time pass.
459 let mut codes = vec![None; entries.len()];
460 let mut bounds = Vec::new();
461 for (at, &code) in order.iter().enumerate() {
462 if at > 0 {
463 match bound_of(&entries, order[at - 1]).order(bound_of(&entries, code)) {
464 Some(Ordering::Less) => bounds.push(bound_of(&entries, code).clone()),
465 Some(Ordering::Equal) => {}
466 Some(Ordering::Greater) | None => return None,
467 }
468 } else {
469 bounds.push(bound_of(&entries, code).clone());
470 }
471 let width = entries[code].as_ref().map_or(0, |(_, width)| *width);
472 codes[code] = Some(((bounds.len() - 1) as u32, width));
473 }
474 Some(Coded { values: Arc::clone(values), codes, bounds })
475 }
476
477 /// Folds what one vector came to into the pass, which is where the sequential half is settled.
478 ///
479 /// The order flags and the run count are a question about adjacent rows, so a vector at a time
480 /// pass cannot answer them alone. It can answer them about its own rows and hand back the two
481 /// ends of itself, and then one comparison against the value before the vector joins the two
482 /// halves. That is what this does, and it is the whole of the sequential dependency.
483 fn fold(&mut self, one: Reduced) {
484 self.rows += one.rows;
485 self.nulls += one.nulls;
486 self.bytes = self.bytes.saturating_add(one.bytes);
487 self.widest = self.widest.max(one.widest);
488 let Some(ends) = one.ends else { return };
489 match self.previous.take() {
490 None => {
491 self.runs = 1;
492 self.first = Some(ends.first.clone());
493 }
494 Some(previous) => self.run(Some(previous.order(&ends.first))),
495 }
496 self.runs += one.descents;
497 if one.descents > 0 {
498 self.ascending = false;
499 }
500 if one.ascents > 0 {
501 self.descending = false;
502 }
503 if takes(&self.low, &ends.low, Ordering::Less) {
504 self.low = Some(ends.low.clone());
505 }
506 if takes(&self.stripe_low, &ends.low, Ordering::Less) {
507 self.stripe_low = Some(ends.low);
508 }
509 if takes(&self.high, &ends.high, Ordering::Greater) {
510 self.high = Some(ends.high.clone());
511 }
512 if takes(&self.stripe_high, &ends.high, Ordering::Greater) {
513 self.stripe_high = Some(ends.high);
514 }
515 self.previous = Some(ends.last);
516 }
517
518 /// The bound this column writes for a signed value, which a decimal column spells differently.
519 fn bound(&self, signed: i128) -> Bound {
520 match self.scale {
521 Some(scale) => Bound::Scaled { unscaled: signed, scale },
522 None => Bound::Int(signed),
523 }
524 }
525
526 /// One vector, a row at a time, for every column the fast path above does not read.
527 ///
528 /// The floats, the unsigned widths, the strings, and every form that is not flat. A string
529 /// column is here rather than in the fast path because its values are not a slice of one width
530 /// and its ends are byte comparisons, and `bytes_value` is already written to not allocate.
531 fn scan_rows(&mut self, vector: &Vector) {
532 // row at a time: the run count and the order flags are a sequential dependency. Whether this
533 // value is below the one before it is a question about a pair of adjacent rows, so there is
534 // no shape of this loop that answers it a vector at a time, and the two typed accessors
535 // below are loads against a slice rather than value construction. What the checker is
536 // looking for is the third arm, which does build a `Value`, and that one runs for a float
537 // column and for a form the first two cannot read and for nothing else.
538 for row in 0..vector.len() {
539 self.rows += 1;
540 if vector.is_null_at(row) {
541 self.nulls += 1;
542 continue;
543 }
544 if let Some(signed) = vector.signed_at(row) {
545 let bound = match self.scale {
546 Some(scale) => Bound::Scaled { unscaled: signed, scale },
547 None => Bound::Int(signed),
548 };
549 self.value(bound, self.fixed.unwrap_or(8));
550 continue;
551 }
552 if let Some(bytes) = vector.bytes_at(row) {
553 self.bytes_value(bytes);
554 continue;
555 }
556 // row at a time: a float and a form neither typed accessor above can read have no slice
557 // to walk, so the value is built for this row and for no other.
558 let value = vector.value_at(row);
559 let width = self.fixed.unwrap_or_else(|| width(&value));
560 match Bound::of_value(&value) {
561 Some(bound) => self.value(bound, width),
562 None => {
563 // A non-null value with no ordered bound. Both ends go rather than the value
564 // being skipped, because an end computed from only the values that had bounds is
565 // an end that answers a MIN with a value the column does not hold.
566 self.bytes = self.bytes.saturating_add(width);
567 self.widest = self.widest.max(width);
568 self.bounded = false;
569 self.ascending = false;
570 self.descending = false;
571 }
572 }
573 }
574 }
575
576 /// One non-null value, as its bound and its width.
577 ///
578 /// Every end is compared before it is copied. The obvious way to write this is to hand the
579 /// bound to each end and let the end keep whichever is smaller, and that costs a clone a row per
580 /// end whether or not the row is one. For an integer that is four copies of a machine word and
581 /// hardly matters. For a string it is four allocations a row, and on SF1 `l_comment` that is
582 /// twenty four million of them for a column with two ends. Compared first, an end is copied once
583 /// on a sorted column and about log n times on a shuffled one.
584 fn value(&mut self, bound: Bound, width: u64) {
585 self.measure(width);
586 let ordering = self.previous.as_ref().map(|previous| previous.order(&bound));
587 if ordering.is_none() {
588 self.first = Some(bound.clone());
589 }
590 self.run(ordering);
591 if takes(&self.low, &bound, Ordering::Less) {
592 self.low = Some(bound.clone());
593 }
594 if takes(&self.high, &bound, Ordering::Greater) {
595 self.high = Some(bound.clone());
596 }
597 if takes(&self.stripe_low, &bound, Ordering::Less) {
598 self.stripe_low = Some(bound.clone());
599 }
600 if takes(&self.stripe_high, &bound, Ordering::Greater) {
601 self.stripe_high = Some(bound.clone());
602 }
603 self.previous = Some(bound);
604 }
605
606 /// The same for a byte string, without a `Vec` a row.
607 ///
608 /// A string column is where the pass above still allocates, because the bound it is handed had
609 /// to be built out of the slice before it could be compared to anything, and the row it keeps as
610 /// the previous one is a new `Vec` every row whether or not any end moved. Here nothing is built
611 /// to be compared, and the buffer the previous row owns is refilled rather than replaced, which
612 /// is an allocation on the first row of the column and none after it.
613 ///
614 /// This is the difference between statistics costing a tenth of the write and costing as much as
615 /// it. At SF1, `lineitem`'s five string columns took nineteen of the pass's twenty eight seconds
616 /// before this and its eleven numeric columns took the other nine.
617 fn bytes_value(&mut self, bytes: &[u8]) {
618 self.measure(bytes.len() as u64);
619 let ordering = match &self.previous {
620 None => {
621 self.first = Some(Bound::Bytes(bytes.to_vec()));
622 None
623 }
624 Some(Bound::Bytes(previous)) => Some(Some(previous.as_slice().cmp(bytes))),
625 // A bound of another domain in a byte column, which a column of one type cannot hold.
626 Some(_) => Some(None),
627 };
628 self.run(ordering);
629 if takes_bytes(&self.low, bytes, Ordering::Less) {
630 fill(&mut self.low, bytes);
631 }
632 if takes_bytes(&self.high, bytes, Ordering::Greater) {
633 fill(&mut self.high, bytes);
634 }
635 if takes_bytes(&self.stripe_low, bytes, Ordering::Less) {
636 fill(&mut self.stripe_low, bytes);
637 }
638 if takes_bytes(&self.stripe_high, bytes, Ordering::Greater) {
639 fill(&mut self.stripe_high, bytes);
640 }
641 fill(&mut self.previous, bytes);
642 }
643
644 /// What one value costs, which is the byte total and the widest of them.
645 fn measure(&mut self, width: u64) {
646 self.bytes = self.bytes.saturating_add(width);
647 self.widest = self.widest.max(width);
648 }
649
650 /// What this value standing above, below or level with the one before it does to the order flags.
651 ///
652 /// The outer `None` is the first value of the column. The inner one is a pair this build cannot
653 /// order, which a column of one type cannot produce and which costs an order claim rather than
654 /// being assumed away.
655 fn run(&mut self, ordering: Option<Option<Ordering>>) {
656 match ordering {
657 None => self.runs = 1,
658 Some(Some(Ordering::Less)) => self.descending = false,
659 Some(Some(Ordering::Greater)) => {
660 self.ascending = false;
661 self.runs += 1;
662 }
663 Some(Some(Ordering::Equal)) => {}
664 Some(None) => {
665 self.ascending = false;
666 self.descending = false;
667 }
668 }
669 }
670
671 /// Starts a stripe, which is `key` in the order the table will be read in.
672 fn open_stripe(&mut self, key: (u64, u64)) {
673 self.stripe_low = None;
674 self.stripe_high = None;
675 self.key = key;
676 self.first = None;
677 self.previous = None;
678 self.ascending = true;
679 self.descending = true;
680 self.runs = 0;
681 }
682
683 fn close_stripe(&mut self) {
684 // Both taken whatever happens, so that a stripe of nothing but nulls leaves neither end
685 // behind for the next stripe to be compared against.
686 if let (Some(low), Some(high)) = (self.stripe_low.take(), self.stripe_high.take()) {
687 self.stripes.push((low, high));
688 }
689 self.pieces.push(Piece {
690 key: self.key,
691 first: self.first.take(),
692 last: self.previous.take(),
693 ascending: self.ascending,
694 descending: self.descending,
695 runs: self.runs,
696 });
697 }
698
699 /// Takes in a pass that read whole stripes of the same column on its own. See [`Gather::absorb`].
700 ///
701 /// Only between stripes, which is the only place a pass is ever handed over: the fields that
702 /// describe the stripe being read are empty then on both sides.
703 fn absorb(&mut self, later: Pass) {
704 self.rows += later.rows;
705 self.nulls += later.nulls;
706 self.bounded &= later.bounded;
707 self.bytes = self.bytes.saturating_add(later.bytes);
708 self.widest = self.widest.max(later.widest);
709 if let Some(low) = later.low {
710 if takes(&self.low, &low, Ordering::Less) {
711 self.low = Some(low);
712 }
713 }
714 if let Some(high) = later.high {
715 if takes(&self.high, &high, Ordering::Greater) {
716 self.high = Some(high);
717 }
718 }
719 self.stripes.extend(later.stripes);
720 self.pieces.extend(later.pieces);
721 }
722
723 /// Puts the stripes' order fields together in the order the table is read in.
724 ///
725 /// A pass that never opened a stripe has nothing here and keeps what it counted as it went.
726 /// Otherwise the pieces are laid end to end by key: each one's own flags hold, and the seam
727 /// between two is one comparison of the last value of the first against the first value of the
728 /// second, which is the comparison the pass would have made had the rows come in that order.
729 /// Every piece that held a value started its run count at one, so a seam that is not a descent
730 /// joins two runs into one and gives one back.
731 fn settle(&mut self) {
732 if self.pieces.is_empty() {
733 return;
734 }
735 let mut pieces = std::mem::take(&mut self.pieces);
736 pieces.sort_by_key(|piece| piece.key);
737 let (mut ascending, mut descending, mut runs) = (true, true, 0_u64);
738 let mut previous: Option<Bound> = None;
739 for piece in pieces {
740 ascending &= piece.ascending;
741 descending &= piece.descending;
742 let (Some(first), Some(last)) = (piece.first, piece.last) else { continue };
743 runs += piece.runs;
744 if let Some(previous) = &previous {
745 match previous.order(&first) {
746 Some(Ordering::Less) => descending = false,
747 Some(Ordering::Greater) => ascending = false,
748 Some(Ordering::Equal) => {}
749 None => {
750 ascending = false;
751 descending = false;
752 }
753 }
754 if previous.order(&first) != Some(Ordering::Greater) {
755 runs = runs.saturating_sub(1);
756 }
757 }
758 previous = Some(last);
759 }
760 self.ascending = ascending;
761 self.descending = descending;
762 self.runs = runs;
763 }
764
765 fn finish(mut self, sketch: Sketch, stripes: Vec<Sketch>) -> Stats {
766 self.settle();
767 let present = self.rows - self.nulls;
768 // The one rule the module doc names. An exact distinct count is one the sketch never had to
769 // throw a value away to keep, and everything downstream of the count follows from this
770 // answer rather than from what the writer hoped.
771 let exact = sketch.is_exact();
772 let distinct = if exact {
773 sketch.len() as u64
774 } else {
775 // Rounded rather than truncated, and clamped under the rows it cannot exceed. An
776 // estimate above the row count is arithmetically possible and is always wrong, and a
777 // planner that sees one concludes a column has more distinct values than rows.
778 #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
779 let estimate = sketch.distinct().round().max(0.0) as u64;
780 estimate.min(present)
781 };
782 let summary = Summary {
783 rows: self.rows,
784 nulls: self.nulls,
785 low: if self.bounded { self.low } else { None },
786 high: if self.bounded { self.high } else { None },
787 // Every end here came from a value the column holds, because this pass read them all.
788 // That is the whole difference between a summary and a zone map, which is allowed to be
789 // wider than its column and so can only skip and never answer.
790 ends_exact: self.bounded,
791 distinct,
792 // Exact or estimated, and never certified. A KMV sketch's relative error is about one
793 // over the square root of k, which is a standard error and not a bound, and Certified
794 // in this codebase means a bound that holds. Calling a one and a half percent standard
795 // error a guarantee is how an estimate gets treated as an answer.
796 distinct_class: if exact { Class::Exact } else { Class::Estimated },
797 // Only from an exact count. A sketch that overflowed cannot tell a column of a million
798 // unique values from one where two of them repeat, and uniqueness is the claim a key
799 // map is built on.
800 unique: exact && distinct == present,
801 order: if present == 0 {
802 Order::Neither
803 } else if self.ascending {
804 Order::Ascending
805 } else if self.descending {
806 Order::Descending
807 } else {
808 Order::Neither
809 },
810 runs: self.runs,
811 overlapping: overlapping(&self.stripes),
812 bytes: self.bytes,
813 widest: self.widest,
814 newest: self.generation,
815 };
816 // `new` rather than `merged` even for the empty case, because the two differ only in
817 // whether the list is checked and an empty list passes. A stripe sketch that is not at
818 // STRIPE_K is a bug in this file and is worth hearing about here rather than at the read.
819 let sketches = match Sketches::new(sketch.clone(), stripes) {
820 Ok(sketches) => sketches,
821 // Unreachable, since every stripe sketch above came out of `narrowed(STRIPE_K)` and a
822 // table cannot hold a million stripes. The merged sketch alone is the answer anyway:
823 // per stripe sketches are an optimization over a summary that is complete without
824 // them, so losing them costs a skipped stripe and never an answer.
825 Err(_) => Sketches::merged(sketch),
826 };
827 Stats { summary, sketches }
828 }
829}
830
831/// Whether an end has to become this bound, which is the question that replaces a clone.
832///
833/// `want` is [`Ordering::Less`] for a low end and [`Ordering::Greater`] for a high one. An end that
834/// is not there yet takes any value. A pair this build cannot order leaves the end alone, which is
835/// what [`Bound::smaller`] does across domains and which a column of one type cannot reach anyway.
836/// A dictionary the pass has read, and the positions and widths its codes stand for.
837///
838/// Kept from one vector to the next, and kept by the identity of the values it was read from rather
839/// than by a guess about what the caller is doing. One Parquet dictionary page serves every data
840/// page of its column chunk, so a load hands over a hundred vectors that share a dictionary, and
841/// reading it once instead of a hundred times is most of what this arm is worth. The `Arc` is held
842/// rather than its address noted, because a freed allocation's address is one a later dictionary can
843/// be handed and a cache keyed on that would read the wrong values and never know.
844#[derive(Debug)]
845struct Coded {
846 values: Arc<Vector>,
847 /// Position and width per code, `None` for a code whose entry is null.
848 codes: Vec<Option<(u32, u64)>>,
849 /// The distinct bounds in ascending order, which is what a position indexes.
850 bounds: Vec<Bound>,
851}
852
853impl Coded {
854 /// One vector of codes, reduced against this dictionary.
855 ///
856 /// The loop this whole arm is for. A code lookup, a bounds check and an integer comparison, for
857 /// a column whose row at a time path was comparing byte strings.
858 fn reduce(&self, codes: &[u32], rows: usize, validity: &Validity) -> Reduced {
859 let nullable = validity.has_nulls(rows);
860 let mut out = Reduced::empty(rows as u64);
861 let (mut low, mut high, mut first, mut last) = (0_u32, 0_u32, 0_u32, 0_u32);
862 for (row, &code) in codes.iter().take(rows).enumerate() {
863 let entry = if nullable && !validity.is_valid(row) {
864 None
865 } else {
866 self.codes.get(code as usize).copied().flatten()
867 };
868 let Some((position, width)) = entry else {
869 out.nulls += 1;
870 continue;
871 };
872 out.bytes = out.bytes.saturating_add(width);
873 out.widest = out.widest.max(width);
874 if out.values == 0 {
875 low = position;
876 high = position;
877 first = position;
878 } else if position < last {
879 out.descents += 1;
880 } else if position > last {
881 out.ascents += 1;
882 }
883 low = low.min(position);
884 high = high.max(position);
885 last = position;
886 out.values += 1;
887 }
888 let at = |position: u32| self.bounds[position as usize].clone();
889 out.ends = (out.values > 0).then(|| Ends {
890 low: at(low),
891 high: at(high),
892 first: at(first),
893 last: at(last),
894 });
895 out
896 }
897}
898
899/// What one vector came to, in the terms the pass folds rather than in the terms it was read in.
900///
901/// The two fast arms read a vector very differently and reduce it to the same nine numbers, so the
902/// folding is written once. Everything here is about the vector alone: nothing in it depends on the
903/// vector before, which is the half [`Pass::fold`] settles.
904#[derive(Debug)]
905struct Reduced {
906 rows: u64,
907 nulls: u64,
908 /// Non-null values, which is what says whether `ends` means anything.
909 values: u64,
910 bytes: u64,
911 widest: u64,
912 /// Adjacent non-null pairs where the later value is the larger, which rules out a descending
913 /// column, and where it is the smaller, which starts a run.
914 ascents: u64,
915 descents: u64,
916 ends: Option<Ends>,
917}
918
919impl Reduced {
920 fn empty(rows: u64) -> Self {
921 Self { rows, nulls: 0, values: 0, bytes: 0, widest: 0, ascents: 0, descents: 0, ends: None }
922 }
923}
924
925/// The four values of a vector the pass needs by name: its two ends, and its two edges.
926#[derive(Debug)]
927struct Ends {
928 low: Bound,
929 high: Bound,
930 /// The first and last non-null values, for joining to the vectors either side.
931 first: Bound,
932 last: Bound,
933}
934
935/// The bound of a dictionary entry that [`Pass::scan_dictionary`] has already found is not null.
936fn bound_of(entries: &[Option<(Bound, u64)>], at: usize) -> &Bound {
937 match &entries[at] {
938 Some((bound, _)) => bound,
939 // Unreachable: every index handed here came out of the filter that dropped the nulls. The
940 // low bound is the answer that costs a wider range rather than a wrong one, if it ever is.
941 None => &Bound::Int(i128::MIN),
942 }
943}
944
945/// What one vector of a flat signed column came to, computed without building a single [`Bound`].
946///
947/// Everything a [`Pass`] needs from a vector that is not about the vector before it. The two ends,
948/// the two rows at the edges so that the joining comparison can be made, and the counts.
949#[derive(Debug)]
950struct Spread {
951 /// Rows in the vector, nulls included.
952 rows: u64,
953 nulls: u64,
954 /// The ends, meaningless when `values` is zero.
955 low: i128,
956 high: i128,
957 /// The first and last non-null values, for joining to the vectors either side.
958 first: i128,
959 last: i128,
960 /// Adjacent non-null pairs where the later value is the smaller, which is what starts a run.
961 descents: u64,
962 /// And where it is the larger, which is what rules out a descending column.
963 ascents: u64,
964 /// Non-null values, which is what the byte total is a multiple of.
965 values: u64,
966}
967
968/// One pass over a vector's non-null values, reading them through `get`.
969///
970/// Generic over the reader rather than over the element type, so that the caller can widen a layout
971/// into an `i128` at the call site and this gets compiled once per layout with the widening inlined.
972fn spread(rows: usize, validity: &Validity, get: impl Fn(usize) -> i128) -> Spread {
973 let mut out = Spread {
974 rows: rows as u64,
975 nulls: 0,
976 low: 0,
977 high: 0,
978 first: 0,
979 last: 0,
980 descents: 0,
981 ascents: 0,
982 values: 0,
983 };
984 let nullable = validity.has_nulls(rows);
985 // row at a time: this is the loop the whole fast path is, and it is a row at a time because the
986 // ascents and the descents are about adjacent rows. No `Value` is built here and none can be:
987 // `get` hands back an `i128` read out of a typed slice.
988 for row in 0..rows {
989 if nullable && !validity.is_valid(row) {
990 out.nulls += 1;
991 continue;
992 }
993 let value = get(row);
994 if out.values == 0 {
995 out.low = value;
996 out.high = value;
997 out.first = value;
998 } else {
999 if value < out.last {
1000 out.descents += 1;
1001 } else if value > out.last {
1002 out.ascents += 1;
1003 }
1004 out.low = out.low.min(value);
1005 out.high = out.high.max(value);
1006 }
1007 out.last = value;
1008 out.values += 1;
1009 }
1010 out
1011}
1012
1013fn takes(held: &Option<Bound>, bound: &Bound, want: Ordering) -> bool {
1014 match held {
1015 None => true,
1016 Some(held) => bound.order(held) == Some(want),
1017 }
1018}
1019
1020/// The same question asked of a slice, so that nothing is built to ask it.
1021fn takes_bytes(held: &Option<Bound>, bytes: &[u8], want: Ordering) -> bool {
1022 match held {
1023 None => true,
1024 Some(Bound::Bytes(held)) => bytes.cmp(held.as_slice()) == want,
1025 Some(_) => false,
1026 }
1027}
1028
1029/// Puts these bytes in an end, reusing the buffer that is already there.
1030///
1031/// The whole of the byte path's advantage. A `Vec` that is cleared and refilled does not allocate
1032/// once it is wide enough, and these ends plus the previous row are where every allocation of the
1033/// value path went.
1034fn fill(held: &mut Option<Bound>, bytes: &[u8]) {
1035 match held {
1036 Some(Bound::Bytes(held)) => {
1037 held.clear();
1038 held.extend_from_slice(bytes);
1039 }
1040 held => *held = Some(Bound::Bytes(bytes.to_vec())),
1041 }
1042}
1043
1044/// Whether any two of these stripe ranges overlap.
1045///
1046/// Sorted by low end and then walked, so this is one sort rather than the square. A pair this cannot
1047/// order counts as overlapping, which is the answer that costs a skipped stripe rather than a wrong
1048/// one.
1049fn overlapping(stripes: &[(Bound, Bound)]) -> bool {
1050 let mut order = (0..stripes.len()).collect::<Vec<_>>();
1051 order
1052 .sort_by(|&one, &other| stripes[one].0.order(&stripes[other].0).unwrap_or(Ordering::Equal));
1053 order.windows(2).any(|pair| {
1054 let before = &stripes[pair[0]].1;
1055 let after = &stripes[pair[1]].0;
1056 before.order(after) != Some(Ordering::Less)
1057 })
1058}
1059
1060/// What every value of this type takes, when they all take the same.
1061///
1062/// `None` for the variable width types, which is the two string ones and nothing else. Read off the
1063/// type once by `Pass::new` rather than off each value.
1064fn fixed_width(ty: &LogicalType) -> Option<u64> {
1065 Some(match ty {
1066 LogicalType::Boolean | LogicalType::TinyInt | LogicalType::UTinyInt => 1,
1067 LogicalType::SmallInt | LogicalType::USmallInt => 2,
1068 LogicalType::Integer | LogicalType::UInteger | LogicalType::Float | LogicalType::Date => 4,
1069 LogicalType::HugeInt | LogicalType::UHugeInt | LogicalType::Decimal { .. } => 16,
1070 LogicalType::Varchar | LogicalType::Blob => return None,
1071 // The eight byte types: the two big integers, the double, and the four time ones. Anything
1072 // else that reaches here is refused a summary by `countable` long before this.
1073 _ => 8,
1074 })
1075}
1076
1077/// What one value takes, for the byte total and the widest value.
1078///
1079/// The logical width and not the stored one. The stored width is what the column's encoding chose
1080/// and is already in the layout; this is what the value costs a plan that has to materialize it,
1081/// which is the number a hash table sizing decision wants.
1082fn width(value: &Value) -> u64 {
1083 match value {
1084 Value::Null => 0,
1085 Value::Boolean(_) | Value::TinyInt(_) | Value::UTinyInt(_) => 1,
1086 Value::SmallInt(_) | Value::USmallInt(_) => 2,
1087 Value::Integer(_) | Value::UInteger(_) | Value::Float(_) | Value::Date(_) => 4,
1088 Value::HugeInt(_) | Value::UHugeInt(_) | Value::Decimal { .. } => 16,
1089 Value::Varchar(text) => text.len() as u64,
1090 Value::Blob(bytes) => bytes.len() as u64,
1091 // The eight byte types and anything else, which is every remaining scalar. A nested value
1092 // reaching here would be counted at eight and is refused a summary long before this by
1093 // `countable`.
1094 _ => 8,
1095 }
1096}
1097
1098/// One column's statistics built as the rows go past on their way into the file.
1099///
1100/// # Why this exists beside [`build_summary`]
1101///
1102/// Section 3.7 gives the build ten percent of the native write time, and [`build_summary`] cannot
1103/// fit inside that however tight its inner loop gets, because it starts by reading the file back. A
1104/// second full read of a committed table, decode included, is not ten percent of the first one. It
1105/// is most of it: on a TPC-H SF1 `lineitem` the standalone build is 11.4 seconds against a write of
1106/// 20.0 seconds of processor time, and the read is the bulk of the 11.4.
1107///
1108/// The writer has the vectors already. It buffers a stripe as chunks and hands one column of all of
1109/// them to each encode worker, so every value is in memory, in `rid` order, on a thread that is
1110/// about to walk it anyway. What is left of the build once the read is taken out is the hashing and
1111/// the comparisons, and those do fit. So this is the same [`Pass`] and the same [`Counts`] driven
1112/// from the write rather than from a reader, and [`build_summary`] stays as the path for a file
1113/// that was written before any of this existed.
1114///
1115/// # No per stripe sketches here
1116///
1117/// Section 3.8 promotes a column when something has declared a relationship or a key over it, and
1118/// [`read_columns`] reads that off the file. A table being written for the first time has no
1119/// sections at all, so the promoted set is empty by construction and there is nothing for this to
1120/// decide. A later checkpoint that declares a key is what promotes the column, and that goes through
1121/// [`build_stats_for`] with the file in front of it.
1122#[derive(Debug)]
1123pub(crate) struct Gather {
1124 pass: Pass,
1125 counts: Counts,
1126}
1127
1128impl Gather {
1129 /// One for a column that can be summarized, and nothing for one that cannot.
1130 ///
1131 /// `None` rather than an error, because a table with an interval column in it still gets
1132 /// summaries for its other fifteen and section 3.1 says the interval column plans the way it
1133 /// planned before.
1134 pub(crate) fn new(ty: &LogicalType, generation: u64) -> Option<Self> {
1135 countable(ty).then(|| Self { pass: Pass::new(ty, generation), counts: Counts::new(1) })
1136 }
1137
1138 /// Folds one whole stripe of this column, in part order.
1139 ///
1140 /// A stripe at a time and not a part at a time, because the stripe is the unit the pass opens
1141 /// and closes its ends over and a caller that fed it parts would have to know that. The key is
1142 /// where the stripe goes once the writer sorts its stripes, which need not be the order they
1143 /// reach this in.
1144 pub(crate) fn stripe<'a>(&mut self, key: (u64, u64), parts: impl Iterator<Item = &'a Vector>) {
1145 self.pass.open_stripe(key);
1146 for vector in parts {
1147 self.counts.add_column(0, vector);
1148 self.pass.scan(vector);
1149 }
1150 self.pass.close_stripe();
1151 }
1152
1153 /// Takes in a gather that folded stripes of the same column on its own, as though this had
1154 /// folded them.
1155 ///
1156 /// This is what lets a stripe be summarized on the thread that encodes it, before the writer's
1157 /// lock is taken. Nothing a stripe adds depends on the stripes before it: the order fields are
1158 /// kept a stripe at a time and put together by key at the end, the ends and the totals are a
1159 /// minimum, a maximum and sums, and the counts union. The one thing that does depend on order
1160 /// is the tally's list, which comes out in the order the stripes are absorbed in, and that is
1161 /// the order they reached the writer in, which is what it was before.
1162 pub(crate) fn absorb(&mut self, later: Gather) {
1163 self.pass.absorb(later.pass);
1164 self.counts.absorb(later.counts);
1165 }
1166
1167 /// How many rows went past, which is what the caller checks against the table's own count.
1168 pub(crate) fn rows(&self) -> u64 {
1169 self.pass.rows
1170 }
1171
1172 /// The summary and the merged sketch, or nothing if the column turned out to be blind.
1173 ///
1174 /// Blind means a form `rudb_storage::count` has no arm for turned up, so the sketch is missing
1175 /// rows and cannot say which. A distinct count that is too low is the one error an estimator has
1176 /// no defence against, so the column gets no sections rather than sections with a number in them
1177 /// nothing can check.
1178 pub(crate) fn finish(self) -> Option<Stats> {
1179 let sketch = self.counts.sketch(0)?;
1180 Some(self.pass.finish(sketch, Vec::new()))
1181 }
1182}
1183
1184/// Everything the columns of a table being written cost so far, which is what the budget is a share
1185/// of.
1186///
1187/// The same sum [`crate::Layout::columns_total`] takes, off the table rather than off a reader,
1188/// because the writer has no reader and the file it would open is not committed yet. Every stripe's
1189/// pages are written by the time this is asked and so are the dictionaries, so the two agree.
1190pub(crate) fn column_bytes(table: &crate::Table) -> u64 {
1191 (0..table.fields.len())
1192 .map(|at| {
1193 crate::sum(table.stripes.iter().map(|stripe| crate::span_bytes(&stripe.pages, at)))
1194 .saturating_add(crate::sum(
1195 table.stripes.iter().map(|stripe| stripe.memberships.bytes(at)),
1196 ))
1197 .saturating_add(crate::sum(
1198 table.stripes.iter().map(|stripe| stripe.sieves.bytes(at)),
1199 ))
1200 .saturating_add(crate::sum(
1201 table.stripes.iter().map(|stripe| stripe.part_ranges.bytes(at)),
1202 ))
1203 .saturating_add(crate::dictionary_bytes(table, at))
1204 })
1205 .fold(0, u64::saturating_add)
1206}
1207
1208/// Which of these payloads fit the allowance, smallest first.
1209///
1210/// Smallest first so that a budget that cannot hold everything holds as many columns as it can. The
1211/// alternative is column order, which would give the summaries to whichever columns the schema
1212/// happened to list early, and there is nothing about being the first column that makes a summary
1213/// worth more.
1214pub(crate) fn within(costs: &[usize], allowance: u64, spent: u64) -> Vec<bool> {
1215 let mut order = (0..costs.len()).collect::<Vec<_>>();
1216 order.sort_by_key(|&at| costs[at]);
1217 let mut spent = spent;
1218 let mut keep = vec![false; costs.len()];
1219 for at in order {
1220 let cost = costs[at] as u64;
1221 if spent.saturating_add(cost) <= allowance {
1222 spent += cost;
1223 keep[at] = true;
1224 }
1225 }
1226 keep
1227}
1228
1229/// What the allowance is for a table whose columns come to this many bytes.
1230pub(crate) fn allowance(column_bytes: u64, share: u64) -> u64 {
1231 (column_bytes.saturating_mul(share) / 100).max(BUDGET_FLOOR)
1232}
1233
1234/// Builds the statistics for each of these columns and attaches them all in one commit.
1235///
1236/// One commit and not one each, for the reason `graph::build_key_maps` gives: a checkpoint that
1237/// published one generation per column would be one chance per column of being interrupted halfway.
1238///
1239/// # Errors
1240///
1241/// If the file cannot be opened, a column cannot be summarized, or the attach fails.
1242pub fn build_stats(path: &Path, table: &str, columns: &[usize]) -> Result<Vec<Built>> {
1243 build_stats_within(path, table, columns, BUDGET_SHARE)
1244}
1245
1246/// The columns of this table the per stripe rule promotes, in column order.
1247///
1248/// Section 3.8's default set: the columns something has declared a relationship or a key over. What
1249/// this build has to go on for that is the file itself, so the answer is the columns that already
1250/// carry a graph section, which is a key map or a forward link. That is not a proxy for the
1251/// question, it is the same question asked of the only party that has been told the answer: a key
1252/// map exists on a column because something declared it a key.
1253///
1254/// Empty is the ordinary answer and it is the right one. A table nothing has declared anything over
1255/// gets table level summaries and no per stripe sketches, which is what section 3.8 says and what
1256/// keeps SF100 inside two percent.
1257///
1258/// The other source the spec names is document 06's observation log, which promotes a column that
1259/// queries turned out to read at the next checkpoint. It is not built yet. When it is, it adds
1260/// columns here and nothing else in this file changes.
1261#[must_use]
1262pub fn read_columns(reader: &Reader) -> Vec<usize> {
1263 let generation = reader.table().generation();
1264 let mut promoted = reader
1265 .table()
1266 .sections()
1267 .iter()
1268 .filter(|held| held.among(section::GRAPH_KINDS) && held.usable(generation))
1269 .filter_map(|held| usize::try_from(held.id).ok())
1270 .collect::<Vec<_>>();
1271 promoted.sort_unstable();
1272 promoted.dedup();
1273 promoted
1274}
1275
1276/// The same, against a budget of `share` percent of the table's stored column bytes.
1277///
1278/// The budget is over the table rather than over a column, and when it binds the cheapest columns
1279/// are admitted first. That is the same degenerate case section 3.7's expected value ordering has
1280/// for a key map with no relationship over it: nothing has said which column a plan will ask about,
1281/// so no summary is worth more than another and the ordering falls back to the denominator. Cheapest
1282/// first is also the order that fits the most summaries in the room there is.
1283///
1284/// A column is all or nothing. Its summary and its sketches are admitted together or neither is,
1285/// because a summary whose distinct count came from a sketch that was then dropped is a number with
1286/// nothing behind it to check it against.
1287///
1288/// # Errors
1289///
1290/// If the file cannot be opened, a column cannot be summarized, or the attach fails.
1291pub fn build_stats_within(
1292 path: &Path,
1293 table: &str,
1294 columns: &[usize],
1295 share: u64,
1296) -> Result<Vec<Built>> {
1297 let promoted = read_columns(&Catalog::open(path)?.table(table)?);
1298 build_stats_for(path, table, columns, &promoted, share)
1299}
1300
1301/// The same, with the per stripe set named rather than read off the file.
1302///
1303/// For a caller that knows something this build does not, which today is the measurement harness and
1304/// tomorrow is whatever reads document 06's observation log. [`build_stats_within`] is the ordinary
1305/// entry point and it asks [`read_columns`].
1306///
1307/// A column in `per_stripe` that is not in `columns` is ignored rather than refused, because the two
1308/// lists answer different questions and a caller that names a promoted column it is not building is
1309/// not making a mistake worth stopping for.
1310///
1311/// # Errors
1312///
1313/// If the file cannot be opened, a column cannot be summarized, or the attach fails.
1314pub fn build_stats_for(
1315 path: &Path,
1316 table: &str,
1317 columns: &[usize],
1318 per_stripe: &[usize],
1319 share: u64,
1320) -> Result<Vec<Built>> {
1321 let reader = Catalog::open(path)?.table(table)?;
1322 let column_bytes = reader.layout().columns_total();
1323 let allowance = allowance(column_bytes, share);
1324 let spent = held_bytes(&reader, columns)?;
1325 let mut report = Vec::with_capacity(columns.len());
1326 let mut payloads = Vec::with_capacity(columns.len());
1327 for &column in columns {
1328 let start = Instant::now();
1329 let stats = build_summary_for(&reader, column, per_stripe.contains(&column))?;
1330 let mut summary = Vec::new();
1331 stats.summary.encode(&mut summary)?;
1332 let mut sketches = Vec::new();
1333 stats.sketches.encode(&mut sketches)?;
1334 report.push(Built {
1335 column,
1336 rows: stats.summary.rows,
1337 distinct: stats.summary.distinct,
1338 exact: stats.summary.distinct_class == Class::Exact,
1339 order: stats.summary.order,
1340 summary_bytes: summary.len(),
1341 sketch_bytes: sketches.len(),
1342 stripes: stats.sketches.stripes.len(),
1343 column_bytes,
1344 built: false,
1345 build: start.elapsed(),
1346 });
1347 payloads.push((column, summary, sketches));
1348 }
1349 let costs = report.iter().map(Built::bytes).collect::<Vec<_>>();
1350 let keep = within(&costs, allowance, spent);
1351 for (one, &keep) in report.iter_mut().zip(&keep) {
1352 one.built = keep;
1353 }
1354 // The reader holds the file open and the attach opens it again to write, so it is dropped first
1355 // for the reason `graph` drops it: the moment the file is written is a moment nothing else in
1356 // this function is reading it.
1357 drop(reader);
1358 let mut attachments = Vec::with_capacity(payloads.len() * 2);
1359 for ((column, summary, sketches), _) in payloads.iter().zip(&keep).filter(|&(_, &keep)| keep) {
1360 let id = u64::try_from(*column).map_err(|_| invalid("column index overflow"))?;
1361 attachments.push(Attachment {
1362 kind: *section::SUMMARY,
1363 id,
1364 flags: 0,
1365 // A summary is a header the whole way down. There is nothing behind it that a reader
1366 // could decide not to read, which is the shape section 3.2's field is for and not a
1367 // misuse of it: the answer to "how much do I read to know what this says" is all of it.
1368 header_bytes: u32::try_from(summary.len())
1369 .map_err(|_| invalid("a summary longer than a u32 can count"))?,
1370 bytes: summary,
1371 });
1372 attachments.push(Attachment {
1373 kind: *section::SKETCHES,
1374 id,
1375 flags: 0,
1376 header_bytes: SKETCH_HEADER,
1377 bytes: sketches,
1378 });
1379 }
1380 crate::attach(path, table, &attachments)?;
1381 Ok(report)
1382}
1383
1384/// What the table's existing statistics sections cost, leaving out the ones this build is replacing.
1385///
1386/// Statistics sections only. The two percent of section 3.8 and the graph layer's ten percent are
1387/// separate shares of the same column bytes, and separate means each counts only what it owns. A
1388/// TPC-H SF10 file's key maps are 7.7 MB against a two percent allowance of 54 MB, so counting them
1389/// here would hand a seventh of the statistics budget to sections that already have one of their
1390/// own, and a table would lose summaries for a reason that has nothing to do with summaries.
1391///
1392/// Reading the extent tables is what this costs, which is one small read per section and not a read
1393/// of a payload. A section whose extent table does not checksum is counted as nothing, because it
1394/// is a section that is already not there.
1395fn held_bytes(reader: &Reader, replacing: &[usize]) -> Result<u64> {
1396 let mut total = 0;
1397 for held in reader.table().sections() {
1398 if !held.among(section::STATISTICS_KINDS) {
1399 continue;
1400 }
1401 let replaced = replacing.iter().any(|&column| u64::try_from(column) == Ok(held.id));
1402 if replaced || !held.usable(reader.table().generation()) {
1403 continue;
1404 }
1405 let Ok(extents) = reader.extents(held) else { continue };
1406 total += extents.iter().map(|extent| u64::from(extent.length)).sum::<u64>();
1407 }
1408 Ok(total)
1409}
1410
1411/// The summary this table carries for a column, when it carries one this build can use.
1412///
1413/// `None` covers every reason there is not one and covering them all is the point. Section 3.1 says
1414/// deleting every statistics section changes no answer, so there is no reason to distinguish *no
1415/// summary was built* from *the summary is stale*, *the payload does not checksum*, or *the layout
1416/// is one a later build invented*. The answer to all four is to plan the query the way it was
1417/// planned before summaries existed.
1418#[must_use]
1419pub fn summary(reader: &Reader, column: usize) -> Option<Summary> {
1420 let bytes = payload(reader, column, section::SUMMARY)?;
1421 Summary::decode(&bytes).ok()
1422}
1423
1424/// The sketches this table carries for a column, same.
1425///
1426/// One more reason for `None` here than above: a sketch built by a hash this build does not use is
1427/// declined by [`Sketches::decode`] rather than merged into anything, which costs a rebuild where
1428/// merging would cost an answer.
1429#[must_use]
1430pub fn sketches(reader: &Reader, column: usize) -> Option<Sketches> {
1431 let bytes = payload(reader, column, section::SKETCHES)?;
1432 Sketches::decode(&bytes).ok()
1433}
1434
1435fn payload(reader: &Reader, column: usize, kind: &[u8; 8]) -> Option<Vec<u8>> {
1436 let table = reader.table();
1437 let id = u64::try_from(column).ok()?;
1438 let held = table.sections().iter().find(|section| section.kind == *kind && section.id == id)?;
1439 if !held.usable(table.generation()) {
1440 return None;
1441 }
1442 reader.payload(held).ok()
1443}
1444
1445/// Whether a type can be summarized at all, which is whether it has a hash rule.
1446#[must_use]
1447pub fn summarizable(ty: &LogicalType) -> bool {
1448 countable(ty)
1449}
1450
1451#[cfg(test)]
1452mod tests {
1453 use std::fs;
1454 use std::path::PathBuf;
1455 use std::sync::Arc;
1456 use std::time::{SystemTime, UNIX_EPOCH};
1457
1458 use rudb_common::Field;
1459 use rudb_encoding::sketch::hash64;
1460 use rudb_storage::count::hash_value;
1461 use rudb_vector::{Chunk, Vector};
1462
1463 use super::*;
1464 use crate::Writer;
1465
1466 fn path(label: &str) -> PathBuf {
1467 let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
1468 std::env::temp_dir().join(format!("rudb-stats-{label}-{}-{stamp}.rdb", std::process::id()))
1469 }
1470
1471 /// A one column table of these values, written a thousand rows to a part.
1472 fn table_of(label: &str, values: &[Option<i64>]) -> PathBuf {
1473 let path = path(label);
1474 let mut writer =
1475 Writer::create(&path, "t", vec![Field::new("v", LogicalType::BigInt)]).expect("new");
1476 for part in values.chunks(1000) {
1477 let held =
1478 part.iter().map(|v| v.map_or(Value::Null, Value::BigInt)).collect::<Vec<_>>();
1479 let chunk =
1480 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("values")])
1481 .expect("one column");
1482 writer.append(&chunk).expect("a part");
1483 }
1484 writer.finish().expect("commit");
1485 path
1486 }
1487
1488 /// The same, with the part size named, for a test that needs more than one stripe.
1489 ///
1490 /// A stripe is up to `STRIPE_PARTS` parts, so small parts are how a test crosses a stripe
1491 /// boundary without writing a hundred and thirty thousand rows to do it.
1492 fn table_of_parts(label: &str, values: &[Option<i64>], per_part: usize) -> PathBuf {
1493 let path = path(label);
1494 let mut writer =
1495 Writer::create(&path, "t", vec![Field::new("v", LogicalType::BigInt)]).expect("new");
1496 for part in values.chunks(per_part) {
1497 let held =
1498 part.iter().map(|v| v.map_or(Value::Null, Value::BigInt)).collect::<Vec<_>>();
1499 let chunk =
1500 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("values")])
1501 .expect("one column");
1502 writer.append(&chunk).expect("a part");
1503 }
1504 writer.finish().expect("commit");
1505 path
1506 }
1507
1508 #[test]
1509 fn stripes_that_arrive_out_of_order_are_summarized_in_the_order_they_are_read() {
1510 // What a parallel load does: three pipeline instances each hand the writer a contiguous
1511 // run of the source as its own stripe, and they finish in whatever order they finish. The
1512 // table reads back sorted by source position, so that is the order the summary is about.
1513 // Every key repeats across a seam, the way an order's line items straddle two stripes.
1514 let path = path("late-stripes");
1515 let mut writer =
1516 Writer::create(&path, "t", vec![Field::new("v", LogicalType::BigInt)]).expect("new");
1517 let part = |from: i64| {
1518 let held = (from..from + 10).map(|v| Value::BigInt(v / 2)).collect::<Vec<_>>();
1519 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("values")])
1520 .expect("one column")
1521 };
1522 for stripe in [2_u64, 0, 1] {
1523 let parts = (0..3)
1524 .map(|at| {
1525 (
1526 (stripe * 3 + at, 0),
1527 part(i64::try_from(stripe * 30 + at * 10).expect("small")),
1528 )
1529 })
1530 .collect();
1531 writer.append_stripe(parts).expect("a stripe");
1532 }
1533 writer.finish().expect("commit");
1534
1535 let reader = reopen(&path);
1536 let summary = summary(&reader, 0).expect("the summary is in the file");
1537 assert_eq!(summary.rows, 90);
1538 assert_eq!(summary.order, Order::Ascending, "the stripes are in order once sorted");
1539 assert_eq!(summary.runs, 1, "and the seams between them are not descents");
1540 assert_eq!(crate::ascending(&reader), vec!["v".to_owned()]);
1541 }
1542
1543 /// The vector at a time pass says exactly what the row at a time pass says.
1544 ///
1545 /// [`Pass::scan_flat`] and [`Pass::scan_dictionary`] took the ordinary columns off
1546 /// [`Pass::scan_rows`] and they are why the build fits inside its share of the write. What they
1547 /// have to be is not fast but identical, so each is driven here over the same vectors in the
1548 /// same stripes as the row at a time pass and the two summaries are compared whole.
1549 ///
1550 /// Six shapes and five types. The shapes, because the fields that differ between them are the
1551 /// order flags and the run count, and those are what a vector at a time pass has to rejoin by
1552 /// hand. The types, because the two arms read a value in three different ways between them and
1553 /// a bound that came out of one has to be the bound that came out of another.
1554 #[test]
1555 fn the_vector_at_a_time_pass_says_what_the_row_at_a_time_pass_says() {
1556 // Coprime with the length, so this visits every value once and every part spans the range.
1557 let shuffled = (0..500_i64).map(|at| Some(1 + at * 307 % 500)).collect::<Vec<_>>();
1558 let shapes: [(&str, Vec<Option<i64>>); 7] = [
1559 ("ascending", (1..=500_i64).map(Some).collect()),
1560 ("descending", (1..=500_i64).rev().map(Some).collect()),
1561 ("constant", vec![Some(7); 500]),
1562 ("shuffled", shuffled),
1563 ("every third null", (1..=500_i64).map(|at| (at % 3 != 0).then_some(at)).collect()),
1564 ("all nulls", vec![None; 500]),
1565 ("twenty values over and over", (0..500_i64).map(|at| Some(at * 7 % 20)).collect()),
1566 ];
1567 let types = [
1568 LogicalType::SmallInt,
1569 LogicalType::Integer,
1570 LogicalType::BigInt,
1571 LogicalType::Decimal { width: 18, scale: 2 },
1572 LogicalType::Varchar,
1573 ];
1574 for (label, values) in &shapes {
1575 for ty in &types {
1576 // Sixty rows to a vector and five vectors to a stripe, so the stripe ends and the
1577 // overlap answer are in the comparison rather than left where they started.
1578 let held = values
1579 .chunks(60)
1580 .map(|part| {
1581 let values = part.iter().map(|value| one(ty, *value)).collect::<Vec<_>>();
1582 Vector::from_values(ty.clone(), &values).expect("values")
1583 })
1584 .collect::<Vec<_>>();
1585 // The same rows again as a dictionary of the twenty distinct values a vector holds,
1586 // in an order that is not the sorted one, so that the positions the arm hands out
1587 // are doing work rather than agreeing with the codes by accident.
1588 let coded = values
1589 .chunks(60)
1590 .map(|part| {
1591 let mut distinct = part.to_vec();
1592 distinct.sort_unstable();
1593 distinct.dedup();
1594 distinct.reverse();
1595 let values =
1596 distinct.iter().map(|value| one(ty, *value)).collect::<Vec<_>>();
1597 let codes = part
1598 .iter()
1599 .map(|value| {
1600 distinct.iter().position(|held| held == value).expect("a code")
1601 as u32
1602 })
1603 .collect::<Vec<_>>();
1604 Vector::dictionary(
1605 codes,
1606 Vector::from_values(ty.clone(), &values).expect("values"),
1607 )
1608 .expect("a dictionary")
1609 })
1610 .collect::<Vec<_>>();
1611 let flat = drive(ty, &held, |pass, vector| {
1612 assert!(pass.scan_flat(vector) || *ty == LogicalType::Varchar, "{label} {ty}");
1613 if *ty == LogicalType::Varchar {
1614 pass.scan_rows(vector);
1615 }
1616 });
1617 let dictionary = drive(ty, &coded, |pass, vector| {
1618 assert!(pass.scan_dictionary(vector), "{label} {ty} is dictionary coded");
1619 });
1620 let rows = drive(ty, &held, Pass::scan_rows);
1621 assert_eq!(flat.summary, rows.summary, "flat: {label} {ty}");
1622 assert_eq!(dictionary.summary, rows.summary, "dictionary: {label} {ty}");
1623 // And again over one dictionary that every vector shares, which is what a Parquet
1624 // load hands over and what the pass keeps its last dictionary for. Only for the
1625 // shapes narrow enough to have one, since a dictionary wider than the vector it
1626 // codes is one this arm turns down.
1627 let Some(shared) = shared(ty, values) else { continue };
1628 let coded = values
1629 .chunks(60)
1630 .map(|part| {
1631 let codes = part.iter().map(|value| code(values, *value)).collect();
1632 Vector::dictionary_over(codes, Arc::clone(&shared)).expect("a dictionary")
1633 })
1634 .collect::<Vec<_>>();
1635 let held = drive(ty, &coded, |pass, vector| {
1636 assert!(pass.scan_dictionary(vector), "{label} {ty} is dictionary coded");
1637 });
1638 assert_eq!(held.summary, rows.summary, "one dictionary: {label} {ty}");
1639 }
1640 }
1641 }
1642
1643 /// The distinct values of a column as one dictionary, or nothing if there are too many of them
1644 /// for [`Pass::scan_dictionary`] to take it.
1645 fn shared(ty: &LogicalType, values: &[Option<i64>]) -> Option<Arc<Vector>> {
1646 let mut distinct = values.to_vec();
1647 distinct.sort_unstable();
1648 distinct.dedup();
1649 // Wider than the sixty rows a vector holds is what the arm turns down, and a test that fed
1650 // it one would be asserting over the row at a time pass twice.
1651 if distinct.len() > 60 {
1652 return None;
1653 }
1654 // Reversed, so the positions the arm hands out are doing work rather than agreeing with the
1655 // codes by accident.
1656 distinct.reverse();
1657 let held = distinct.iter().map(|value| one(ty, *value)).collect::<Vec<_>>();
1658 Some(Arc::new(Vector::from_values(ty.clone(), &held).expect("values")))
1659 }
1660
1661 /// Where a value sits in the dictionary [`shared`] builds.
1662 fn code(values: &[Option<i64>], value: Option<i64>) -> u32 {
1663 let mut distinct = values.to_vec();
1664 distinct.sort_unstable();
1665 distinct.dedup();
1666 distinct.reverse();
1667 distinct.iter().position(|held| *held == value).expect("a code") as u32
1668 }
1669
1670 /// One value of this type, or a null, for the equivalence test above.
1671 fn one(ty: &LogicalType, value: Option<i64>) -> Value {
1672 let Some(value) = value else { return Value::Null };
1673 match ty {
1674 LogicalType::SmallInt => Value::SmallInt(value as i16),
1675 LogicalType::Integer => Value::Integer(value as i32),
1676 LogicalType::BigInt => Value::BigInt(value),
1677 LogicalType::Varchar => Value::Varchar(format!("v{value:04}")),
1678 _ => Value::Decimal { unscaled: i128::from(value), width: 18, scale: 2 },
1679 }
1680 }
1681
1682 /// A whole pass over these vectors, five to a stripe, read by whichever arm the caller names.
1683 fn drive(ty: &LogicalType, held: &[Vector], mut scan: impl FnMut(&mut Pass, &Vector)) -> Stats {
1684 let mut pass = Pass::new(ty, 1);
1685 for (at, stripe) in held.chunks(5).enumerate() {
1686 pass.open_stripe((at as u64, 0));
1687 for vector in stripe {
1688 scan(&mut pass, vector);
1689 }
1690 pass.close_stripe();
1691 }
1692 pass.finish(Sketch::of(&[]), Vec::new())
1693 }
1694
1695 /// A one column table of intervals, which is a type with no hash rule and so a table this
1696 /// build writes no statistics section for.
1697 ///
1698 /// The only way left to make a file whose table names no sections, now that an ordinary write
1699 /// writes them. See the criterion 3 test for why stamping the version back onto a file that has
1700 /// them does not do it.
1701 fn table_of_intervals(label: &str, months: &[i32]) -> PathBuf {
1702 let path = path(label);
1703 let mut writer =
1704 Writer::create(&path, "t", vec![Field::new("v", LogicalType::Interval)]).expect("new");
1705 for part in months.chunks(1000) {
1706 let held = part
1707 .iter()
1708 .map(|months| Value::Interval { months: *months, days: 0, micros: 0 })
1709 .collect::<Vec<_>>();
1710 let chunk = Chunk::new(vec![
1711 Vector::from_values(LogicalType::Interval, &held).expect("values"),
1712 ])
1713 .expect("one column");
1714 writer.append(&chunk).expect("a part");
1715 }
1716 writer.finish().expect("commit");
1717 path
1718 }
1719
1720 /// Every value of the one column, in rid order, which is what a scan of this table answers.
1721 fn rows_of(reader: &Reader) -> Vec<Value> {
1722 let mut out = Vec::new();
1723 for part in 0..reader.parts() {
1724 let chunk = reader.read(part, &[0]).expect("a part reads back");
1725 for row in 0..chunk.len() {
1726 out.push(chunk.value_at(0, row));
1727 }
1728 }
1729 out
1730 }
1731
1732 fn reopen(path: &PathBuf) -> Reader {
1733 Catalog::open(path).expect("reopen").table("t").expect("the table")
1734 }
1735
1736 #[test]
1737 fn a_summary_built_over_a_file_says_what_the_column_holds() {
1738 // End to end: the column goes to disk, comes back through the reader, and every field of
1739 // the summary is the truth about it. Three thousand rows so the scan crosses parts, because
1740 // a pass that read them in the wrong order would be right about one part and wrong about
1741 // the order fields for the rest.
1742 let values = (1..=3000_i64).map(Some).collect::<Vec<_>>();
1743 let path = table_of("sorted", &values);
1744 let built = build_stats(&path, "t", &[0]).expect("build");
1745 assert_eq!(built.len(), 1);
1746 assert!(built[0].built, "a one column table is nowhere near the budget");
1747 assert_eq!(built[0].rows, 3000);
1748 assert_eq!(built[0].distinct, 3000);
1749 assert!(built[0].exact, "three thousand values is under the default k");
1750 assert_eq!(built[0].order, Order::Ascending);
1751
1752 let reader = reopen(&path);
1753 let summary = summary(&reader, 0).expect("the summary is in the file");
1754 assert_eq!(summary.rows, 3000);
1755 assert_eq!(summary.nulls, 0);
1756 assert_eq!(summary.low, Some(Bound::Int(1)));
1757 assert_eq!(summary.high, Some(Bound::Int(3000)));
1758 assert!(summary.ends_exact);
1759 assert!(summary.unique, "a sorted run of distinct values is a key candidate");
1760 assert_eq!(summary.runs, 1, "one ascending run");
1761 assert_eq!(summary.distinct_class, Class::Exact);
1762 assert_eq!(summary.newest, reader.table().generation());
1763
1764 let sketches = sketches(&reader, 0).expect("the sketches are in the file");
1765 assert!(sketches.merged.is_exact());
1766 assert!(sketches.stripes.is_empty(), "the per stripe rule gives this column none");
1767
1768 fs::remove_file(&path).expect("clean up");
1769 }
1770
1771 #[test]
1772 fn nulls_are_counted_and_do_not_reach_the_ends_or_the_sketch() {
1773 // The distinction that costs an answer if it is got wrong. A null is a row and is not a
1774 // value, so it moves `rows` and `nulls` and moves nothing else.
1775 let values: Vec<Option<i64>> =
1776 (0..2000).map(|at| if at % 3 == 0 { None } else { Some(at) }).collect();
1777 let path = table_of("nulls", &values);
1778 build_stats(&path, "t", &[0]).expect("build");
1779
1780 let reader = reopen(&path);
1781 let summary = summary(&reader, 0).expect("the summary");
1782 let nulls = values.iter().filter(|v| v.is_none()).count() as u64;
1783 assert_eq!(summary.rows, 2000);
1784 assert_eq!(summary.nulls, nulls);
1785 assert_eq!(summary.present(), 2000 - nulls);
1786 assert_eq!(summary.distinct, 2000 - nulls, "a null is not a distinct value");
1787 assert_eq!(summary.low, Some(Bound::Int(1)), "zero is null here");
1788 assert!(summary.unique);
1789
1790 fs::remove_file(&path).expect("clean up");
1791 }
1792
1793 #[test]
1794 fn a_column_that_repeats_is_not_reported_unique_and_a_descending_one_is_seen() {
1795 let values = (0..2000_i64).map(|at| Some(-(at / 2))).collect::<Vec<_>>();
1796 let path = table_of("repeats", &values);
1797 build_stats(&path, "t", &[0]).expect("build");
1798
1799 let reader = reopen(&path);
1800 let summary = summary(&reader, 0).expect("the summary");
1801 assert_eq!(summary.distinct, 1000);
1802 assert!(!summary.unique, "every value appears twice");
1803 assert_eq!(summary.order, Order::Descending);
1804 assert_eq!(summary.runs, 1000, "a descending column is a run per distinct value");
1805
1806 fs::remove_file(&path).expect("clean up");
1807 }
1808
1809 #[test]
1810 fn a_column_past_the_default_k_is_estimated_and_says_so() {
1811 // The rule the module doc names, at the point where it bites. Past k the sketch threw values
1812 // away, so the count is an estimate, and the class has to say so or a COUNT(DISTINCT) is
1813 // answered out of metadata with a number that is close and wrong.
1814 let values = (0..20_000_i64).map(Some).collect::<Vec<_>>();
1815 let path = table_of("estimated", &values);
1816 let built = build_stats(&path, "t", &[0]).expect("build");
1817 assert!(!built[0].exact, "twenty thousand values is past the default k");
1818
1819 let reader = reopen(&path);
1820 let summary = summary(&reader, 0).expect("the summary");
1821 assert_eq!(summary.distinct_class, Class::Estimated);
1822 assert!(!summary.unique, "uniqueness is never claimed off an estimate");
1823 assert!(summary.distinct > 17_000 && summary.distinct <= 20_000, "{}", summary.distinct);
1824 assert!(summary.distinct <= summary.present(), "more distinct values than rows");
1825
1826 fs::remove_file(&path).expect("clean up");
1827 }
1828
1829 #[test]
1830 fn a_shuffled_column_is_neither_ordered_nor_one_run() {
1831 let values = (0..2000_i64).map(|at| Some((at * 7919) % 2000)).collect::<Vec<_>>();
1832 let path = table_of("shuffled", &values);
1833 build_stats(&path, "t", &[0]).expect("build");
1834
1835 let reader = reopen(&path);
1836 let summary = summary(&reader, 0).expect("the summary");
1837 assert_eq!(summary.order, Order::Neither);
1838 assert!(summary.runs > 100, "a shuffle is many runs, not one: {}", summary.runs);
1839 assert_eq!(summary.low, Some(Bound::Int(0)));
1840 assert_eq!(summary.high, Some(Bound::Int(1999)));
1841
1842 fs::remove_file(&path).expect("clean up");
1843 }
1844
1845 #[test]
1846 fn a_column_something_declared_a_key_over_is_sketched_per_stripe_and_a_plain_one_is_not() {
1847 // Section 3.8's rule, both halves of it. Nothing has declared anything over this column, so
1848 // the first build gives it the table level summary and no per stripe sketches, which is the
1849 // state most columns are in and is what keeps SF100 inside two percent. A key map is then
1850 // built over it, which is something declaring it a key, and the next build promotes it.
1851 let values = (1..=19_200_i64).map(Some).collect::<Vec<_>>();
1852 let path = table_of_parts("promoted", &values, 100);
1853
1854 let plain = build_stats(&path, "t", &[0]).expect("build");
1855 assert_eq!(plain[0].stripes, 0, "nothing has declared anything over this column yet");
1856
1857 crate::graph::build_key_maps(&path, "t", &[0]).expect("a key map declares it a key");
1858 let promoted = build_stats(&path, "t", &[0]).expect("rebuild");
1859 assert!(promoted[0].stripes > 1, "{} stripes, wanted more than one", promoted[0].stripes);
1860 assert!(promoted[0].built, "and they fit");
1861 // The equality rather than a tolerance. The merged sketch of a promoted column is the union
1862 // of its stripe sketches at the column's own k, and a union of bottom-k sketches at one k
1863 // is the bottom-k of everything they saw, so it holds the same hashes as the single sketch
1864 // the plain build made. Promotion changes where the counting is reset and nothing else.
1865 assert_eq!(promoted[0].distinct, plain[0].distinct, "the merged count did not move");
1866
1867 let reader = reopen(&path);
1868 let sketches = sketches(&reader, 0).expect("the sketches came back");
1869 assert_eq!(sketches.stripes.len(), promoted[0].stripes);
1870 assert!(
1871 sketches.stripes.iter().all(|stripe| stripe.k() == STRIPE_K),
1872 "a stripe sketch is written down at the smaller k"
1873 );
1874 let floor = sketches.floor(0, sketches.stripes.len()).expect("a floor over every stripe");
1875 let actual = 19_200.0;
1876 assert!(
1877 (floor - actual).abs() / actual < 0.25,
1878 "{floor:.0} over every stripe against {actual:.0}"
1879 );
1880
1881 drop(reader);
1882 fs::remove_file(&path).expect("clean up");
1883 }
1884
1885 #[test]
1886 fn a_file_from_before_the_section_table_opens_and_every_statistic_is_unknown() {
1887 // Exit criterion 3 of #762, the statistics half of it. A build that knows about summaries
1888 // opens a file written by a build that did not, with no rewrite and no repair, states
1889 // nothing about that file's columns, and reads back exactly what the same rows read back
1890 // out of a file this build wrote.
1891 //
1892 // `None` is what `Unknown` is at this layer, and the two readers answer it for every reason
1893 // there is rather than distinguishing them, which is section 3.1: there is nothing a caller
1894 // could do differently on hearing *the file predates statistics* rather than *the section
1895 // does not checksum*, because both are answered by planning the query the way it was
1896 // planned before statistics existed.
1897 //
1898 // The older file is a table of a type with no hash rule, with its version stamped back. A
1899 // build before section 3.8 wrote no section block at all, and a table this build writes no
1900 // sections for is that file on disk, so there is no fixture to go stale and no second
1901 // encoder to drift.
1902 //
1903 // The obvious construction, stamping the version back onto a file that does carry
1904 // summaries, does not work and is worth saying why. The section block is found by a magic
1905 // at the end of the directory rather than by the number in the header, so a stamped file
1906 // with sections in it is a file with sections in it, and the test would be asserting
1907 // nothing.
1908 let months = (1..=3000_i32).collect::<Vec<_>>();
1909 let older = table_of_intervals("before_sections", &months);
1910 let current = table_of("with_sections", &(1..=3000_i64).map(Some).collect::<Vec<_>>());
1911
1912 let file = fs::OpenOptions::new().write(true).open(&older).expect("reopen to patch");
1913 crate::write_at(&file, 8, &22_u32.to_le_bytes()).expect("stamp the older format");
1914 drop(file);
1915
1916 let new = reopen(¤t);
1917 assert!(summary(&new, 0).is_some(), "the file this build wrote says what it holds");
1918
1919 let old = reopen(&older);
1920 assert!(old.table().sections().is_empty(), "an older file names no sections");
1921 assert!(summary(&old, 0).is_none(), "and so says nothing about its columns");
1922 assert!(sketches(&old, 0).is_none());
1923 assert!(read_columns(&old).is_empty(), "nor promotes any of them");
1924 assert_eq!(old.table().rows(), 3000, "and reads every row it holds");
1925 assert_eq!(
1926 rows_of(&old).first(),
1927 Some(&Value::Interval { months: 1, days: 0, micros: 0 }),
1928 "with the values it was written with"
1929 );
1930
1931 drop(new);
1932 drop(old);
1933 fs::remove_file(¤t).expect("clean up");
1934 fs::remove_file(&older).expect("clean up");
1935 }
1936
1937 #[test]
1938 fn the_stripe_ends_say_whether_a_scan_can_skip_and_a_shuffle_says_it_cannot() {
1939 // The per stripe ends, which is the one thing the pass tracks that nothing else checks and
1940 // which a scan reads to skip a whole stripe. A sorted column's stripes do not overlap and a
1941 // shuffled column's every stripe spans the column, so the same rows in a different order
1942 // give the opposite answer. Three stripes, so that the ends are opened and closed more than
1943 // once and a pass that never reset them would be caught.
1944 let sorted = (1..=19_200_i64).map(Some).collect::<Vec<_>>();
1945 let ordered = table_of_parts("stripes_sorted", &sorted, 100);
1946 build_stats(&ordered, "t", &[0]).expect("build");
1947 let reader = reopen(&ordered);
1948 let ordered_summary = summary(&reader, 0).expect("the summary");
1949 assert!(!ordered_summary.overlapping, "a sorted column's stripes are disjoint");
1950 assert_eq!(ordered_summary.low, Some(Bound::Int(1)));
1951 assert_eq!(ordered_summary.high, Some(Bound::Int(19_200)));
1952 drop(reader);
1953
1954 // A fixed stride rather than a random shuffle, so a failure is the same failure twice. The
1955 // stride and the row count share no factor, so this visits every value exactly once and
1956 // every stripe ends up holding values from very nearly the whole range.
1957 let shuffled = (0..19_200_i64).map(|at| Some(1 + at * 7919 % 19_200)).collect::<Vec<_>>();
1958 let mixed = table_of_parts("stripes_shuffled", &shuffled, 100);
1959 build_stats(&mixed, "t", &[0]).expect("build");
1960 let reader = reopen(&mixed);
1961 let mixed_summary = summary(&reader, 0).expect("the summary");
1962 assert!(mixed_summary.overlapping, "a shuffled column's stripes all span it");
1963 assert_eq!(mixed_summary.low, Some(Bound::Int(1)), "the same values in a different order");
1964 assert_eq!(mixed_summary.high, Some(Bound::Int(19_200)));
1965 drop(reader);
1966
1967 fs::remove_file(&ordered).expect("clean up");
1968 fs::remove_file(&mixed).expect("clean up");
1969 }
1970
1971 #[test]
1972 fn the_graph_sections_do_not_count_against_the_statistics_budget() {
1973 // The direction of box 4 that costs more, because the two percent is the smaller share. A
1974 // TPC-H SF10 file's key maps are 7.7 MB against an allowance of 54 MB, so a statistics
1975 // build that counted them would start a seventh of the way through a budget it was given
1976 // all of, and columns at the far end of a wide table would go unsummarized for a reason
1977 // that has nothing to do with summaries.
1978 let values = (1..=3000_i64).map(Some).collect::<Vec<_>>();
1979 let path = table_of("apart", &values);
1980 crate::graph::build_key_maps(&path, "t", &[0]).expect("a key map first");
1981
1982 let reader = reopen(&path);
1983 let graph = reader
1984 .table()
1985 .sections()
1986 .iter()
1987 .filter(|held| held.among(section::GRAPH_KINDS))
1988 .count();
1989 assert_eq!(graph, 1, "the key map is in the file");
1990 assert_eq!(held_bytes(&reader, &[0]).expect("held"), 0, "and it is not the statistics'");
1991
1992 drop(reader);
1993 fs::remove_file(&path).expect("clean up");
1994 }
1995
1996 #[test]
1997 fn deleting_the_sections_changes_nothing_but_whether_they_are_there() {
1998 // Section 3.1, as close to directly as a test can put it. The same file, read once with the
1999 // sections and once with the generation moved past them, and the reader opens and scans the
2000 // same either way.
2001 let values = (1..=1500_i64).map(Some).collect::<Vec<_>>();
2002 let path = table_of("invariant", &values);
2003 build_stats(&path, "t", &[0]).expect("build");
2004
2005 let reader = reopen(&path);
2006 assert!(summary(&reader, 0).is_some());
2007 let generation = reader.table().generation();
2008 let held: Vec<_> = reader
2009 .table()
2010 .sections()
2011 .iter()
2012 .filter(|s| s.kind == *section::SUMMARY || s.kind == *section::SKETCHES)
2013 .copied()
2014 .collect();
2015 assert_eq!(held.len(), 2, "a summary and a sketch section");
2016 for section in &held {
2017 assert!(section.usable(generation));
2018 assert!(!section.usable(generation + 1), "a rewrite invalidates rather than corrupts");
2019 }
2020 let rows: usize =
2021 (0..reader.parts()).map(|part| reader.read(part, &[0]).expect("a part").len()).sum();
2022 assert_eq!(rows, 1500, "the scan is the scan whether the sections are read or not");
2023
2024 fs::remove_file(&path).expect("clean up");
2025 }
2026
2027 #[test]
2028 fn a_string_column_is_read_through_the_typed_path_and_measured_by_its_bytes() {
2029 // The other fast path. A varchar has no fixed width, so the byte total and the widest value
2030 // are measured per value, and the ends are the string ends rather than the hash ends.
2031 let path = path("strings");
2032 let mut writer =
2033 Writer::create(&path, "t", vec![Field::new("v", LogicalType::Varchar)]).expect("new");
2034 let words = ["alpha", "bravo", "charlie", "delta", "alpha"];
2035 let held = words.iter().map(|w| Value::Varchar((*w).into())).collect::<Vec<_>>();
2036 let chunk =
2037 Chunk::new(vec![Vector::from_values(LogicalType::Varchar, &held).expect("words")])
2038 .expect("one column");
2039 writer.append(&chunk).expect("a part");
2040 writer.finish().expect("commit");
2041 build_stats(&path, "t", &[0]).expect("build");
2042
2043 let reader = reopen(&path);
2044 let summary = summary(&reader, 0).expect("the summary");
2045 assert_eq!(summary.rows, 5);
2046 assert_eq!(summary.distinct, 4, "alpha twice");
2047 assert!(!summary.unique);
2048 assert_eq!(summary.bytes, words.iter().map(|w| w.len() as u64).sum::<u64>());
2049 assert_eq!(summary.widest, 7, "charlie");
2050 assert_eq!(summary.low, Some(Bound::Bytes(b"alpha".to_vec())));
2051 assert_eq!(summary.high, Some(Bound::Bytes(b"delta".to_vec())));
2052
2053 drop(reader);
2054 fs::remove_file(&path).expect("clean up");
2055 }
2056
2057 #[test]
2058 fn a_type_with_no_hash_rule_is_refused_by_name_rather_than_summarized_as_empty() {
2059 let path = table_of("refused", &[Some(1)]);
2060 let reader = reopen(&path);
2061 assert!(summarizable(&LogicalType::BigInt));
2062 assert!(!summarizable(&LogicalType::Interval));
2063 assert!(build_summary(&reader, 1).is_err(), "a column past the end");
2064 drop(reader);
2065 fs::remove_file(&path).expect("clean up");
2066 }
2067
2068 #[test]
2069 fn the_stored_sketch_depends_on_the_value_rule_and_not_only_on_the_hash() {
2070 // HASH_IDENTITY pins `hash64`, which is half of what a stored sketch depends on. The other
2071 // half is the rule that turns a value into the bytes `hash64` sees, and that rule lives in
2072 // `rudb_storage::count`. Changing it without bumping HASH_IDENTITY would leave every stored
2073 // sketch readable, accepted, and built over a different universe than the one a new sketch
2074 // is built over, which is exactly the merge the identity exists to prevent.
2075 //
2076 // So the rule is pinned here. If this fails because `hash_value` changed on purpose, the fix
2077 // is to bump HASH_IDENTITY and then update these numbers, in that order.
2078 assert_eq!(hash_value(&Value::BigInt(1)), Some(hash64(&1_u128.to_le_bytes())));
2079 assert_eq!(hash_value(&Value::Integer(1)), hash_value(&Value::BigInt(1)));
2080 assert_eq!(hash_value(&Value::Varchar("a".into())), Some(hash64(b"a")));
2081 assert_eq!(hash_value(&Value::Null), None);
2082 }
2083}