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