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