Skip to main content

rudb_native/
lib.rs

1//! Rudb's single-file columnar snapshot format.
2//!
3//! A committed directory names independently readable column pages. The first version handles
4//! scalar columns and one table; the file header already has two generation slots so an unfinished
5//! replacement directory cannot hide the last complete one.
6//!
7//! # Parts and stripes
8//!
9//! A part is one appended chunk, which is a thousand rows, and it is the unit a scan decodes and
10//! hands to the pipeline. A stripe is sixty four parts, and it is the unit the directory describes
11//! and the unit the file is laid out in: one page per column per stripe, holding that column's
12//! sixty four part payloads end to end.
13//!
14//! The two are separate because they are sized by different pressures. A part wants to be small
15//! because it is a vector and vectors live in cache. A stripe wants to be large because everything
16//! the directory holds is per stripe and the directory is one buffer that has to be read and
17//! decoded before a single row can be answered. A hundred million rows of the hundred and five
18//! column ClickBench table is ninety seven thousand parts, and a directory with a page entry and a
19//! pair of bounds per part per column is several hundred megabytes, which is what made that load
20//! fail before this split existed. Sixty four parts to a stripe divides that by sixty four.
21//!
22//! Where the parts of a page start is not in the directory either, for the same reason. Each
23//! stripe writes one index page holding a length and a checksum per part per column, and a reader
24//! preads the sixty four entries belonging to the column it wants. A scan reads the whole column
25//! page once and slices it; a sparse row fetch reads the index entries and then only the part it
26//! needs.
27
28#![forbid(unsafe_code)]
29
30use std::cmp::Ordering;
31use std::collections::{HashMap, VecDeque};
32use std::fs::{File, OpenOptions};
33use std::io::{Read, Seek, SeekFrom};
34use std::mem::{size_of, size_of_val};
35use std::path::Path;
36use std::sync::atomic::{AtomicUsize, Ordering as Atomic};
37use std::sync::{Arc, Mutex, OnceLock};
38
39use rudb_common::bounds::{Bound, Op};
40use rudb_common::{Error, Field, LogicalType, Result, Value};
41use rudb_encoding::{bitpack, chooser, integer, string};
42use rudb_storage::sieve::Sieve;
43use rudb_storage::{Probe, Range, Zone};
44use rudb_vector::string::StringColumn;
45use rudb_vector::validity::Validity;
46use rudb_vector::{Buffer, Chunk, Data, Packed, TextSource, Vector};
47
48const MAGIC: &[u8; 8] = b"RUDBNV10";
49const DIRECTORY: &[u8; 8] = b"RUDBDI10";
50const FORMAT: u32 = 19;
51const HEADER: u64 = 80;
52const SLOT_BYTES: usize = 28;
53const MAX_PAGE: usize = 256 * 1024 * 1024;
54const MAX_DIRECTORY: usize = 128 * 1024 * 1024;
55const FREQUENCIES: &[u8; 8] = b"RUDBFQ2\0";
56const FREQUENCY_CANDIDATES: usize = 32_768;
57const FREQUENCY_ENTRIES: usize = 512;
58const FREQUENCY_BUILD_RANK: usize = 10;
59const FREQUENCY_ORDINALS: usize = 65_536;
60/// The most threads the two per column passes at the end of a commit are spread over.
61///
62/// A table like `hits` has ninety numeric columns, so on a machine with more cores than this the
63/// cap is what decides how long the frequencies take rather than the columns are. It is here at all
64/// because each worker holds a candidate table and a decoded part, and a hundred of those at once
65/// on a narrow machine would be worse than waiting.
66const MAX_FREQUENCY_WORKERS: usize = 32;
67
68/// The most threads one stripe's encode is spread over.
69///
70/// Higher than the frequency cap because this is the load itself rather than a pass at the end of
71/// it, and the work is one column of sixty four parts, which is large enough that a thread that
72/// takes one is not a thread that was started for nothing. A machine with more cores than this has
73/// the rest of them on the Parquet read, which is still one thread and is the other half of #808.
74const MAX_ENCODE_WORKERS: usize = 32;
75
76/// The most bytes one column of one part may spend on a membership sieve.
77///
78/// A part is a thousand rows, so a filter sized for every one of them being distinct is about
79/// thirteen hundred bytes and this never binds in practice. It is here so that a part that somehow
80/// arrives much wider than a vector cannot put an unbounded index in the file. What does bind is the
81/// rule in `encode_column` that a sieve may not be as large as the part it indexes, which is a cap
82/// per column rather than one number for the whole file.
83const SIEVE_BUDGET: usize = 8 * 1024;
84
85fn io(error: std::io::Error) -> Error {
86    Error::io(error.to_string())
87}
88
89fn invalid(message: &str) -> Error {
90    Error::invalid_input(format!("invalid rudb native file: {message}"))
91}
92
93/// Adds a sequence of byte counts without an overflow the caller has to think about.
94fn sum(counts: impl Iterator<Item = u64>) -> u64 {
95    counts.fold(0, u64::saturating_add)
96}
97
98/// One column's span out of a per column list, or zero when the list is shorter than the column.
99fn span_bytes(spans: &[Span], at: usize) -> u64 {
100    spans.get(at).map_or(0, |span| u64::from(span.length))
101}
102
103/// One column's page out of a per column list, or zero when that column has no page at all.
104fn page_bytes(pages: &[Option<Page>], at: usize) -> u64 {
105    pages.get(at).and_then(Option::as_ref).map_or(0, Page::bytes)
106}
107
108fn checksum(bytes: &[u8]) -> u64 {
109    const P1: u64 = 11_400_714_785_074_694_791;
110    const P2: u64 = 14_029_467_366_897_019_727;
111    const P3: u64 = 1_609_587_929_392_839_161;
112    const P4: u64 = 9_650_029_242_287_828_579;
113    const P5: u64 = 2_870_177_450_012_600_261;
114    let round = |state: u64, word: u64| {
115        state.wrapping_add(word.wrapping_mul(P2)).rotate_left(31).wrapping_mul(P1)
116    };
117    let merge = |state: u64, lane: u64| (state ^ round(0, lane)).wrapping_mul(P1).wrapping_add(P4);
118    let word =
119        |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("eight checksum bytes"));
120
121    let mut at = 0;
122    let mut hash = if bytes.len() >= 32 {
123        let mut one = P1.wrapping_add(P2);
124        let mut two = P2;
125        let mut three = 0;
126        let mut four = 0_u64.wrapping_sub(P1);
127        while at + 32 <= bytes.len() {
128            one = round(one, word(at));
129            two = round(two, word(at + 8));
130            three = round(three, word(at + 16));
131            four = round(four, word(at + 24));
132            at += 32;
133        }
134        let combined = one
135            .rotate_left(1)
136            .wrapping_add(two.rotate_left(7))
137            .wrapping_add(three.rotate_left(12))
138            .wrapping_add(four.rotate_left(18));
139        merge(merge(merge(merge(combined, one), two), three), four)
140    } else {
141        P5
142    };
143    hash = hash.wrapping_add(bytes.len() as u64);
144    while at + 8 <= bytes.len() {
145        hash ^= round(0, word(at));
146        hash = hash.rotate_left(27).wrapping_mul(P1).wrapping_add(P4);
147        at += 8;
148    }
149    if at + 4 <= bytes.len() {
150        let tail = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four checksum bytes"));
151        hash ^= u64::from(tail).wrapping_mul(P1);
152        hash = hash.rotate_left(23).wrapping_mul(P2).wrapping_add(P3);
153        at += 4;
154    }
155    while at < bytes.len() {
156        hash ^= u64::from(bytes[at]).wrapping_mul(P5);
157        hash = hash.rotate_left(11).wrapping_mul(P1);
158        at += 1;
159    }
160    hash ^= hash >> 33;
161    hash = hash.wrapping_mul(P2);
162    hash ^= hash >> 29;
163    hash = hash.wrapping_mul(P3);
164    hash ^ (hash >> 32)
165}
166
167#[derive(Debug, Clone, Copy)]
168struct Slot {
169    offset: u64,
170    length: u32,
171    generation: u64,
172    hash: u64,
173}
174
175impl Slot {
176    fn bytes(self) -> [u8; SLOT_BYTES] {
177        let mut result = [0; SLOT_BYTES];
178        result[..8].copy_from_slice(&self.offset.to_le_bytes());
179        result[8..12].copy_from_slice(&self.length.to_le_bytes());
180        result[12..20].copy_from_slice(&self.generation.to_le_bytes());
181        result[20..28].copy_from_slice(&self.hash.to_le_bytes());
182        result
183    }
184
185    fn read(bytes: &[u8]) -> Self {
186        Self {
187            offset: u64::from_le_bytes(bytes[..8].try_into().expect("eight bytes")),
188            length: u32::from_le_bytes(bytes[8..12].try_into().expect("four bytes")),
189            generation: u64::from_le_bytes(bytes[12..20].try_into().expect("eight bytes")),
190            hash: u64::from_le_bytes(bytes[20..28].try_into().expect("eight bytes")),
191        }
192    }
193}
194
195#[derive(Debug, Clone, Copy)]
196struct Page {
197    offset: u64,
198    length: u32,
199    hash: u64,
200}
201
202impl Page {
203    /// How much of the file this page takes, for [`Reader::layout`].
204    fn bytes(&self) -> u64 {
205        u64::from(self.length)
206    }
207}
208
209#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
210enum FrequencyValue {
211    Null,
212    Integer(i128),
213    Code(u32),
214}
215
216#[derive(Debug, Clone)]
217struct FrequencyEntry {
218    value: FrequencyValue,
219    count: u64,
220}
221
222/// Exact leading frequencies for one column.
223///
224/// Values outside `entries` occur at most `omitted_max` times. This lets a count-descending TopN
225/// use the synopsis only when its last winner is strictly above every omitted value.
226#[derive(Debug, Clone)]
227struct FrequencySummary {
228    entries: Vec<FrequencyEntry>,
229    omitted_max: u64,
230    ordinals: Vec<u64>,
231}
232
233/// Sparse row ordinals covered by a numeric frequency candidate set.
234#[derive(Debug, Clone, PartialEq, Eq)]
235pub struct FrequencyOccurrences {
236    /// Upper bound for the frequency of every value absent from the fetched rows.
237    pub omitted_max: u64,
238    /// Table-wide row ordinals in ascending order.
239    pub ordinals: Vec<u64>,
240}
241
242/// Where one column's page for one stripe sits in the file.
243///
244/// A column page has no checksum of its own because every part inside it carries one, and the
245/// stripe's index page holds those. Checking a part on the way out of the page covers exactly the
246/// bytes a reader is about to decode, and covers them once whether the reader took the whole page
247/// or pulled one part out of the middle of it.
248#[derive(Debug, Clone, Copy, Default)]
249struct Span {
250    offset: u64,
251    length: u32,
252}
253
254/// One independently readable stripe of a table.
255#[derive(Debug, Clone)]
256pub struct Stripe {
257    rows: usize,
258    /// Rows in each part, in source order. Kept in the directory so that mapping a row ordinal to a
259    /// part, which every sparse fetch does, never reads the file.
260    parts: Vec<u32>,
261    /// The index page: one section per column, holding a length and a checksum for every part and
262    /// then a checksum of the section itself, so that a reader can pread one column's section and
263    /// still know it is intact.
264    index: Span,
265    pages: Vec<Span>,
266    memberships: Vec<Option<Page>>,
267    /// One page per column holding the membership sieve of every part of the stripe, for the
268    /// columns that have one. A column whose parts all declined a sieve has no page at all.
269    sieves: Vec<Option<Page>>,
270    zone: Zone,
271}
272
273impl Stripe {
274    /// Number of rows in this stripe.
275    #[must_use]
276    pub fn rows(&self) -> usize {
277        self.rows
278    }
279
280    /// Number of parts in this stripe.
281    #[must_use]
282    pub fn parts(&self) -> usize {
283        self.parts.len()
284    }
285}
286
287/// The committed table directory.
288#[derive(Debug, Clone)]
289pub struct Table {
290    name: String,
291    fields: Vec<Field>,
292    stripes: Vec<Stripe>,
293    rows: usize,
294    dictionaries: Vec<Option<Page>>,
295    frequencies: Vec<Option<FrequencySummary>>,
296    /// How many distinct values each column holds, for the columns that know.
297    ///
298    /// A dictionary entry is made the first time a value is seen and nothing ever removes one, so
299    /// the size of the dictionary is the number of distinct values in the column. That is the whole
300    /// story for a column with no null in it, and the wrong number by one for a column with a null
301    /// in it, because a null row is written as the code for the empty string and makes an entry the
302    /// dictionary would not otherwise have. The writer knows which case it is, since it counts the
303    /// non-null rows that use each code while it builds the frequency summary, and the reader cannot
304    /// work it out from the dictionary alone. So the writer settles it here.
305    distincts: Vec<Option<u64>>,
306}
307
308impl Table {
309    /// The SQL table name held by this snapshot.
310    #[must_use]
311    pub fn name(&self) -> &str {
312        &self.name
313    }
314
315    /// Columns in their SQL order.
316    #[must_use]
317    pub fn fields(&self) -> &[Field] {
318        &self.fields
319    }
320
321    /// Committed row count.
322    #[must_use]
323    pub fn rows(&self) -> usize {
324        self.rows
325    }
326
327    /// Independently readable stripes.
328    #[must_use]
329    pub fn stripes(&self) -> &[Stripe] {
330        &self.stripes
331    }
332}
333
334/// Where one column's bytes went, taken from the directory rather than by reading pages.
335#[derive(Debug, Clone)]
336pub struct ColumnLayout {
337    /// The column's name, so a report does not have to carry the field list beside this.
338    pub name: String,
339    /// The type, spelled the way the catalog spells it.
340    pub kind: String,
341    /// Every stripe's page of this column added up, which is the encoded data itself.
342    pub pages: u64,
343    /// Every stripe's exact code membership page for this column.
344    pub memberships: u64,
345    /// Every stripe's membership sieve page for this column.
346    pub sieves: u64,
347    /// The table wide dictionary of this column, if it has one.
348    pub dictionary: u64,
349}
350
351impl ColumnLayout {
352    /// Everything this column costs, which is what the file would lose if the column went.
353    #[must_use]
354    pub fn total(&self) -> u64 {
355        self.pages
356            .saturating_add(self.memberships)
357            .saturating_add(self.sieves)
358            .saturating_add(self.dictionary)
359    }
360}
361
362/// Where a whole file's bytes went.
363///
364/// Every number here comes out of the committed directory, so taking it costs one directory read
365/// however large the file is. That is the point: a 45 GB table has to be able to say where it went
366/// without being read, or nobody will ask.
367///
368/// The parts that are not a column are kept apart rather than shared out over the columns. The
369/// stripe index page holds a section per column and could be split, and the directory and the
370/// header cannot be, so splitting one of the three and not the others would read as if the columns
371/// accounted for everything. They do not, and the gap is the thing worth looking at.
372#[derive(Debug, Clone)]
373pub struct Layout {
374    /// The size of the file on disk.
375    pub file: u64,
376    /// Committed rows.
377    pub rows: usize,
378    /// Committed stripes.
379    pub stripes: usize,
380    /// Committed parts, which is how many chunks a scan reads.
381    pub parts: usize,
382    /// One entry per column, in the table's column order.
383    pub columns: Vec<ColumnLayout>,
384    /// Every stripe's index page, which carries a length and a checksum for every part of every
385    /// column and is charged per stripe rather than per column.
386    pub indexes: u64,
387    /// The committed directory itself, the one that was read to build this.
388    pub directory: u64,
389    /// The fixed header, which holds the magic, the format and the two directory slots.
390    pub header: u64,
391}
392
393impl Layout {
394    /// Everything the columns cost together.
395    #[must_use]
396    pub fn columns_total(&self) -> u64 {
397        self.columns.iter().map(ColumnLayout::total).fold(0, u64::saturating_add)
398    }
399
400    /// What the file holds that this does not account for.
401    ///
402    /// A committed file is written once and never rewritten in place, so an earlier directory and
403    /// the pages of an earlier snapshot are still in it. That is the honest place for them: they
404    /// are bytes on disk that no column owns.
405    #[must_use]
406    pub fn unaccounted(&self) -> u64 {
407        self.file
408            .saturating_sub(self.columns_total())
409            .saturating_sub(self.indexes)
410            .saturating_sub(self.directory)
411            .saturating_sub(self.header)
412    }
413}
414
415/// Appends pages and commits a new directory for one table.
416#[derive(Debug)]
417struct GlobalDictionary {
418    primary: HashMap<u64, u32>,
419    collisions: HashMap<u64, Vec<u32>>,
420    offsets: Vec<u32>,
421    payload: Vec<u8>,
422    counts: Vec<u64>,
423    nulls: u64,
424}
425
426impl GlobalDictionary {
427    fn new() -> Self {
428        Self {
429            primary: HashMap::new(),
430            collisions: HashMap::new(),
431            offsets: vec![0],
432            payload: Vec::new(),
433            counts: Vec::new(),
434            nulls: 0,
435        }
436    }
437
438    fn bytes(&self, code: u32) -> Option<&[u8]> {
439        let start = *self.offsets.get(code as usize)? as usize;
440        let end = *self.offsets.get(code as usize + 1)? as usize;
441        self.payload.get(start..end)
442    }
443
444    fn code(&mut self, text: &str) -> Result<u32> {
445        let hash = checksum(text.as_bytes());
446        if let Some(&code) = self.primary.get(&hash) {
447            if self.bytes(code) == Some(text.as_bytes()) {
448                return Ok(code);
449            }
450            if let Some(codes) = self.collisions.get(&hash) {
451                if let Some(code) =
452                    codes.iter().copied().find(|&code| self.bytes(code) == Some(text.as_bytes()))
453                {
454                    return Ok(code);
455                }
456            }
457            let code = self.insert(text)?;
458            self.collisions.entry(hash).or_default().push(code);
459            return Ok(code);
460        }
461        let code = self.insert(text)?;
462        self.primary.insert(hash, code);
463        Ok(code)
464    }
465
466    fn insert(&mut self, text: &str) -> Result<u32> {
467        let code = u32::try_from(self.offsets.len() - 1)
468            .map_err(|_| invalid("global dictionary has too many values"))?;
469        self.payload.extend_from_slice(text.as_bytes());
470        self.offsets.push(
471            u32::try_from(self.payload.len())
472                .map_err(|_| invalid("global dictionary payload exceeds 4 GiB"))?,
473        );
474        self.counts.push(0);
475        Ok(code)
476    }
477
478    /// This dictionary's values in sorted order, each as the first eight bytes of the value and the
479    /// code that holds it, so entry `rank` describes the value that sits at `rank` when the values
480    /// are sorted by their bytes.
481    ///
482    /// Codes themselves stay in first appearance order, which is what lets the writer hand one out
483    /// the moment it sees a value rather than waiting for the last stripe, and which also keeps a
484    /// stripe's codes close together because the data is clustered. This is what puts the values
485    /// back in order for anything that needs it, and it is separate from the codes so that getting
486    /// it costs a sort of the distinct values at the end rather than a rewrite of every code page.
487    ///
488    /// The sort compares the first eight bytes as one integer before it compares the values, which
489    /// settles almost every pair without touching the payload. Padding with zero on the right is
490    /// order preserving for byte strings, because a shorter value differs from a longer one that
491    /// starts the same way at a position where the shorter one has run out, and zero is below every
492    /// byte that could be there. A pair the head cannot settle falls through to the bytes.
493    ///
494    /// The heads are kept rather than thrown away once the sort is over, because a reader searching
495    /// this order wants exactly the same comparison and for exactly the same reason. Eight bytes an
496    /// entry of file is what buys a binary search that reads no values at all in the ordinary case.
497    fn ranked(&self) -> Vec<(u64, u32)> {
498        let count = self.offsets.len() - 1;
499        let mut ranked = (0..count)
500            .map(|code| {
501                let code = code as u32;
502                (head(self.bytes(code).unwrap_or_default()), code)
503            })
504            .collect::<Vec<_>>();
505        ranked.sort_unstable_by(|left, right| {
506            left.0.cmp(&right.0).then_with(|| self.bytes(left.1).cmp(&self.bytes(right.1)))
507        });
508        ranked
509    }
510
511    fn observe(&mut self, code: u32, null: bool) -> Result<()> {
512        if null {
513            self.nulls = self.nulls.saturating_add(1);
514            return Ok(());
515        }
516        let count = self
517            .counts
518            .get_mut(code as usize)
519            .ok_or_else(|| invalid("global dictionary count code is out of range"))?;
520        *count = count.saturating_add(1);
521        Ok(())
522    }
523}
524
525/// Appends pages and commits a new directory for one table.
526#[derive(Debug)]
527pub struct Writer {
528    file: File,
529    /// Where the next write goes, counted here rather than asked of the file.
530    ///
531    /// The file's own cursor is not ours. Building the numeric frequencies reads pages back through
532    /// [`read_at`], and a positional read is only positional about where it reads from: `pread`
533    /// leaves the cursor alone, and the call Windows has for it moves the cursor to the end of what
534    /// it read. A writer that asked the file where it was would then write the directory over a
535    /// page it had already written, which is what it did.
536    at: u64,
537    table: Table,
538    generation: u64,
539    /// The first and the last source position in every stripe, in the order the stripes were
540    /// written.
541    order: Vec<((u64, u64), (u64, u64))>,
542    next_order: u64,
543    dictionaries: Vec<Option<GlobalDictionary>>,
544    pending: Vec<PendingChunk>,
545}
546
547/// A chunk that has arrived and is waiting for the rest of its stripe.
548///
549/// The rows are kept rather than the pages they encode to, which is the whole of #808's first half.
550/// Encoding on arrival put every column of every part on the thread that called `append_at`, and
551/// that thread is the only one the load has. Encoding at the flush instead means a stripe's worth
552/// of work is on the table at once, and a stripe splits by column into a hundred and five pieces
553/// that share nothing.
554#[derive(Debug)]
555struct PendingChunk {
556    order: (u64, u64),
557    chunk: Chunk,
558}
559
560/// One column's share of a stripe, which is what one encode worker produces.
561///
562/// Indexed by part, so a stripe is a column of these and the write loop reads down one of them.
563/// That is also the order the loop wanted: `flush_pending` walks a column at a time and lays its
564/// parts next to each other, and it used to reach across a row of parts to do it.
565#[derive(Debug)]
566struct ColumnStripe {
567    pages: Vec<Vec<u8>>,
568    codes: Vec<Option<Vec<u32>>>,
569    sieves: Vec<Option<Sieve>>,
570    ranges: Vec<Range>,
571}
572
573/// Roughly what encoding a column of this type costs, for ordering the encode queue.
574///
575/// Only the order matters and only roughly. A string column hashes and copies every value into a
576/// dictionary and is in a different class from everything else, and among the fixed widths the wide
577/// ones carry more bytes through the cascade than the narrow ones. Anything finer than that would
578/// be a cost model, and the queue already absorbs a wrong guess: it only has to avoid finishing on
579/// a column nobody else can help with.
580fn weight(ty: &LogicalType) -> usize {
581    match ty {
582        LogicalType::Varchar | LogicalType::Blob => 64,
583        LogicalType::BigInt
584        | LogicalType::UBigInt
585        | LogicalType::Timestamp
586        | LogicalType::Double
587        | LogicalType::Decimal { .. } => 8,
588        LogicalType::Integer | LogicalType::UInteger | LogicalType::Date | LogicalType::Float => 4,
589        LogicalType::SmallInt | LogicalType::USmallInt => 2,
590        _ => 1,
591    }
592}
593
594/// Parts in one stripe.
595///
596/// Sixty four thousand rows is the smallest stripe that keeps the ClickBench directory in single
597/// digit megabytes at a hundred million rows, and it puts a four byte column's page at a quarter of
598/// a megabyte, which is the size a sequential read wants. Larger stripes buy a smaller directory
599/// and cost a sparse fetch, which has to read a page index before it can reach one part.
600pub const STRIPE_PARTS: usize = 64;
601
602/// Bytes one part takes in a stripe's index page: four for the length, eight for the checksum.
603const INDEX_ENTRY: usize = size_of::<u32>() + size_of::<u64>();
604
605/// Bytes one column's section of a stripe's index page takes, including its own trailing checksum.
606fn index_section(parts: usize) -> Result<usize> {
607    parts
608        .checked_mul(INDEX_ENTRY)
609        .and_then(|bytes| bytes.checked_add(size_of::<u64>()))
610        .ok_or_else(|| invalid("index page length overflow"))
611}
612
613impl Writer {
614    /// Creates a new v10 file and its first table.
615    ///
616    /// # Errors
617    ///
618    /// If the file exists, a field has no scalar encoding, or the path cannot be written.
619    pub fn create(
620        path: impl AsRef<Path>,
621        name: impl Into<String>,
622        fields: Vec<Field>,
623    ) -> Result<Self> {
624        for field in &fields {
625            type_tag(&field.ty)?;
626        }
627        let file =
628            OpenOptions::new().write(true).read(true).create_new(true).open(path).map_err(io)?;
629        let mut header = [0; HEADER as usize];
630        header[..8].copy_from_slice(MAGIC);
631        header[8..12].copy_from_slice(&FORMAT.to_le_bytes());
632        write_at(&file, 0, &header)?;
633        Ok(Self {
634            file,
635            at: HEADER,
636            dictionaries: fields
637                .iter()
638                .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
639                .collect(),
640            table: Table {
641                name: name.into(),
642                dictionaries: vec![None; fields.len()],
643                distincts: vec![None; fields.len()],
644                fields,
645                stripes: Vec::new(),
646                rows: 0,
647                frequencies: Vec::new(),
648            },
649            generation: 1,
650            order: Vec::new(),
651            next_order: 0,
652            pending: Vec::with_capacity(STRIPE_PARTS),
653        })
654    }
655
656    /// Appends bytes at the end of the file and moves the writer's own offset past them.
657    ///
658    /// Every write in here goes through this, so that [`Writer::at`] is the only answer to where
659    /// anything is and the file's cursor is never consulted for it.
660    fn put(&mut self, bytes: &[u8]) -> Result<()> {
661        write_at(&self.file, self.at, bytes)?;
662        self.at = self
663            .at
664            .checked_add(bytes.len() as u64)
665            .ok_or_else(|| invalid("native file length overflow"))?;
666        Ok(())
667    }
668
669    /// Writes one chunk as independently readable column pages.
670    ///
671    /// # Errors
672    ///
673    /// If its width or types differ from the declared table, or a page exceeds its bound.
674    pub fn append(&mut self, chunk: &Chunk) -> Result<()> {
675        let order = (self.next_order, 0);
676        self.next_order = self.next_order.saturating_add(1);
677        self.append_at(order, chunk)
678    }
679
680    /// Writes one chunk and records its source position for directory ordering.
681    ///
682    /// Pages may be encoded by parallel pipeline instances and reach the file in completion order.
683    /// The stripe they land in is sorted by this key at commit, and [`Self::finish`] rejects a
684    /// sequence whose parts do not come out in source order once the stripes are sorted, because a
685    /// stripe groups whatever arrived together and cannot put a late part back where it belongs.
686    ///
687    /// # Errors
688    ///
689    /// The same as [`Self::append`].
690    pub fn append_at(&mut self, order: (u64, u64), chunk: &Chunk) -> Result<()> {
691        if chunk.is_empty() {
692            return Ok(());
693        }
694        self.admit(chunk)?;
695        if self.pending.last().is_some_and(|last| last.order > order) {
696            self.flush_pending()?;
697        }
698        // Cloned rather than encoded, and a clone of a chunk that owns its buffers is a copy of
699        // them. Sixty four parts of a hundred and five columns is tens of megabytes held for the
700        // length of a stripe and a few seconds of memory traffic over a whole ClickBench load,
701        // against the hundreds of seconds of encode this is what lets off one thread.
702        self.pending.push(PendingChunk { order, chunk: chunk.clone() });
703        if self.pending.len() == STRIPE_PARTS {
704            self.flush_pending()?;
705        }
706        Ok(())
707    }
708
709    /// Writes a run of chunks as one stripe of its own.
710    ///
711    /// [`Self::append_at`] decides where a stripe ends by watching the orders go past, which works
712    /// when one caller hands over every chunk in source order and does not when several do. A
713    /// writer being fed by more than one pipeline instance sees the orders interleave, and a stripe
714    /// that ends every time two of them cross is a stripe of one or two parts.
715    ///
716    /// So the grouping moves to the caller. Whoever is buffering hands over a run it already knows
717    /// is contiguous and in order, and gets a stripe holding exactly that run. The orders still
718    /// have to come out in source order once the stripes are sorted, which [`Self::finish`] checks,
719    /// so the runs from different callers may interleave with each other but may not overlap.
720    ///
721    /// # Errors
722    ///
723    /// The same as [`Self::append`], and if the run is longer than [`STRIPE_PARTS`].
724    pub fn append_stripe(&mut self, parts: Vec<((u64, u64), Chunk)>) -> Result<()> {
725        if parts.len() > STRIPE_PARTS {
726            return Err(invalid("a stripe was handed more parts than it holds"));
727        }
728        // Whatever an earlier caller left behind is its own stripe rather than the front of this
729        // one, because the two runs are from different places in the source and a stripe is a run.
730        self.flush_pending()?;
731        for (order, chunk) in parts {
732            if chunk.is_empty() {
733                continue;
734            }
735            self.admit(&chunk)?;
736            self.pending.push(PendingChunk { order, chunk });
737        }
738        self.flush_pending()
739    }
740
741    /// Checks a chunk against the declared table and counts its rows in.
742    fn admit(&mut self, chunk: &Chunk) -> Result<()> {
743        if chunk.width() != self.table.fields.len() {
744            return Err(invalid("chunk width differs from table schema"));
745        }
746        for (index, field) in self.table.fields.iter().enumerate() {
747            if chunk.column(index)?.logical_type() != &field.ty {
748                return Err(invalid("chunk type differs from table schema"));
749            }
750        }
751        self.table.rows = self
752            .table
753            .rows
754            .checked_add(chunk.len())
755            .ok_or_else(|| invalid("row count overflow"))?;
756        Ok(())
757    }
758
759    /// Encodes one column's parts of a stripe, with the column's dictionary to itself.
760    ///
761    /// Nothing here is shared with another column. The dictionary belongs to this one, the sieve
762    /// reads only this one, and the page bytes go in a vector of this one's own. That is why the
763    /// fan out below can hand a whole column to a thread and take a plain `&mut` on the dictionary
764    /// rather than making it something several threads can grow at once, which is the harder half
765    /// of #808 and is still open.
766    fn encode_column(
767        index: usize,
768        held: &[PendingChunk],
769        mut dictionary: Option<&mut GlobalDictionary>,
770    ) -> Result<ColumnStripe> {
771        let mut stripe = ColumnStripe {
772            pages: Vec::with_capacity(held.len()),
773            codes: Vec::with_capacity(held.len()),
774            sieves: Vec::with_capacity(held.len()),
775            ranges: Vec::with_capacity(held.len()),
776        };
777        for pending in held {
778            let column = pending.chunk.column(index)?;
779            let (bytes, unique) = encode(column, dictionary.as_deref_mut())?;
780            if bytes.len() > MAX_PAGE {
781                return Err(invalid("column page exceeds the configured bound"));
782            }
783            // The range is built first because the sieve reads it rather than walking the column a
784            // second time to find out how wide it is.
785            let range = Range::of(column);
786            // A column with a global dictionary already has an exact membership index per stripe,
787            // so an approximate one beside it would cost a hash of every string in the table to
788            // answer a question that is already answered. What it would buy is the finer grain, a
789            // part rather than a stripe, and that is worth coming back for on its own.
790            //
791            // A sieve at least as large as the part it indexes is not written. A reader reads the
792            // sieve to decide whether to read the part, so when the sieve is the larger of the two
793            // it has already spent more than the read it is trying to avoid, and that holds even if
794            // it rejects every time. It is a necessary condition rather than the whole rule, which
795            // is that a sieve pays when its bytes are under the rejection rate times the part's,
796            // but the rejection rate depends on what a query probes for and the writer does not
797            // know that. The necessary half needs two numbers that are both in hand here.
798            let sieve = match dictionary {
799                Some(_) => None,
800                None => Sieve::of(column, &range, SIEVE_BUDGET)
801                    .filter(|sieve| sieve.len() < bytes.len()),
802            };
803            stripe.pages.push(bytes);
804            stripe.codes.push(unique);
805            stripe.sieves.push(sieve);
806            stripe.ranges.push(range);
807        }
808        Ok(stripe)
809    }
810
811    /// Encodes a whole stripe, one column to a worker.
812    ///
813    /// The columns are handed out through a queue rather than dealt in equal piles, because they
814    /// are nothing like equal: `URL` on ClickBench is a global dictionary of sixty one million
815    /// strings and `IsMobile` is a byte. A pile that happened to hold the four large string columns
816    /// would be the whole stripe and the other workers would be waiting on it. The queue is sorted
817    /// so the expensive ones are taken first, which is the classic answer to a last job that runs
818    /// longer than everything after it.
819    fn encode_columns(&mut self, held: &[PendingChunk]) -> Result<Vec<ColumnStripe>> {
820        let width = self.table.fields.len();
821        let workers = std::thread::available_parallelism()
822            .map_or(1, usize::from)
823            .min(MAX_ENCODE_WORKERS)
824            .min(width);
825        if workers <= 1 || held.len() <= 1 {
826            return self
827                .dictionaries
828                .iter_mut()
829                .enumerate()
830                .map(|(index, dictionary)| Self::encode_column(index, held, dictionary.as_mut()))
831                .collect();
832        }
833        // The dictionaries are moved out and back rather than borrowed, because a worker that takes
834        // the next column off a queue cannot be holding a borrow of the vector the queue came from.
835        let mut jobs: Vec<(usize, Option<GlobalDictionary>)> =
836            std::mem::take(&mut self.dictionaries).into_iter().enumerate().collect();
837        // Popped from the back, so the expensive columns go last in the vector.
838        jobs.sort_by_key(|(index, _)| weight(&self.table.fields[*index].ty));
839        let queue = Mutex::new(jobs);
840        let pieces = std::thread::scope(|scope| {
841            (0..workers)
842                .map(|_| {
843                    scope.spawn(|| {
844                        let mut mine = Vec::new();
845                        loop {
846                            let taken = queue
847                                .lock()
848                                .map_err(|_| Error::internal("a native encode worker panicked"))?
849                                .pop();
850                            let Some((index, mut dictionary)) = taken else { break };
851                            let encoded = Self::encode_column(index, held, dictionary.as_mut())?;
852                            mine.push((index, dictionary, encoded));
853                        }
854                        Ok(mine)
855                    })
856                })
857                .collect::<Vec<_>>()
858                .into_iter()
859                .map(|handle| {
860                    handle.join().map_err(|_| Error::internal("a native encode worker panicked"))?
861                })
862                .collect::<Result<Vec<_>>>()
863        })?;
864        let mut dictionaries: Vec<Option<GlobalDictionary>> = (0..width).map(|_| None).collect();
865        let mut encoded: Vec<Option<ColumnStripe>> = (0..width).map(|_| None).collect();
866        for piece in pieces {
867            for (index, dictionary, stripe) in piece {
868                dictionaries[index] = dictionary;
869                encoded[index] = Some(stripe);
870            }
871        }
872        self.dictionaries = dictionaries;
873        encoded
874            .into_iter()
875            .map(|stripe| stripe.ok_or_else(|| Error::internal("a column was never encoded")))
876            .collect()
877    }
878
879    /// Writes the buffered parts as one stripe, each column's parts contiguous on disk.
880    fn flush_pending(&mut self) -> Result<()> {
881        if self.pending.is_empty() {
882            return Ok(());
883        }
884        let width = self.table.fields.len();
885        // Held here rather than read off the writer, because writing a page needs the writer and
886        // the borrow checker is right that those are two different uses of it.
887        let mut held = std::mem::take(&mut self.pending);
888        let parts = held.len();
889        let encoded = self.encode_columns(&held)?;
890        let mut pages = Vec::with_capacity(width);
891        let mut memberships = vec![None; width];
892        let mut ranges = Vec::with_capacity(width);
893        let mut index = Vec::with_capacity(width.saturating_mul(index_section(parts)?));
894        for stripe in &encoded {
895            let offset = self.at;
896            let section = index.len();
897            let mut length = 0_usize;
898            for bytes in &stripe.pages {
899                write_at(&self.file, self.at + length as u64, bytes)?;
900                put_u32(
901                    &mut index,
902                    u32::try_from(bytes.len()).map_err(|_| invalid("part length overflow"))?,
903                );
904                put_u64(&mut index, checksum(bytes));
905                length = length
906                    .checked_add(bytes.len())
907                    .ok_or_else(|| invalid("column page length overflow"))?;
908            }
909            let hash = checksum(&index[section..]);
910            put_u64(&mut index, hash);
911            if length > MAX_PAGE {
912                return Err(invalid("column page exceeds the configured bound"));
913            }
914            self.at = self
915                .at
916                .checked_add(length as u64)
917                .ok_or_else(|| invalid("native file length overflow"))?;
918            pages.push(Span {
919                offset,
920                length: u32::try_from(length).map_err(|_| invalid("page length overflow"))?,
921            });
922            ranges.push(merged_range(stripe.ranges.iter().cloned()));
923        }
924        for (membership, stripe) in memberships.iter_mut().zip(&encoded) {
925            if stripe.codes.iter().all(Option::is_none) {
926                continue;
927            }
928            let lists = stripe
929                .codes
930                .iter()
931                .map(|codes| codes.clone().unwrap_or_default())
932                .collect::<Vec<_>>();
933            let bytes = encode_membership(&merged_codes(lists));
934            let offset = self.at;
935            self.put(&bytes)?;
936            *membership = Some(Page {
937                offset,
938                length: u32::try_from(bytes.len())
939                    .map_err(|_| invalid("membership page length overflow"))?,
940                hash: checksum(&bytes),
941            });
942        }
943        let mut sieves = vec![None; width];
944        for (page, stripe) in sieves.iter_mut().zip(&encoded) {
945            if stripe.sieves.iter().all(Option::is_none) {
946                continue;
947            }
948            let bytes = encode_sieves(stripe.sieves.iter())?;
949            let offset = self.at;
950            self.put(&bytes)?;
951            *page = Some(Page {
952                offset,
953                length: u32::try_from(bytes.len())
954                    .map_err(|_| invalid("sieve page length overflow"))?,
955                hash: checksum(&bytes),
956            });
957        }
958        let offset = self.at;
959        self.put(&index)?;
960        let index = Span {
961            offset,
962            length: u32::try_from(index.len())
963                .map_err(|_| invalid("index page length overflow"))?,
964        };
965        let mut rows = 0_usize;
966        let mut lengths = Vec::with_capacity(parts);
967        let mut span = None;
968        for pending in held.drain(..) {
969            let part = pending.chunk.len();
970            rows = rows.checked_add(part).ok_or_else(|| invalid("row count overflow"))?;
971            lengths.push(u32::try_from(part).map_err(|_| invalid("part row count overflow"))?);
972            span = Some(
973                span.map_or((pending.order, pending.order), |(first, _)| (first, pending.order)),
974            );
975        }
976        self.order.push(span.ok_or_else(|| invalid("a stripe was flushed with no parts"))?);
977        self.table.stripes.push(Stripe {
978            rows,
979            parts: lengths,
980            index,
981            pages,
982            memberships,
983            sieves,
984            zone: Zone::from_ranges(ranges),
985        });
986        // Back where it came from, empty, so the next stripe buffers into the same allocation.
987        self.pending = held;
988        Ok(())
989    }
990
991    /// Finds exact heavy hitters without keeping a hash table for every numeric column while the
992    /// load is live. The pages are already in the target file, so one column at a time uses a
993    /// bounded Misra-Gries candidate table and then recounts only those candidates.
994    fn numeric_frequency(&self, column: usize) -> Result<Option<FrequencySummary>> {
995        let ty = &self.table.fields[column].ty;
996        if !matches!(
997            ty,
998            LogicalType::TinyInt
999                | LogicalType::SmallInt
1000                | LogicalType::Integer
1001                | LogicalType::BigInt
1002                | LogicalType::UTinyInt
1003                | LogicalType::USmallInt
1004                | LogicalType::UInteger
1005                | LogicalType::UBigInt
1006                | LogicalType::Date
1007                | LogicalType::Timestamp
1008        ) {
1009            return Ok(None);
1010        }
1011        let mut candidates: HashMap<FrequencyValue, u32> = HashMap::new();
1012        let mut decrements = 0_u64;
1013        self.visit_numeric(column, |_, value| {
1014            if let Some(count) = candidates.get_mut(&value) {
1015                *count = count.saturating_add(1);
1016            } else if candidates.len() < FREQUENCY_CANDIDATES {
1017                candidates.insert(value, 1);
1018            } else {
1019                candidates.retain(|_, count| {
1020                    *count -= 1;
1021                    *count != 0
1022                });
1023                decrements = decrements.saturating_add(1);
1024            }
1025        })?;
1026        let (exact, ordinals) = if decrements == 0 {
1027            (
1028                candidates
1029                    .into_iter()
1030                    .map(|(value, count)| (value, u64::from(count)))
1031                    .collect::<HashMap<_, _>>(),
1032                Vec::new(),
1033            )
1034        } else {
1035            let mut lower = candidates.values().copied().collect::<Vec<_>>();
1036            lower.sort_unstable_by(|left, right| right.cmp(left));
1037            if lower.len() < FREQUENCY_BUILD_RANK
1038                || u64::from(lower[FREQUENCY_BUILD_RANK - 1]) <= decrements
1039            {
1040                return Ok(None);
1041            }
1042            let mut exact =
1043                candidates.into_keys().map(|value| (value, 0_u64)).collect::<HashMap<_, _>>();
1044            let mut ordinals = Vec::new();
1045            let mut exceeded = false;
1046            self.visit_numeric(column, |ordinal, value| {
1047                if let Some(count) = exact.get_mut(&value) {
1048                    *count = count.saturating_add(1);
1049                    if !exceeded {
1050                        if ordinals.len() < FREQUENCY_ORDINALS {
1051                            ordinals.push(ordinal);
1052                        } else {
1053                            ordinals.clear();
1054                            exceeded = true;
1055                        }
1056                    }
1057                }
1058            })?;
1059            (exact, ordinals)
1060        };
1061        let mut entries = exact
1062            .into_iter()
1063            .map(|(value, count)| FrequencyEntry { value, count })
1064            .collect::<Vec<_>>();
1065        entries.sort_unstable_by(|left, right| {
1066            right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
1067        });
1068        let omitted_max =
1069            entries.get(FREQUENCY_ENTRIES).map_or(decrements, |entry| decrements.max(entry.count));
1070        entries.truncate(FREQUENCY_ENTRIES);
1071        Ok(Some(FrequencySummary { entries, omitted_max, ordinals }))
1072    }
1073
1074    fn visit_numeric(
1075        &self,
1076        column: usize,
1077        mut visit: impl FnMut(u64, FrequencyValue),
1078    ) -> Result<()> {
1079        let ty = &self.table.fields[column].ty;
1080        let mut start = 0_u64;
1081        for stripe in &self.table.stripes {
1082            let spans = read_index(&self.file, stripe, column)?;
1083            let page = stripe.pages[column];
1084            let mut bytes = vec![0; page.length as usize];
1085            read_at(&self.file, page.offset, &mut bytes)?;
1086            for (span, &rows) in spans.iter().zip(&stripe.parts) {
1087                let part = part_bytes(&bytes, *span)?;
1088                if checksum(part) != span.hash {
1089                    return Err(invalid("column page checksum differs while building frequencies"));
1090                }
1091                let rows = rows as usize;
1092                let vector = decode(ty, rows, part, None)?;
1093                // row at a time: frequency construction visits decoded values to update bounded candidates.
1094                for row in 0..rows {
1095                    let value = if vector.is_null_at(row) {
1096                        FrequencyValue::Null
1097                    } else {
1098                        // An unsigned column has no signed reading, and the documented fallback is
1099                        // the value itself. Every unsigned width the format stores fits in the
1100                        // `i128` a candidate is keyed by, so nothing is lost on the way through.
1101                        let widened = match vector.signed_at(row) {
1102                            Some(value) => Some(value),
1103                            None => match vector.value_at(row) {
1104                                Value::UTinyInt(value) => Some(i128::from(value)),
1105                                Value::USmallInt(value) => Some(i128::from(value)),
1106                                Value::UInteger(value) => Some(i128::from(value)),
1107                                Value::UBigInt(value) => Some(i128::from(value)),
1108                                _ => None,
1109                            },
1110                        };
1111                        FrequencyValue::Integer(widened.ok_or_else(|| {
1112                            invalid("numeric frequency page did not contain an integer value")
1113                        })?)
1114                    };
1115                    visit(start.saturating_add(row as u64), value);
1116                }
1117                start = start.saturating_add(rows as u64);
1118            }
1119        }
1120        Ok(())
1121    }
1122
1123    /// Builds independent numeric synopses concurrently after all column pages are committed.
1124    ///
1125    /// The columns go through a queue rather than being cut into equal runs, because they are not
1126    /// equally expensive and they are not shuffled. A `BIGINT` column carries eight times the bytes
1127    /// of a `TINYINT` through the decode, and a run of them sits together in a schema the way it
1128    /// sits together in `hits`, so a worker that was handed the wrong six columns finishes long
1129    /// after one that was handed the right six and the whole phase waits for it.
1130    fn numeric_frequencies(&self) -> Result<Vec<Option<FrequencySummary>>> {
1131        let mut columns = self
1132            .table
1133            .fields
1134            .iter()
1135            .enumerate()
1136            .filter_map(|(column, field)| {
1137                matches!(
1138                    field.ty,
1139                    LogicalType::TinyInt
1140                        | LogicalType::SmallInt
1141                        | LogicalType::Integer
1142                        | LogicalType::BigInt
1143                        | LogicalType::UTinyInt
1144                        | LogicalType::USmallInt
1145                        | LogicalType::UInteger
1146                        | LogicalType::UBigInt
1147                        | LogicalType::Date
1148                        | LogicalType::Timestamp
1149                )
1150                .then_some(column)
1151            })
1152            .collect::<Vec<_>>();
1153        let workers = std::thread::available_parallelism()
1154            .map_or(1, usize::from)
1155            .min(MAX_FREQUENCY_WORKERS)
1156            .min(columns.len());
1157        if workers <= 1 {
1158            let mut frequencies = vec![None; self.table.fields.len()];
1159            for column in columns {
1160                frequencies[column] = self.numeric_frequency(column)?;
1161            }
1162            return Ok(frequencies);
1163        }
1164        // Popped from the back, so the expensive columns are the ones taken first and the cheap ones
1165        // are what is left to fill in behind them.
1166        columns.sort_by_key(|&column| weight(&self.table.fields[column].ty));
1167        let queue = Mutex::new(columns);
1168        let pieces = std::thread::scope(|scope| {
1169            (0..workers)
1170                .map(|_| {
1171                    scope.spawn(|| {
1172                        let mut mine = Vec::new();
1173                        loop {
1174                            let taken = queue
1175                                .lock()
1176                                .map_err(|_| Error::internal("a native frequency worker panicked"))?
1177                                .pop();
1178                            let Some(column) = taken else { break };
1179                            mine.push((column, self.numeric_frequency(column)?));
1180                        }
1181                        Ok(mine)
1182                    })
1183                })
1184                .collect::<Vec<_>>()
1185                .into_iter()
1186                .map(|handle| {
1187                    handle
1188                        .join()
1189                        .map_err(|_| Error::internal("a native frequency worker panicked"))?
1190                })
1191                .collect::<Result<Vec<_>>>()
1192        })?;
1193        let mut frequencies = vec![None; self.table.fields.len()];
1194        for piece in pieces {
1195            for (column, summary) in piece {
1196                frequencies[column] = summary;
1197            }
1198        }
1199        Ok(frequencies)
1200    }
1201
1202    /// Commits the directory and syncs the file before publishing its header slot.
1203    ///
1204    /// # Errors
1205    ///
1206    /// If directory encoding, writing, or syncing fails.
1207    pub fn finish(mut self) -> Result<Table> {
1208        self.flush_pending()?;
1209        let mut stripes = std::mem::take(&mut self.order)
1210            .into_iter()
1211            .zip(std::mem::take(&mut self.table.stripes))
1212            .collect::<Vec<_>>();
1213        stripes.sort_by_key(|(order, _)| order.0);
1214        let mut previous: Option<(u64, u64)> = None;
1215        for ((first, last), _) in &stripes {
1216            if previous.is_some_and(|previous| previous >= *first) {
1217                return Err(invalid("chunks did not arrive in source order"));
1218            }
1219            previous = Some(*last);
1220        }
1221        self.table.stripes = stripes.into_iter().map(|(_, stripe)| stripe).collect();
1222        self.table.frequencies = self.numeric_frequencies()?;
1223        let dictionaries = std::mem::take(&mut self.dictionaries);
1224        let orders = rankings(&dictionaries)?;
1225        for (index, (dictionary, order)) in dictionaries.into_iter().zip(orders).enumerate() {
1226            let Some(dictionary) = dictionary else { continue };
1227            // A code nothing counted is a code no non-null row of this column holds, which is the
1228            // empty string a null was written as and nothing else, because a code is only ever made
1229            // by a row asking for one.
1230            self.table.distincts[index] =
1231                Some(dictionary.counts.iter().filter(|count| **count != 0).count() as u64);
1232            self.table.frequencies[index] = Some(code_frequency(&dictionary));
1233            let encoded = encode_global_dictionary(dictionary, &order)?;
1234            let offset = self.at;
1235            self.put(&encoded.index)?;
1236            self.put(&encoded.ranks)?;
1237            for block in &encoded.payload {
1238                self.put(block)?;
1239            }
1240            let payload_len =
1241                encoded.payload.iter().try_fold(0_usize, |len, block| len.checked_add(block.len()));
1242            let length = payload_len
1243                .and_then(|len| len.checked_add(encoded.index.len()))
1244                .and_then(|len| len.checked_add(encoded.ranks.len()))
1245                .ok_or_else(|| invalid("dictionary page length overflow"))?;
1246            self.table.dictionaries[index] = Some(Page {
1247                offset,
1248                length: u32::try_from(length)
1249                    .map_err(|_| invalid("dictionary page length overflow"))?,
1250                hash: checksum(&encoded.index),
1251            });
1252        }
1253        let directory = encode_directory(&self.table)?;
1254        if directory.len() > MAX_DIRECTORY {
1255            return Err(invalid("directory exceeds the configured bound"));
1256        }
1257        let offset = self.at;
1258        self.put(&directory)?;
1259        self.file.sync_all().map_err(io)?;
1260        let slot = Slot {
1261            offset,
1262            length: u32::try_from(directory.len())
1263                .map_err(|_| invalid("directory length overflow"))?,
1264            generation: self.generation,
1265            hash: checksum(&directory),
1266        };
1267        // The one write that is not an append, and the last one. It goes back over the slot in the
1268        // header, so it names its offset rather than going through `put`, and `at` does not move.
1269        write_at(&self.file, 16, &slot.bytes())?;
1270        self.file.sync_all().map_err(io)?;
1271        Ok(self.table)
1272    }
1273}
1274
1275/// Reads committed native column pages without holding the table in memory.
1276#[derive(Debug, Clone)]
1277pub struct Reader {
1278    file: Arc<File>,
1279    table: Arc<Table>,
1280    dictionaries: Arc<Vec<OnceLock<Arc<Vector>>>>,
1281    /// Held while a global dictionary is being opened, one per column.
1282    ///
1283    /// The [`OnceLock`] above says whether one has been opened, which is the question a reader that
1284    /// already has it needs answered and is free. It does not say whether one is being opened, and
1285    /// the difference matters because every worker of a scan wants the same dictionary at the same
1286    /// moment. Without this they all miss, all read the page, all verify it and all decode it, and
1287    /// all but one throw the answer away. ClickBench 38 reads the URL dictionary, which is 515,958
1288    /// entries, and was paying for it twice.
1289    loading: Arc<Vec<Mutex<()>>>,
1290    /// How many global dictionaries have been opened. A scan of a dictionary column should open its
1291    /// dictionary once however many workers it has, and the test that says so is the only thing
1292    /// keeping it that way.
1293    opened: Arc<AtomicUsize>,
1294    /// The membership sieves of one stripe of one column, by column and then by stripe, read the
1295    /// first time a probe asks about them. A query filters on one or two columns and never looks at
1296    /// the rest, so reading these at open would be the whole index for the sake of a fraction of it.
1297    sieves: Arc<Vec<Vec<SieveSlot>>>,
1298    /// Which stripe and which part of it every part of the table is, by table wide part number.
1299    places: Arc<Vec<Place>>,
1300    cache: Arc<Vec<Mutex<Cached>>>,
1301    /// How many whole stripe pages have been read, which is what the sharing above is judged on. A
1302    /// scan of a column should read each of its stripes once however many workers it has.
1303    pages: Arc<AtomicUsize>,
1304    /// How many index sections have been read. A scan of a column should read each of its stripes
1305    /// once here too, and the test that says so is the only thing keeping it that way.
1306    indexes: Arc<AtomicUsize>,
1307    /// How many stripes of one column the page cache keeps. See [`CACHED_STRIPES_PER_COLUMN`] for
1308    /// what sets it and [`Reader::keep_stripes`] for who raises it.
1309    kept: Arc<AtomicUsize>,
1310    /// The file's size when it was opened, for [`Reader::layout`].
1311    size: u64,
1312    /// The committed directory's size, for [`Reader::layout`].
1313    directory: u64,
1314    /// What opening the file cost, which is a number rather than a claim.
1315    opening: Opening,
1316}
1317
1318/// What [`Reader::open`] read before it returned.
1319///
1320/// `spec/stats/04-in-memory.md` section 4.2 says opening a table reads the header and the directory
1321/// and nothing else, and once that document's statistics are in the file the tempting change is to
1322/// load a column summary or two on the way past, because they are small and the next query will
1323/// want them. A hundred milliseconds of that is a hundred milliseconds nobody asked for, and an
1324/// embedded database is opened by processes that are about to run one trivial query.
1325///
1326/// So the claim gets a number. Both of these are fixed by the schema and the stripe count and are
1327/// independent of how many rows the file holds, and the test that says so is what stops the
1328/// tempting change from landing quietly.
1329#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1330pub struct Opening {
1331    /// How many times the file was read. The header, then each directory slot that looked valid
1332    /// enough to check, so three at the most.
1333    pub reads: u32,
1334    /// How many bytes those reads asked for.
1335    pub bytes: u64,
1336}
1337
1338/// What a reader has read, while it was being opened and since.
1339#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1340pub struct Reads {
1341    /// What opening cost, before any query had been planned.
1342    pub opening: Opening,
1343    /// Whole stripe pages read since.
1344    pub pages: usize,
1345    /// Index sections read since.
1346    pub indexes: usize,
1347    /// Global dictionaries opened since. One per dictionary column that a query touched, however
1348    /// many workers touched it, which is a claim only a test can keep true.
1349    pub dictionaries: usize,
1350}
1351
1352/// Where one table wide part number lands.
1353#[derive(Debug, Clone, Copy)]
1354struct Place {
1355    stripe: u32,
1356    part: u32,
1357    rows: u32,
1358}
1359
1360/// One part's bytes inside one column page.
1361#[derive(Debug, Clone, Copy)]
1362struct PartSpan {
1363    start: usize,
1364    length: usize,
1365    hash: u64,
1366}
1367
1368/// What a reader holds for one stripe of one column.
1369///
1370/// The index is small and is loaded whether the caller wants the whole page or one part of it. The
1371/// page is loaded only by a scan, because a sparse fetch that wants a thousand rows out of sixty
1372/// four thousand would be reading sixty four times what it uses.
1373#[derive(Debug, Clone)]
1374struct CachedColumn {
1375    stripe: usize,
1376    index: Arc<Vec<PartSpan>>,
1377    page: Option<Arc<Vec<u8>>>,
1378}
1379
1380/// One column's stripes a reader holds, and which of them somebody is reading right now.
1381///
1382/// The pages are one slot per stripe of the table rather than a list of the ones being kept, so
1383/// finding a page is an index and not a walk. That matters because the walk happened under the
1384/// lock, once per part per column, and a scan that gives a whole stripe to each of thirty two
1385/// workers keeps enough pages that walking them was the longest thing the lock was held for. The
1386/// slots cost a pointer per stripe per column, which on the ClickBench file is eight kilobytes
1387/// against the forty megabytes of pages they point at. `order` is which of them are filled, oldest
1388/// first, because that is the one thing the slots cannot say by themselves.
1389///
1390/// `loading` is what keeps a scan from reading the same page once per worker. It is a list and not
1391/// a set because it holds at most one stripe per worker on the column and is walked far less often
1392/// than a hash of it would be built.
1393///
1394/// `index` is every index this reader has ever read for the column, one slot per stripe, and it is
1395/// never evicted. An index is a few hundred bytes and a page is a quarter of a megabyte, so the two
1396/// do not belong under the same budget. Riding in the page cache meant a worker that came back to a
1397/// stripe after its page had been evicted read the index again with it, which on the full
1398/// ClickBench file was about thirteen hundred reads out of a hundred and fourteen thousand.
1399#[derive(Debug, Default)]
1400struct Cached {
1401    pages: Vec<Option<Arc<Vec<u8>>>>,
1402    order: VecDeque<usize>,
1403    loading: Vec<usize>,
1404    index: Vec<Option<Arc<Vec<PartSpan>>>>,
1405}
1406
1407/// Stripes of one column a reader keeps the bytes of, when nobody has asked for more.
1408///
1409/// This has to hold at least as many stripes as a column has workers in it at once, or the workers
1410/// evict each other's pages and read them again. Four is what a scan that hands parts out in order
1411/// needs, because then every worker is within a few parts of every other and at most a couple of
1412/// stripes are open at a time. A scan that hands a whole stripe to each worker has one stripe open
1413/// per worker for the length of that stripe, and it says so with [`Reader::keep_stripes`] rather
1414/// than paying for sixteen slots on every table that is read one part at a time.
1415///
1416/// It multiplies by the page size, which is a quarter of a megabyte for a four byte column, and by
1417/// the number of columns a query touches.
1418const CACHED_STRIPES_PER_COLUMN: usize = 4;
1419
1420/// The sieves of one stripe of one column, once somebody has asked for them.
1421type SieveSlot = OnceLock<Arc<Vec<Option<Sieve>>>>;
1422
1423#[derive(Debug)]
1424struct NativeText {
1425    file: Arc<File>,
1426    /// How many values the dictionary holds.
1427    values: usize,
1428    /// Where each value ends inside its payload block, packed at `offset_bits` in runs of
1429    /// [`TEXT_OFFSET_RUN`].
1430    ///
1431    /// Ends rather than starts, because then a block of 1,024 values is 1,024 numbers rather than
1432    /// 1,025: the start of a value is the end of the one before it, and the first value of a block
1433    /// starts at zero by construction. Relative to the block rather than to the payload, because a
1434    /// reader decodes a whole block and slices it, so an offset into the payload is a number it
1435    /// would have to subtract a base from anyway.
1436    offsets: Vec<u8>,
1437    /// Bits one offset is packed at, which is what the largest block of this column spans and is the
1438    /// same for every block of it.
1439    offset_bits: usize,
1440    /// How many entries the sorted order has, which is the value count.
1441    ranks: usize,
1442    /// Where the sorted order starts in the file. It is read a block at a time and only when
1443    /// something searches it, so a query that never compares this column against a literal never
1444    /// touches it at all.
1445    rank_at: u64,
1446    /// Where each block of the sorted order ends, as a byte offset from `rank_at`. A block is packed
1447    /// at whatever width its own heads need, so unlike the entries it replaced its length is not
1448    /// arithmetic on the block number.
1449    rank_ends: Vec<u64>,
1450    rank_hashes: Vec<u64>,
1451    rank_blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1452    /// Bits one code is packed at, which is what the value count needs and is the same for every
1453    /// block of the column.
1454    code_bits: usize,
1455    /// The sorted order turned round, built the first time a reader asks for it.
1456    ///
1457    /// Four bytes per value against the four the offsets already hold, so a column that has this is
1458    /// carrying half again what it carried before rather than something of a new order. It is built
1459    /// only when something asks, which is a grouped min or max over this column and nothing else,
1460    /// and that reader was going to read the payload of this column once per row otherwise.
1461    code_ranks: OnceLock<Option<Vec<u32>>>,
1462    payload: u64,
1463    /// Where each block of the payload ends in the file, as a byte offset from `payload`. The
1464    /// blocks are stored back to back, so a block starts where the one before it ended.
1465    ends: Vec<u64>,
1466    hashes: Vec<u64>,
1467    /// The payload, read and decoded a block at a time and kept after that.
1468    blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1469}
1470
1471/// How many values of a dictionary go in one block of the payload.
1472///
1473/// The block is the unit the string cascade encodes, the unit a checksum covers, and the unit a
1474/// reader has to decode to get at a single value, so it is the one number the payload format turns
1475/// on. Blocking by values rather than by bytes is what keeps a value out of two blocks at once: the
1476/// block holding a code is `code / TEXT_PAYLOAD_VALUES` and nothing has to be stitched.
1477///
1478/// A probe on the five ClickBench columns that have a dictionary worth the name, written up on
1479/// #347, measured the ratio and the decode speed at 128, 256, 512, 1,024 and 4,096 values. Both get
1480/// better all the way up, because front coding and the LZ matcher have more to look back at and
1481/// because the per chunk setup is spread over more values. What stops it is the point read: a query
1482/// that wants ten values has to decode ten blocks, so the block is what a lookup costs. At 1,024
1483/// values a block is between 67 KB and 394 KB decoded across those five columns, and the ratios are
1484/// 2.3 to 4.5. Going up to 4,096 buys two to six percent more and makes a block as much as 1.5 MB.
1485/// Going down to 512 gives up five to nine percent.
1486const TEXT_PAYLOAD_VALUES: usize = 1024;
1487
1488/// How many offsets go in one packed run.
1489///
1490/// A payload block holds 1,024 values and `bitpack::pack_tail` takes fewer than 1,024 at a time,
1491/// since a whole unit of that many belongs in the transposed layout instead. So the offsets of a
1492/// block go in two runs. Five hundred and twelve values at any width is a whole number of bytes, so
1493/// a run starts where a multiply says it does and nothing is padded.
1494const TEXT_OFFSET_RUN: usize = 512;
1495
1496/// Bytes at the front of a global dictionary index: the value count, the values a payload block
1497/// holds, the block count and the bits an offset is packed at.
1498const DICTIONARY_HEADER: usize = 16;
1499
1500/// How many entries of a dictionary's sorted order sit in one block that is read and checked as a
1501/// unit.
1502///
1503/// Five hundred and twelve entries is between two and three kilobytes on the ClickBench string
1504/// columns, which is well under a page. A binary search over half a million entries makes nineteen
1505/// probes, and the first ten land in ten different blocks while the last nine land in the one block
1506/// that holds the answer, so the whole search reads about thirty kilobytes of a megabyte of order. A
1507/// smaller block would save a little on the early probes, cost a checksum and an end list four times
1508/// as long, and give the heads less to share a base with. A larger one would read more than it uses
1509/// on every probe.
1510const TEXT_RANK_BLOCK: usize = 512;
1511
1512/// Bytes at the front of a rank block, which is the base of its heads and the width they are packed
1513/// at.
1514///
1515/// An entry used to be twelve bytes flat, eight for the head and four for the code, and on the five
1516/// ClickBench columns that have a dictionary worth the name that was 744 MB of a 12.2 GB file. Both
1517/// halves of it are nearly empty. The heads are the first eight bytes of the values in sorted order,
1518/// so a block of five hundred and twelve of them spans a tiny slice of the column, and on a column of
1519/// URLs they are all `http://w` and the block holds one distinct head. The codes are positions in a
1520/// dictionary of eighteen million, which is twenty five bits and not thirty two.
1521///
1522/// So a block now writes the smallest head in it, the bits the largest is above that, and the heads
1523/// and the codes packed at the width each needs. A block where every head agrees costs nine bytes
1524/// and the codes.
1525const RANK_BLOCK_HEADER: usize = size_of::<u64>() + 1;
1526
1527impl NativeText {
1528    /// One block of the payload, read and decoded the first time anything asks for a value in it.
1529    ///
1530    /// The bytes handed back are the values of the block laid end to end, which is what the offsets
1531    /// describe, so a caller slices it with the offsets it already has. Where the block sits in the
1532    /// file is the only thing the caller cannot work out for itself, because the stored form is
1533    /// shorter than the decoded one and by a different amount in every block.
1534    fn payload_block(&self, block: usize) -> Result<Option<&[u8]>> {
1535        let Some(slot) = self.blocks.get(block) else { return Ok(None) };
1536        let bytes = slot
1537            .get_or_init(|| {
1538                let start = if block == 0 { 0 } else { self.ends[block - 1] };
1539                let end = self.ends[block];
1540                let len = end
1541                    .checked_sub(start)
1542                    .ok_or_else(|| invalid("global dictionary block ends before it starts"))?;
1543                let mut stored = vec![
1544                    0;
1545                    usize::try_from(len).map_err(|_| invalid(
1546                        "global dictionary block does not fit in memory"
1547                    ))?
1548                ];
1549                read_at(&self.file, self.payload + start, &mut stored)?;
1550                if checksum(&stored) != self.hashes[block] {
1551                    return Err(invalid("global dictionary payload checksum differs"));
1552                }
1553                let first = block * TEXT_PAYLOAD_VALUES;
1554                let last = (first + TEXT_PAYLOAD_VALUES).min(self.values);
1555                let want = self.end_within(last - 1)? as usize;
1556                let values = string::decode_flat(&stored)?;
1557                if values.len() != last - first {
1558                    return Err(invalid("global dictionary block holds the wrong value count"));
1559                }
1560                let bytes = values.into_bytes();
1561                if bytes.len() != want {
1562                    return Err(invalid("global dictionary block decodes to the wrong length"));
1563                }
1564                Ok(bytes)
1565            })
1566            .as_ref()
1567            .map_err(Clone::clone)?;
1568        Ok(Some(bytes.as_slice()))
1569    }
1570
1571    /// Where the value at `index` ends inside its payload block.
1572    fn end_within(&self, index: usize) -> Result<u32> {
1573        let run = index / TEXT_OFFSET_RUN;
1574        let bytes = self
1575            .offsets
1576            .get(run * TEXT_OFFSET_RUN / 8 * self.offset_bits..)
1577            .ok_or_else(|| invalid("global dictionary offsets are short"))?;
1578        let end = bitpack::tail_at(bytes, self.offset_bits, index % TEXT_OFFSET_RUN)
1579            .map_err(|_| invalid("global dictionary offsets are short"))?;
1580        u32::try_from(end).map_err(|_| invalid("global dictionary offset is past the payload"))
1581    }
1582
1583    /// Where the value at `index` starts inside its payload block, which is where the value before
1584    /// it ended unless it is the first of the block.
1585    fn start_within(&self, index: usize) -> Result<u32> {
1586        if index % TEXT_PAYLOAD_VALUES == 0 { Ok(0) } else { self.end_within(index - 1) }
1587    }
1588
1589    /// Where the value at `index` starts and ends inside its payload block.
1590    fn span_within(&self, index: usize) -> Result<(u32, u32)> {
1591        let end = self.end_within(index)?;
1592        let start = self.start_within(index)?;
1593        if start > end {
1594            return Err(invalid("global dictionary value ends before it starts"));
1595        }
1596        Ok((start, end))
1597    }
1598
1599    /// The block of the sorted order that holds `rank`, and where in it that rank sits.
1600    ///
1601    /// The block is read from the file and checked against the hash the index carries for it the
1602    /// first time anything asks, and kept after that, the same way a payload block is. A search
1603    /// makes about as many probes as the order has bits, so the whole search reads a handful of
1604    /// these and never the rest.
1605    fn rank_parts(&self, rank: usize) -> Result<(&[u8], usize)> {
1606        let slot = self
1607            .rank_blocks
1608            .get(rank / TEXT_RANK_BLOCK)
1609            .ok_or_else(|| invalid("global dictionary rank is past the order"))?;
1610        let block = slot
1611            .get_or_init(|| {
1612                let which = rank / TEXT_RANK_BLOCK;
1613                let start = if which == 0 { 0 } else { self.rank_ends[which - 1] };
1614                let end = self.rank_ends[which];
1615                let mut bytes = vec![0; (end - start) as usize];
1616                read_at(&self.file, self.rank_at + start, &mut bytes)?;
1617                if checksum(&bytes)
1618                    != *self
1619                        .rank_hashes
1620                        .get(rank / TEXT_RANK_BLOCK)
1621                        .ok_or_else(|| invalid("global dictionary rank block has no checksum"))?
1622                {
1623                    return Err(invalid("global dictionary rank checksum differs"));
1624                }
1625                Ok(bytes)
1626            })
1627            .as_ref()
1628            .map_err(Clone::clone)?;
1629        Ok((block.as_slice(), rank % TEXT_RANK_BLOCK))
1630    }
1631
1632    /// The first eight bytes of the value at `rank`, as the integer a comparison reads.
1633    fn head_at(&self, rank: usize) -> Result<u64> {
1634        let (block, within) = self.rank_parts(rank)?;
1635        let (base, width, packed) = rank_heads(block)?;
1636        let above = bitpack::tail_at(packed, width, within)
1637            .map_err(|_| invalid("global dictionary rank block is short of heads"))?;
1638        Ok(base.wrapping_add(above))
1639    }
1640
1641    /// The packed codes of one rank block, which follow the heads on the next byte boundary.
1642    fn rank_codes<'block>(&self, block: &'block [u8], count: usize) -> Result<&'block [u8]> {
1643        let (_, width, packed) = rank_heads(block)?;
1644        packed
1645            .get(bitpack::tail_len(count, width)..)
1646            .ok_or_else(|| invalid("global dictionary rank block is short of codes"))
1647    }
1648
1649    /// How many entries the block holding `rank` has, which is a full block except at the end.
1650    fn rank_block_len(&self, rank: usize) -> usize {
1651        let first = rank / TEXT_RANK_BLOCK * TEXT_RANK_BLOCK;
1652        TEXT_RANK_BLOCK.min(self.ranks - first)
1653    }
1654}
1655
1656/// The base, the width and the packed bytes of one rank block's heads.
1657fn rank_heads(block: &[u8]) -> Result<(u64, usize, &[u8])> {
1658    let header = block
1659        .get(..RANK_BLOCK_HEADER)
1660        .ok_or_else(|| invalid("global dictionary rank block is short"))?;
1661    let base = u64::from_le_bytes(header[..8].try_into().expect("eight bytes"));
1662    let width = header[8] as usize;
1663    if width > 64 {
1664        return Err(invalid("global dictionary rank block packs heads past a word"));
1665    }
1666    Ok((base, width, &block[RANK_BLOCK_HEADER..]))
1667}
1668
1669/// Bits one offset of a dictionary takes, which is what its widest payload block spans.
1670///
1671/// One width for the whole column rather than one a block. A block is 1,024 values of the same
1672/// column, so the blocks of a column are within a factor of two of each other on every ClickBench
1673/// string column, and a width a block would save a fraction of a bit and cost a byte a block plus
1674/// the arithmetic that finds where a block starts.
1675fn offset_width(offsets: &[u32]) -> usize {
1676    let values = offsets.len() - 1;
1677    let mut span = 0;
1678    for first in (0..values).step_by(TEXT_PAYLOAD_VALUES) {
1679        let last = (first + TEXT_PAYLOAD_VALUES).min(values);
1680        span = span.max(offsets[last] - offsets[first]);
1681    }
1682    (u32::BITS - span.leading_zeros()) as usize
1683}
1684
1685/// How many bytes `values` offsets take at `bits`, which is what the reader has to know before it
1686/// has read any of them.
1687fn offset_bytes(values: usize, bits: usize) -> usize {
1688    let full = values / TEXT_OFFSET_RUN;
1689    let rest = values % TEXT_OFFSET_RUN;
1690    full * TEXT_OFFSET_RUN / 8 * bits + bitpack::tail_len(rest, bits)
1691}
1692
1693/// The end of every value within its payload block, packed a run at a time.
1694fn encode_offsets(offsets: &[u32], bits: usize, out: &mut Vec<u8>) -> Result<()> {
1695    let values = offsets.len() - 1;
1696    let mut run = Vec::with_capacity(TEXT_OFFSET_RUN);
1697    for first in (0..values).step_by(TEXT_OFFSET_RUN) {
1698        let last = (first + TEXT_OFFSET_RUN).min(values);
1699        let base = offsets[first / TEXT_PAYLOAD_VALUES * TEXT_PAYLOAD_VALUES];
1700        run.clear();
1701        run.extend((first..last).map(|value| u64::from(offsets[value + 1] - base)));
1702        bitpack::pack_tail(&run, bits, out)
1703            .map_err(|_| invalid("global dictionary offsets do not pack"))?;
1704    }
1705    Ok(())
1706}
1707
1708/// How many bits a code of a dictionary of `values` entries takes.
1709fn code_width(values: usize) -> usize {
1710    match u64::try_from(values).unwrap_or(u64::MAX) {
1711        0 | 1 => 0,
1712        last => (u64::BITS - (last - 1).leading_zeros()) as usize,
1713    }
1714}
1715
1716impl TextSource for NativeText {
1717    fn len(&self) -> usize {
1718        self.values
1719    }
1720
1721    fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
1722        if index >= self.values {
1723            return Ok(None);
1724        }
1725        let (start, end) = self.span_within(index)?;
1726        if start == end {
1727            return Ok(Some(&[]));
1728        }
1729        // A block holds a fixed number of values rather than a fixed number of bytes, so the value
1730        // is in one block and the offsets already say where in it.
1731        let block = index / TEXT_PAYLOAD_VALUES;
1732        let Some(bytes) = self.payload_block(block)? else { return Ok(None) };
1733        Ok(bytes.get(start as usize..end as usize))
1734    }
1735
1736    fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
1737        if index >= self.values {
1738            return Ok(None);
1739        }
1740        let (start, end) = self.span_within(index)?;
1741        Ok(Some((end - start) as usize))
1742    }
1743
1744    fn ranks(&self) -> Option<usize> {
1745        (self.ranks > 0).then_some(self.ranks)
1746    }
1747
1748    fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
1749        // The head settles the probe unless the two values start with the same eight bytes, and
1750        // only then is a value read. On a column of URLs that is the difference between a search
1751        // that touches one block of the payload and a search that touches nineteen of them.
1752        let settled = self.head_at(rank)?.cmp(&head(wanted));
1753        if settled != Ordering::Equal {
1754            return Ok(settled);
1755        }
1756        let code = self.code_at_rank(rank)?;
1757        let bytes = self
1758            .bytes_at(code as usize)?
1759            .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
1760        Ok(bytes.cmp(wanted))
1761    }
1762
1763    fn code_at_rank(&self, rank: usize) -> Result<u32> {
1764        let (block, within) = self.rank_parts(rank)?;
1765        let codes = self.rank_codes(block, self.rank_block_len(rank))?;
1766        let code = bitpack::tail_at(codes, self.code_bits, within)
1767            .map_err(|_| invalid("global dictionary rank block is short of codes"))?;
1768        let code = u32::try_from(code)
1769            .map_err(|_| invalid("global dictionary order names a code it does not have"))?;
1770        if code as usize >= self.len() {
1771            return Err(invalid("global dictionary order names a code it does not have"));
1772        }
1773        Ok(code)
1774    }
1775
1776    fn code_ranks(&self) -> Option<&[u32]> {
1777        // The order is a permutation of the positions, so inverting it needs every position to be
1778        // named exactly once. Anything else and the slice would have holes, and a caller indexing
1779        // it by a code would read a rank that belongs to nothing.
1780        if self.ranks == 0 || self.ranks != self.len() {
1781            return None;
1782        }
1783        self.code_ranks
1784            .get_or_init(|| {
1785                let mut ranks = vec![u32::MAX; self.ranks];
1786                // A block at a time rather than a rank at a time, because reading it per rank pays
1787                // for the bounds check, the division and the lock on every one of them.
1788                for first in (0..self.ranks).step_by(TEXT_RANK_BLOCK) {
1789                    let (block, _) = self.rank_parts(first).ok()?;
1790                    let count = self.rank_block_len(first);
1791                    let codes = self.rank_codes(block, count).ok()?;
1792                    for (within, code) in bitpack::unpack_tail(codes, self.code_bits, count)
1793                        .ok()?
1794                        .into_iter()
1795                        .enumerate()
1796                    {
1797                        let code = usize::try_from(code).ok()?;
1798                        *ranks.get_mut(code)? = u32::try_from(first + within).ok()?;
1799                    }
1800                }
1801                if ranks.contains(&u32::MAX) {
1802                    return None;
1803                }
1804                Some(ranks)
1805            })
1806            .as_deref()
1807    }
1808
1809    fn footprint(&self) -> usize {
1810        self.offsets.capacity()
1811            + self
1812                .code_ranks
1813                .get()
1814                .and_then(Option::as_ref)
1815                .map_or(0, |ranks| ranks.capacity() * size_of::<u32>())
1816            + self.rank_hashes.capacity() * size_of::<u64>()
1817            + self.rank_ends.capacity() * size_of::<u64>()
1818            + self.rank_blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1819            + self
1820                .rank_blocks
1821                .iter()
1822                .filter_map(OnceLock::get)
1823                .filter_map(|result| result.as_ref().ok())
1824                .map(Vec::capacity)
1825                .sum::<usize>()
1826            + self.blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1827            + self.hashes.capacity() * size_of::<u64>()
1828            + self.ends.capacity() * size_of::<u64>()
1829            + self
1830                .blocks
1831                .iter()
1832                .filter_map(OnceLock::get)
1833                .filter_map(|result| result.as_ref().ok())
1834                .map(Vec::capacity)
1835                .sum::<usize>()
1836    }
1837}
1838
1839/// Every table wide part number in order, with the stripe it belongs to.
1840fn places(table: &Table) -> Result<Vec<Place>> {
1841    let mut places = Vec::with_capacity(table.stripes.len().saturating_mul(STRIPE_PARTS));
1842    for (at, stripe) in table.stripes.iter().enumerate() {
1843        let index = u32::try_from(at).map_err(|_| invalid("too many stripes"))?;
1844        for (part, &rows) in stripe.parts.iter().enumerate() {
1845            places.push(Place {
1846                stripe: index,
1847                part: u32::try_from(part).map_err(|_| invalid("too many parts in a stripe"))?,
1848                rows,
1849            });
1850        }
1851    }
1852    Ok(places)
1853}
1854
1855/// Reads one column's section of a stripe's index page.
1856///
1857/// The section carries its own checksum, so a reader that wants one column out of a hundred and
1858/// five preads a few hundred bytes and still knows that what it got is what was written.
1859fn read_index(file: &File, stripe: &Stripe, column: usize) -> Result<Vec<PartSpan>> {
1860    let parts = stripe.parts.len();
1861    let section = index_section(parts)?;
1862    let at = column.checked_mul(section).ok_or_else(|| invalid("index page offset overflow"))?;
1863    let end = at.checked_add(section).ok_or_else(|| invalid("index page offset overflow"))?;
1864    if end > stripe.index.length as usize {
1865        return Err(invalid("index page is shorter than its columns"));
1866    }
1867    let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1868    let mut bytes = vec![0; section];
1869    let offset = stripe
1870        .index
1871        .offset
1872        .checked_add(at as u64)
1873        .ok_or_else(|| invalid("index page offset overflow"))?;
1874    read_at(file, offset, &mut bytes)?;
1875    let entries = section - size_of::<u64>();
1876    let stored = u64::from_le_bytes(bytes[entries..].try_into().expect("eight bytes"));
1877    if checksum(&bytes[..entries]) != stored {
1878        // With where it was read from, because the two ways this fires look identical from the
1879        // message alone: a file somebody damaged, and a file we wrote to the wrong offset.
1880        return Err(invalid(&format!(
1881            "index page section checksum differs, column {column} of {parts} parts at {offset}, \
1882             wanted {stored:016x} and got {:016x}",
1883            checksum(&bytes[..entries]),
1884        )));
1885    }
1886    let mut spans = Vec::with_capacity(parts);
1887    let mut start = 0_usize;
1888    for part in 0..parts {
1889        let at = part * INDEX_ENTRY;
1890        let length = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four bytes")) as usize;
1891        let hash = u64::from_le_bytes(bytes[at + 4..at + 12].try_into().expect("eight bytes"));
1892        spans.push(PartSpan { start, length, hash });
1893        start = start.checked_add(length).ok_or_else(|| invalid("column page length overflow"))?;
1894    }
1895    if start != page.length as usize {
1896        return Err(invalid("column page length differs from its index"));
1897    }
1898    Ok(spans)
1899}
1900
1901/// One part's bytes out of a whole column page.
1902fn part_bytes(page: &[u8], span: PartSpan) -> Result<&[u8]> {
1903    let end = span.start.checked_add(span.length).ok_or_else(|| invalid("part range overflow"))?;
1904    page.get(span.start..end).ok_or_else(|| invalid("part exceeds its column page"))
1905}
1906
1907/// Puts one stripe of one column in the cache, dropping the stripe that has been there longest.
1908///
1909/// The index goes in its own slot and stays. Only the page is under the budget, and `kept` is how
1910/// many pages that budget is.
1911fn remember(cached: &mut Cached, held: &CachedColumn, kept: usize) {
1912    if let Some(slot) = cached.index.get_mut(held.stripe) {
1913        if slot.is_none() {
1914            *slot = Some(Arc::clone(&held.index));
1915        }
1916    }
1917    let Some(page) = held.page.clone() else { return };
1918    let Some(slot) = cached.pages.get_mut(held.stripe) else { return };
1919    if slot.is_none() {
1920        cached.order.push_back(held.stripe);
1921    }
1922    *slot = Some(page);
1923    while cached.order.len() > kept.max(1) {
1924        let Some(oldest) = cached.order.pop_front() else { break };
1925        if let Some(slot) = cached.pages.get_mut(oldest) {
1926            *slot = None;
1927        }
1928    }
1929}
1930
1931impl Reader {
1932    /// Opens the highest valid directory slot.
1933    ///
1934    /// # Errors
1935    ///
1936    /// If the file has no valid committed directory or a directory pointer is out of bounds.
1937    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1938        let mut file = File::open(path).map_err(io)?;
1939        let size = file.metadata().map_err(io)?.len();
1940        if size < HEADER {
1941            return Err(invalid("file is shorter than its header"));
1942        }
1943        let mut header = [0; HEADER as usize];
1944        file.read_exact(&mut header).map_err(io)?;
1945        let mut opening = Opening { reads: 1, bytes: HEADER };
1946        let version = u32::from_le_bytes([header[8], header[9], header[10], header[11]]);
1947        // The two halves are worth telling apart. A wrong magic is a file that was never ours and
1948        // the answer is to look at the path. A wrong version is our own file from another build,
1949        // and the number this build wants is the only thing that tells the reader whether to
1950        // rebuild the file or to go back to the binary that wrote it.
1951        if &header[..8] != MAGIC {
1952            return Err(invalid("the header does not begin with a rudb native magic"));
1953        }
1954        if version != FORMAT {
1955            return Err(invalid(&format!(
1956                "the file is format {version} and this build reads format {FORMAT}, so it has to \
1957                 be written again"
1958            )));
1959        }
1960        let mut selected = None;
1961        for start in [16, 16 + SLOT_BYTES] {
1962            let slot = Slot::read(&header[start..start + SLOT_BYTES]);
1963            if slot.generation == 0 || slot.length == 0 || slot.length as usize > MAX_DIRECTORY {
1964                continue;
1965            }
1966            let Some(end) = slot.offset.checked_add(u64::from(slot.length)) else { continue };
1967            if slot.offset < HEADER || end > size {
1968                continue;
1969            }
1970            let mut bytes = vec![0; slot.length as usize];
1971            file.seek(SeekFrom::Start(slot.offset)).map_err(io)?;
1972            file.read_exact(&mut bytes).map_err(io)?;
1973            opening.reads += 1;
1974            opening.bytes += u64::from(slot.length);
1975            if checksum(&bytes) == slot.hash
1976                && selected
1977                    .as_ref()
1978                    .is_none_or(|(old, _): &(Slot, Vec<u8>)| old.generation < slot.generation)
1979            {
1980                selected = Some((slot, bytes));
1981            }
1982        }
1983        let (slot, bytes) =
1984            selected.ok_or_else(|| invalid("no committed directory slot is valid"))?;
1985        let table = decode_directory(&bytes, size)?;
1986        let places = places(&table)?;
1987        let dictionaries = (0..table.fields.len()).map(|_| OnceLock::new()).collect();
1988        let table_fields = table.fields.len();
1989        let stripes = table.stripes.len();
1990        let cache = (0..table.fields.len())
1991            .map(|_| {
1992                Mutex::new(Cached {
1993                    pages: (0..stripes).map(|_| None).collect(),
1994                    index: (0..stripes).map(|_| None).collect(),
1995                    ..Cached::default()
1996                })
1997            })
1998            .collect::<Vec<_>>();
1999        let sieves = (0..table.fields.len())
2000            .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
2001            .collect();
2002        Ok(Self {
2003            file: Arc::new(file),
2004            table: Arc::new(table),
2005            dictionaries: Arc::new(dictionaries),
2006            loading: Arc::new((0..table_fields).map(|_| Mutex::new(())).collect()),
2007            opened: Arc::new(AtomicUsize::new(0)),
2008            sieves: Arc::new(sieves),
2009            places: Arc::new(places),
2010            cache: Arc::new(cache),
2011            pages: Arc::new(AtomicUsize::new(0)),
2012            indexes: Arc::new(AtomicUsize::new(0)),
2013            kept: Arc::new(AtomicUsize::new(CACHED_STRIPES_PER_COLUMN)),
2014            size,
2015            directory: u64::from(slot.length),
2016            opening,
2017        })
2018    }
2019
2020    /// What this reader has read so far, and what opening it cost.
2021    ///
2022    /// Public because the claim of `spec/stats/04-in-memory.md` section 4.2 is about this number
2023    /// and a claim nobody can check is a comment. A caller that wants to know whether opening a
2024    /// file touched the data asks here, and gets an answer that does not depend on what the page
2025    /// cache happened to hold.
2026    #[must_use]
2027    pub fn reads(&self) -> Reads {
2028        Reads {
2029            opening: self.opening,
2030            pages: self.pages.load(Atomic::Relaxed),
2031            indexes: self.indexes.load(Atomic::Relaxed),
2032            dictionaries: self.opened.load(Atomic::Relaxed),
2033        }
2034    }
2035
2036    /// Where the file's bytes went, from the directory alone.
2037    ///
2038    /// No page is read, so this costs the same on a 45 GB table as on an empty one. See [`Layout`]
2039    /// for what is charged where and for why the three things that are not columns stay separate.
2040    #[must_use]
2041    pub fn layout(&self) -> Layout {
2042        let table = &self.table;
2043        let stripes = table.stripes.as_slice();
2044        let columns = table
2045            .fields
2046            .iter()
2047            .enumerate()
2048            .map(|(at, field)| ColumnLayout {
2049                name: field.name.clone(),
2050                kind: field.ty.to_string(),
2051                pages: sum(stripes.iter().map(|stripe| span_bytes(&stripe.pages, at))),
2052                memberships: sum(stripes.iter().map(|stripe| page_bytes(&stripe.memberships, at))),
2053                sieves: sum(stripes.iter().map(|stripe| page_bytes(&stripe.sieves, at))),
2054                dictionary: page_bytes(&table.dictionaries, at),
2055            })
2056            .collect();
2057        Layout {
2058            file: self.size,
2059            rows: table.rows,
2060            stripes: stripes.len(),
2061            parts: self.places.len(),
2062            columns,
2063            indexes: sum(stripes.iter().map(|stripe| u64::from(stripe.index.length))),
2064            directory: self.directory,
2065            header: HEADER,
2066        }
2067    }
2068
2069    /// How many parts the table has, which is how many chunks a scan of it reads.
2070    #[must_use]
2071    pub fn parts(&self) -> usize {
2072        self.places.len()
2073    }
2074
2075    /// The parts of each stripe, in table wide part numbers.
2076    ///
2077    /// A scan that wants one worker to own the page it reads hands work out in these runs. The
2078    /// stripes are contiguous in part numbering and all but the last hold sixty four parts, but a
2079    /// stripe can be flushed early when rows arrive out of order, so the runs are read off the
2080    /// directory rather than worked out from a constant.
2081    #[must_use]
2082    pub fn stripe_parts(&self) -> Vec<std::ops::Range<usize>> {
2083        let mut runs = Vec::with_capacity(self.table.stripes.len());
2084        let mut start = 0;
2085        for stripe in &self.table.stripes {
2086            let end = start + stripe.parts.len();
2087            runs.push(start..end);
2088            start = end;
2089        }
2090        runs
2091    }
2092
2093    /// Asks the page cache to keep `stripes` stripes of every column instead of the default.
2094    ///
2095    /// This only ever raises the number. A scan that gives each worker a whole stripe has one page
2096    /// per column per worker open at once, and a cache smaller than that is worse than no cache at
2097    /// all: every worker's page is evicted by the others before it has finished its stripe, so it
2098    /// reads a quarter of a megabyte for every part it takes out of it.
2099    pub fn keep_stripes(&self, stripes: usize) {
2100        self.kept.fetch_max(stripes, Atomic::Relaxed);
2101    }
2102
2103    /// Rows in one part, or zero when the part number is past the table.
2104    #[must_use]
2105    pub fn part_rows(&self, at: usize) -> usize {
2106        self.places.get(at).map_or(0, |place| place.rows as usize)
2107    }
2108
2109    /// The committed table directory.
2110    #[must_use]
2111    pub fn table(&self) -> &Table {
2112        &self.table
2113    }
2114
2115    /// Exact leading frequencies when the stored synopsis proves a count-descending prefix.
2116    ///
2117    /// The returned list can be longer than `top`. Keeping the stored tail lets a later TopN apply
2118    /// additional ordering keys without losing a value tied with the requested boundary.
2119    ///
2120    /// # Errors
2121    ///
2122    /// If the column is outside the schema or a stored value does not fit its declared type.
2123    pub fn top_frequencies(&self, column: usize, top: usize) -> Result<Option<Vec<(Value, u64)>>> {
2124        let field = self
2125            .table
2126            .fields
2127            .get(column)
2128            .ok_or_else(|| invalid("frequency column index out of range"))?;
2129        let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2130            return Ok(None);
2131        };
2132        if top == 0 || summary.entries.len() < top {
2133            return Ok(None);
2134        }
2135        let boundary = summary.entries[top - 1].count;
2136        if boundary <= summary.omitted_max {
2137            return Ok(None);
2138        }
2139        self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
2140    }
2141
2142    /// Every value of one column with the number of rows holding it, when the synopsis is complete.
2143    ///
2144    /// The heavy hitter pass keeps a bounded set of candidates and decrements them all when it runs
2145    /// out of room, so what it usually ends with is the leading values and a bound on everything it
2146    /// dropped. `omitted_max` of zero says that never happened: no candidate was ever decremented and
2147    /// the entries did not overflow the stored budget, so the list is every distinct value of the
2148    /// column with an exact count, and a null counts as a value of its own rather than being skipped.
2149    ///
2150    /// That makes a whole class of question answerable without reading a row. How many rows hold a
2151    /// value, how many do not, and what a `GROUP BY` of that column with a count over it produces are
2152    /// all in here. It is only ever true of a column with few enough distinct values, which is the
2153    /// case worth having, because that is exactly the column a grouping or an equality filter would
2154    /// otherwise walk every row to answer.
2155    ///
2156    /// `None` when the column has no synopsis, or has one that dropped anything.
2157    ///
2158    /// # Errors
2159    ///
2160    /// If the column is outside the schema or a stored value does not fit its declared type.
2161    pub fn exact_frequencies(&self, column: usize) -> Result<Option<Vec<(Value, u64)>>> {
2162        let field = self
2163            .table
2164            .fields
2165            .get(column)
2166            .ok_or_else(|| invalid("frequency column index out of range"))?;
2167        let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2168            return Ok(None);
2169        };
2170        if summary.omitted_max > 0 {
2171            return Ok(None);
2172        }
2173        self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
2174    }
2175
2176    /// Turns stored frequency entries into values of the column's own type.
2177    fn decode_frequencies(
2178        &self,
2179        column: usize,
2180        ty: &LogicalType,
2181        entries: &[FrequencyEntry],
2182    ) -> Result<Vec<(Value, u64)>> {
2183        let dictionary = if *ty == LogicalType::Varchar { self.dictionary(column)? } else { None };
2184        let mut out = Vec::with_capacity(entries.len());
2185        for entry in entries {
2186            let value = match entry.value {
2187                FrequencyValue::Null => Value::Null,
2188                FrequencyValue::Integer(value) => match *ty {
2189                    LogicalType::TinyInt => Value::TinyInt(
2190                        i8::try_from(value)
2191                            .map_err(|_| invalid("frequency TINYINT is out of range"))?,
2192                    ),
2193                    LogicalType::UTinyInt => Value::UTinyInt(
2194                        u8::try_from(value)
2195                            .map_err(|_| invalid("frequency UTINYINT is out of range"))?,
2196                    ),
2197                    LogicalType::USmallInt => Value::USmallInt(
2198                        u16::try_from(value)
2199                            .map_err(|_| invalid("frequency USMALLINT is out of range"))?,
2200                    ),
2201                    LogicalType::UInteger => Value::UInteger(
2202                        u32::try_from(value)
2203                            .map_err(|_| invalid("frequency UINTEGER is out of range"))?,
2204                    ),
2205                    LogicalType::UBigInt => Value::UBigInt(
2206                        u64::try_from(value)
2207                            .map_err(|_| invalid("frequency UBIGINT is out of range"))?,
2208                    ),
2209                    LogicalType::SmallInt => Value::SmallInt(
2210                        i16::try_from(value)
2211                            .map_err(|_| invalid("frequency SMALLINT is out of range"))?,
2212                    ),
2213                    LogicalType::Integer => Value::Integer(
2214                        i32::try_from(value)
2215                            .map_err(|_| invalid("frequency INTEGER is out of range"))?,
2216                    ),
2217                    LogicalType::BigInt => Value::BigInt(
2218                        i64::try_from(value)
2219                            .map_err(|_| invalid("frequency BIGINT is out of range"))?,
2220                    ),
2221                    LogicalType::Date => Value::Date(
2222                        i32::try_from(value)
2223                            .map_err(|_| invalid("frequency DATE is out of range"))?,
2224                    ),
2225                    LogicalType::Timestamp => Value::Timestamp(
2226                        i64::try_from(value)
2227                            .map_err(|_| invalid("frequency TIMESTAMP is out of range"))?,
2228                    ),
2229                    _ => return Err(invalid("integer frequency belongs to another type")),
2230                },
2231                FrequencyValue::Code(code) => dictionary
2232                    .as_ref()
2233                    .ok_or_else(|| invalid("frequency code has no dictionary"))?
2234                    .try_value_at(code as usize)?,
2235            };
2236            out.push((value, entry.count));
2237        }
2238        Ok(out)
2239    }
2240
2241    /// Sparse rows belonging to the bounded numeric frequency candidate set.
2242    ///
2243    /// The list is omitted when collecting it would exceed the fixed storage budget. A composite
2244    /// aggregate may accept a result over these rows only when its requested boundary is strictly
2245    /// greater than `omitted_max`.
2246    ///
2247    /// # Errors
2248    ///
2249    /// If the column is outside the schema.
2250    pub fn frequency_occurrences(&self, column: usize) -> Result<Option<FrequencyOccurrences>> {
2251        self.table
2252            .fields
2253            .get(column)
2254            .ok_or_else(|| invalid("frequency column index out of range"))?;
2255        let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2256            return Ok(None);
2257        };
2258        if summary.ordinals.is_empty() {
2259            return Ok(None);
2260        }
2261        Ok(Some(FrequencyOccurrences {
2262            omitted_max: summary.omitted_max,
2263            ordinals: summary.ordinals.clone(),
2264        }))
2265    }
2266
2267    /// How many distinct values one column holds, counting a null as no value.
2268    ///
2269    /// A string column of this format is written against one dictionary that covers the whole table.
2270    /// A code is handed out the first time a value is seen and nothing ever removes one, so the
2271    /// number of codes is the number of distinct values exactly rather than an estimate. That makes
2272    /// `COUNT(DISTINCT column)` over a whole table a question the directory already knows the answer
2273    /// to, and the alternative is a hash table with a row per distinct value built from a pass over
2274    /// every row.
2275    ///
2276    /// A null in the column used to make this `None` and no longer does. A null row is written as
2277    /// the code for the empty string, so a nullable column's dictionary can hold an empty string
2278    /// that no row of it actually has, and the dictionary on its own does not say which case it is.
2279    /// The writer does know, because it counts the non-null rows that use each code on its way to
2280    /// the frequency summary, so it records how many codes any row holds and the directory carries
2281    /// that number. This reads it rather than the size of the dictionary, which also means the
2282    /// dictionary page is not opened to answer.
2283    ///
2284    /// `None` for a column the file has no dictionary for, which is every column that is not a
2285    /// string. A sketch would answer that approximately and SQL asked for the exact number.
2286    ///
2287    /// # Errors
2288    ///
2289    /// If the column is outside the schema.
2290    pub fn distinct_values(&self, column: usize) -> Result<Option<u64>> {
2291        self.table
2292            .distincts
2293            .get(column)
2294            .copied()
2295            .ok_or_else(|| invalid("distinct column index out of range"))
2296    }
2297
2298    /// How many rows of one column are null, added up over the stripes.
2299    ///
2300    /// Every stripe records this exactly when it is written, because a null count is not a bound
2301    /// that is allowed to be wide the way a minimum and a maximum are: a filter that reads one too
2302    /// many is slow and a `COUNT` that reads one too many is wrong. Adding up a few hundred numbers
2303    /// already in memory is what makes `COUNT(column)` over a whole table free.
2304    ///
2305    /// # Errors
2306    ///
2307    /// If the column is outside the schema.
2308    pub fn null_count(&self, column: usize) -> Result<u64> {
2309        if column >= self.table.fields.len() {
2310            return Err(invalid("null count column index out of range"));
2311        }
2312        let mut nulls = 0_u64;
2313        for stripe in &self.table.stripes {
2314            let range = stripe
2315                .zone
2316                .column(column)
2317                .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2318            nulls = nulls
2319                .checked_add(range.nulls as u64)
2320                .ok_or_else(|| invalid("null count overflow"))?;
2321        }
2322        Ok(nulls)
2323    }
2324
2325    /// The smallest and the largest value of one string column, from the order beside its values.
2326    ///
2327    /// The dictionary holds exactly the values the column holds, so the first and the last of them
2328    /// in sorted order are the column's minimum and maximum. Two reads of a rank block settle what
2329    /// otherwise walks a million rows.
2330    ///
2331    /// `None` when the column is not a string, when the file was written before version 9 and so has
2332    /// no order, when the column has no values at all, or when it has a null in it, which is the
2333    /// placeholder again: the empty string a null is written as would sort ahead of every real
2334    /// value and be reported as the minimum.
2335    ///
2336    /// # Errors
2337    ///
2338    /// If the column is outside the schema, or a rank names a code the dictionary does not have.
2339    pub fn text_extremes(&self, column: usize) -> Result<Option<(Value, Value)>> {
2340        if self.null_count(column)? > 0 {
2341            return Ok(None);
2342        }
2343        let Some(dictionary) = self.dictionary(column)? else { return Ok(None) };
2344        let Some(ranks) = dictionary.ranks() else { return Ok(None) };
2345        if ranks == 0 {
2346            return Ok(None);
2347        }
2348        let low = text_at_rank(&dictionary, 0)?;
2349        let high = text_at_rank(&dictionary, ranks - 1)?;
2350        Ok(Some((low, high)))
2351    }
2352
2353    /// The smallest and the largest value of one column, when every stripe wrote exact ends.
2354    ///
2355    /// A stripe's ends are allowed to be wider than the truth, because a bound that rules out a
2356    /// chunk that could not match is still correct when it rules out nothing. That is what makes
2357    /// them cheap to write for a bit packed or a dictionary column, and it is also what stops them
2358    /// answering a `MIN`. So each stripe says which of the two it wrote, and this answers only when
2359    /// all of them walked their rows.
2360    ///
2361    /// `None` for a column with no ends, for an empty table, and for a column any stripe of which
2362    /// guessed. Nulls need no special case, because the ends skip them the same way `MIN` does.
2363    ///
2364    /// One case is given up on that did not have to be. A stripe merges the ends of its sixty four
2365    /// parts, and a part with no ends at all erases the merged ones, because a part whose rows are
2366    /// not covered by the stripe's ends is a stripe that would skip rows it should keep. A part of
2367    /// nothing but nulls has no rows to cover and so did not need to erase anything, but the merge
2368    /// cannot tell that part from a part whose layout it could not read. So a column with a chunk
2369    /// of nothing but nulls in the middle of it goes and reads the rows. That is slow and right,
2370    /// and the fix is a row count per part rather than anything here.
2371    ///
2372    /// # Errors
2373    ///
2374    /// If the column is outside the schema.
2375    pub fn exact_extremes(&self, column: usize) -> Result<Option<(Bound, Bound)>> {
2376        if column >= self.table.fields.len() {
2377            return Err(invalid("extremes column index out of range"));
2378        }
2379        let mut low: Option<Bound> = None;
2380        let mut high: Option<Bound> = None;
2381        for stripe in &self.table.stripes {
2382            let range = stripe
2383                .zone
2384                .column(column)
2385                .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2386            if !range.exact {
2387                return Ok(None);
2388            }
2389            // A stripe of nothing but nulls has no ends and says nothing about the column's, which
2390            // is why this skips it rather than giving up on the whole column. A stripe that has
2391            // rows and still has no end is a layout whose values this cannot see, and skipping that
2392            // one would answer with an end taken from the other stripes, so it gives up instead.
2393            let (Some(small), Some(large)) = (range.low.as_ref(), range.high.as_ref()) else {
2394                if stripe.rows > range.nulls {
2395                    return Ok(None);
2396                }
2397                continue;
2398            };
2399            low = Some(low.map_or_else(|| small.clone(), |held| held.smaller(small.clone())));
2400            high = Some(high.map_or_else(|| large.clone(), |held| held.larger(large.clone())));
2401        }
2402        Ok(low.zip(high))
2403    }
2404
2405    /// The sum of one integer column and how many rows went into it, when every stripe wrote one.
2406    ///
2407    /// The count beside the sum is the non-null rows, because that is what a `SUM` adds up and what
2408    /// an `AVG` divides by, and a caller that had to work it out from the row count and the null
2409    /// count would be doing the same walk twice.
2410    ///
2411    /// `None` for anything that is not an integer column, for a file written by something that did
2412    /// not record it, and when adding the stripes together would overflow.
2413    ///
2414    /// # Errors
2415    ///
2416    /// If the column is outside the schema.
2417    pub fn exact_sum(&self, column: usize) -> Result<Option<(i128, u64)>> {
2418        if column >= self.table.fields.len() {
2419            return Err(invalid("sum column index out of range"));
2420        }
2421        let mut total = 0_i128;
2422        let mut rows = 0_u64;
2423        for stripe in &self.table.stripes {
2424            let range = stripe
2425                .zone
2426                .column(column)
2427                .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2428            let Some(part) = range.sum else { return Ok(None) };
2429            let Some(sum) = total.checked_add(part) else { return Ok(None) };
2430            total = sum;
2431            rows = rows.saturating_add(stripe.rows as u64 - range.nulls as u64);
2432        }
2433        Ok(Some((total, rows)))
2434    }
2435
2436    /// The global dictionary of a column, opened once however many workers ask for it at once.
2437    ///
2438    /// The unlocked look is first because it is the answer every time after the first and it costs a
2439    /// load. Everybody who misses it queues on [`Self::loading`] and looks again on the way in, so
2440    /// the one who arrived first does the reading and the rest take what it left. Waiting is the
2441    /// cheaper thing to do: the work behind the lock is a page read, a checksum and the decode of a
2442    /// dictionary that can hold half a million entries, and the alternative is every worker of the
2443    /// scan doing all of it and all but one dropping the result on the floor.
2444    fn dictionary(&self, column: usize) -> Result<Option<Arc<Vector>>> {
2445        let Some(page) = self.table.dictionaries[column] else { return Ok(None) };
2446        if let Some(dictionary) = self.dictionaries[column].get() {
2447            return Ok(Some(Arc::clone(dictionary)));
2448        }
2449        let _queued = self.loading[column].lock().map_err(|_| invalid("a poisoned dictionary"))?;
2450        if let Some(dictionary) = self.dictionaries[column].get() {
2451            return Ok(Some(Arc::clone(dictionary)));
2452        }
2453        self.opened.fetch_add(1, Atomic::Relaxed);
2454        let dictionary = Arc::new(open_global_dictionary(
2455            Arc::clone(&self.file),
2456            page,
2457            &self.table.fields[column].ty,
2458        )?);
2459        let _ = self.dictionaries[column].set(Arc::clone(&dictionary));
2460        Ok(Some(dictionary))
2461    }
2462
2463    /// Reads only the named columns from one part.
2464    ///
2465    /// The whole stripe page each column lives in is read and kept, because a scan asks for the
2466    /// parts of a stripe one after another and this is what turns sixty four reads into one.
2467    ///
2468    /// # Errors
2469    ///
2470    /// If a part, column, page, or checksum is invalid.
2471    pub fn read(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2472        self.read_impl(part, columns, true)
2473    }
2474
2475    /// Reads named columns from one part without keeping the stripe page it came out of.
2476    ///
2477    /// This is for sparse row fetches after a selective TopN or filter, which reach a few parts of
2478    /// a stripe rather than all of them. A caller that will read most of a stripe should use
2479    /// [`Self::read`] instead, because this reads and discards the page index every time.
2480    ///
2481    /// # Errors
2482    ///
2483    /// If a part, column, page, or checksum is invalid.
2484    pub fn read_sparse(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2485        self.read_impl(part, columns, false)
2486    }
2487
2488    /// Whether an exact global-code membership index proves that the stripe holding a part cannot
2489    /// contain any of the sorted candidate codes.
2490    ///
2491    /// # Errors
2492    ///
2493    /// If the part, column, index page, checksum, or delta stream is invalid.
2494    pub fn skips_codes(&self, part: usize, column: usize, candidates: &[u32]) -> Result<bool> {
2495        if candidates.is_empty() {
2496            return Ok(true);
2497        }
2498        if candidates.windows(2).any(|pair| pair[0] >= pair[1]) {
2499            return Err(Error::internal("native code candidates are not sorted and unique"));
2500        }
2501        let stripe = self.stripe_of(part)?;
2502        let Some(page) = stripe.memberships.get(column).copied().flatten() else {
2503            return Ok(false);
2504        };
2505        let mut bytes = vec![0; page.length as usize];
2506        read_at(&self.file, page.offset, &mut bytes)?;
2507        if checksum(&bytes) != page.hash {
2508            return Err(invalid("membership page checksum differs"));
2509        }
2510        let codes = decode_membership(&bytes)?;
2511        let mut left = 0;
2512        let mut right = 0;
2513        while left < codes.len() && right < candidates.len() {
2514            match codes[left].cmp(&candidates[right]) {
2515                Ordering::Less => left += 1,
2516                Ordering::Greater => right += 1,
2517                Ordering::Equal => return Ok(false),
2518            }
2519        }
2520        Ok(true)
2521    }
2522
2523    fn stripe_of(&self, part: usize) -> Result<&Stripe> {
2524        let place = self.places.get(part).ok_or_else(|| invalid("part index out of range"))?;
2525        self.table
2526            .stripes
2527            .get(place.stripe as usize)
2528            .ok_or_else(|| invalid("stripe index out of range"))
2529    }
2530
2531    /// The page index of one column of one stripe, and its page when the caller wants all of it.
2532    ///
2533    /// A scan hands parts out in order, so every worker on a column crosses into a new stripe within
2534    /// a few parts of the others and they all want the same page at the same moment. This used to
2535    /// let all of them read it, which cost the scan as many copies of every page as it had workers.
2536    /// On the full ClickBench file a `MIN(EventDate), MAX(EventDate)` moved 3.2 GB off the disk to
2537    /// look at 400 MB of column.
2538    ///
2539    /// A worker that finds the page it wants already being read neither waits for it nor reads it
2540    /// again. It comes back with the index alone, which sends [`Reader::read_impl`] down the path
2541    /// that reads the one part it came for, a few kilobytes against a quarter of a megabyte, and it
2542    /// picks the page up from the cache on its next part. Waiting would be the other way to avoid
2543    /// the duplicate read and it is worse: the pages that matter are the wide string ones, they take
2544    /// milliseconds to copy even warm, and every other worker would be stopped for all of it.
2545    ///
2546    /// The file is never read under the lock.
2547    fn held(&self, at: usize, stripe: &Stripe, column: usize, whole: bool) -> Result<CachedColumn> {
2548        let cache = self.cache.get(column).ok_or_else(|| invalid("column index out of range"))?;
2549        let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2550        let known = cached.index.get(at).and_then(Clone::clone);
2551        let page = cached.pages.get(at).and_then(Clone::clone);
2552        if let Some(index) = known.clone() {
2553            if !whole || page.is_some() {
2554                return Ok(CachedColumn { stripe: at, index, page });
2555            }
2556        }
2557        if cached.loading.contains(&at) {
2558            drop(cached);
2559            // The index is almost always already here, because somebody read this stripe to get
2560            // into the loading list in the first place, so this branch usually costs no read at
2561            // all and the one part read in `read_impl` is all the losing worker pays for.
2562            if let Some(index) = known {
2563                return Ok(CachedColumn { stripe: at, index, page: None });
2564            }
2565            let held = self.page_of(stripe, column, at, false, None)?;
2566            let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2567            remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2568            return Ok(held);
2569        }
2570        cached.loading.push(at);
2571        drop(cached);
2572
2573        let read = self.page_of(stripe, column, at, whole, known);
2574
2575        // The stripe leaves the loading list and its page enters the cache under one lock. Doing
2576        // them separately would leave a moment where another worker sees neither and reads the
2577        // page a second time, which is the whole thing this is here to stop.
2578        let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2579        if let Some(position) = cached.loading.iter().position(|loading| *loading == at) {
2580            cached.loading.remove(position);
2581        }
2582        let held = read?;
2583        remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2584        Ok(held)
2585    }
2586
2587    /// Reads one stripe's index for a column, and its page when the caller wants all of it.
2588    ///
2589    /// `known` is the index when the reader has already read it, which after the first worker
2590    /// through a stripe it always has, because [`remember`] keeps every index for the life of the
2591    /// reader. Without that a scan reads the index again on every part that misses the page cache.
2592    fn page_of(
2593        &self,
2594        stripe: &Stripe,
2595        column: usize,
2596        at: usize,
2597        whole: bool,
2598        known: Option<Arc<Vec<PartSpan>>>,
2599    ) -> Result<CachedColumn> {
2600        let index = match known {
2601            Some(index) => index,
2602            None => {
2603                self.indexes.fetch_add(1, Atomic::Relaxed);
2604                Arc::new(read_index(&self.file, stripe, column)?)
2605            }
2606        };
2607        let page = if whole {
2608            self.pages.fetch_add(1, Atomic::Relaxed);
2609            let span = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2610            let mut bytes = vec![0; span.length as usize];
2611            read_at(&self.file, span.offset, &mut bytes)?;
2612            Some(Arc::new(bytes))
2613        } else {
2614            None
2615        };
2616        Ok(CachedColumn { stripe: at, index, page })
2617    }
2618
2619    fn read_impl(&self, at: usize, columns: &[usize], whole: bool) -> Result<Chunk> {
2620        let place = *self.places.get(at).ok_or_else(|| invalid("part index out of range"))?;
2621        let index = place.stripe as usize;
2622        let stripe =
2623            self.table.stripes.get(index).ok_or_else(|| invalid("stripe index out of range"))?;
2624        let rows = place.rows as usize;
2625        let mut picked = Vec::with_capacity(columns.len());
2626        for &column in columns {
2627            let field = self
2628                .table
2629                .fields
2630                .get(column)
2631                .ok_or_else(|| invalid("column index out of range"))?;
2632            let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2633            let held = self.held(index, stripe, column, whole)?;
2634            let span = *held
2635                .index
2636                .get(place.part as usize)
2637                .ok_or_else(|| invalid("part index out of range"))?;
2638            let owned;
2639            let bytes = match &held.page {
2640                Some(held) => part_bytes(held, span)?,
2641                None => {
2642                    let offset = page
2643                        .offset
2644                        .checked_add(span.start as u64)
2645                        .ok_or_else(|| invalid("part range overflow"))?;
2646                    let mut bytes = vec![0; span.length];
2647                    read_at(&self.file, offset, &mut bytes)?;
2648                    owned = bytes;
2649                    &owned
2650                }
2651            };
2652            if checksum(bytes) != span.hash {
2653                return Err(invalid(&format!(
2654                    "column page checksum differs, column {column} part {} at {}+{} of {} bytes, \
2655                     wanted {:016x} and got {:016x}",
2656                    place.part,
2657                    page.offset,
2658                    span.start,
2659                    span.length,
2660                    span.hash,
2661                    checksum(bytes),
2662                )));
2663            }
2664            let dictionary = self.dictionary(column)?;
2665            picked.push(decode(&field.ty, rows, bytes, dictionary)?);
2666        }
2667        Chunk::with_rows(picked, rows)
2668    }
2669
2670    /// Whether persisted statistics prove that a part cannot match the predicates.
2671    ///
2672    /// Two of them. The bounds are per stripe, so every part of a stripe gets the same answer from
2673    /// those and a scan that skips one part that way skips all sixty four. The sieves are per part
2674    /// and answer equality, which is the test bounds are worst at: a column of identifiers has every
2675    /// stripe covering nearly the whole of its type, so the bounds keep every part of it and the
2676    /// sieve keeps the ones that really hold the value.
2677    ///
2678    /// The bounds go first because they are already in memory and the sieves are a read.
2679    #[must_use]
2680    pub fn skips(&self, part: usize, probes: &[Probe]) -> bool {
2681        let Some(place) = self.places.get(part).copied() else { return false };
2682        let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
2683        if stripe.zone.skips(probes) {
2684            return true;
2685        }
2686        probes.iter().any(|probe| self.sifted(place, probe))
2687    }
2688
2689    /// Whether the bounds of one stripe prove that none of its parts can match the predicates.
2690    ///
2691    /// The cheap half of [`Self::skips`], asked about a whole stripe at once. The bounds live in the
2692    /// directory and are already in memory, so this answers without touching the file, and that is
2693    /// the reason it is worth having on its own: a caller that wants to know roughly where the work
2694    /// is before it starts any workers can ask this about sixteen stripes for nothing, where asking
2695    /// [`Self::skips`] about nine hundred parts would read and decode a sieve page per stripe first.
2696    ///
2697    /// It keeps stripes that [`Self::skips`] would rule out part by part, which is the right way for
2698    /// it to be wrong: the parts are still checked when they are read.
2699    #[must_use]
2700    pub fn stripe_skips(&self, stripe: usize, probes: &[Probe]) -> bool {
2701        self.table.stripes.get(stripe).is_some_and(|held| held.zone.skips(probes))
2702    }
2703
2704    /// Whether the sieve of one part rules out one probe.
2705    ///
2706    /// Only equality. An ordered comparison is what the bounds are for and a sieve says nothing
2707    /// about it, and a read that cannot answer keeps the part, which is the answer a caller with no
2708    /// sieve gets anyway.
2709    fn sifted(&self, place: Place, probe: &Probe) -> bool {
2710        if probe.op != Op::Equal {
2711            return false;
2712        }
2713        match self.stripe_sieves(place.stripe as usize, probe.column) {
2714            Some(sieves) => sieves
2715                .get(place.part as usize)
2716                .and_then(Option::as_ref)
2717                .is_some_and(|sieve| sieve.excludes(&probe.value)),
2718            None => false,
2719        }
2720    }
2721
2722    /// The sieves of one stripe of one column, read once and kept.
2723    ///
2724    /// `None` when the column has no sieves in that stripe, when the page is damaged, and when the
2725    /// bytes are not a page this version can read. A sieve is an index over data that is still there
2726    /// and a caller that cannot read one reads the rows, so this is the one place in the file where
2727    /// a bad checksum is a slow query rather than an error.
2728    fn stripe_sieves(&self, stripe: usize, column: usize) -> Option<&[Option<Sieve>]> {
2729        let slot = self.sieves.get(column)?.get(stripe)?;
2730        if let Some(held) = slot.get() {
2731            return Some(held);
2732        }
2733        let page = self.table.stripes.get(stripe)?.sieves.get(column).copied().flatten()?;
2734        let mut bytes = vec![0; page.length as usize];
2735        read_at(&self.file, page.offset, &mut bytes).ok()?;
2736        if checksum(&bytes) != page.hash {
2737            return None;
2738        }
2739        let sieves = Arc::new(decode_sieves(&bytes).ok()?);
2740        let _ = slot.set(sieves);
2741        slot.get().map(|held| held.as_slice())
2742    }
2743}
2744
2745/// The value sitting at one position of a dictionary's sorted order.
2746fn text_at_rank(dictionary: &Vector, rank: usize) -> Result<Value> {
2747    let code = dictionary.code_at_rank(rank)? as usize;
2748    let text = dictionary
2749        .try_text_at(code)?
2750        .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
2751    Ok(Value::Varchar(text.into()))
2752}
2753
2754/// Writes one span of a file at an offset, without depending on where the cursor is.
2755///
2756/// The writer owns an offset of its own and passes it in here, so that nothing it writes depends on
2757/// a cursor that a read is entitled to move. Both of these can come back short and both loop.
2758#[cfg(unix)]
2759fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2760    use std::os::unix::fs::FileExt;
2761    while !bytes.is_empty() {
2762        let written = file.write_at(bytes, offset).map_err(io)?;
2763        if written == 0 {
2764            return Err(invalid("a write to the native file wrote nothing"));
2765        }
2766        offset += written as u64;
2767        bytes = &bytes[written..];
2768    }
2769    Ok(())
2770}
2771
2772/// The same write, on the call Windows spells differently.
2773#[cfg(windows)]
2774fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2775    use std::os::windows::fs::FileExt;
2776    while !bytes.is_empty() {
2777        let written = file.seek_write(bytes, offset).map_err(io)?;
2778        if written == 0 {
2779            return Err(invalid("a write to the native file wrote nothing"));
2780        }
2781        offset += written as u64;
2782        bytes = &bytes[written..];
2783    }
2784    Ok(())
2785}
2786
2787/// Somewhere that is neither, where the cursor is all there is.
2788#[cfg(not(any(unix, windows)))]
2789fn write_at(file: &File, offset: u64, bytes: &[u8]) -> Result<()> {
2790    use std::io::Write;
2791    let mut file = file.try_clone().map_err(io)?;
2792    file.seek(SeekFrom::Start(offset)).map_err(io)?;
2793    file.write_all(bytes).map_err(io)
2794}
2795
2796/// Reads one span of a file at an offset, without moving a cursor anybody else can see.
2797///
2798/// Every reader of a table shares one [`File`] behind an [`Arc`], and a grouped aggregate reads its
2799/// pages from several threads at once, so this has to be positional. Seeking and then reading is
2800/// two calls with a gap in the middle, and in that gap another thread's seek lands and the read
2801/// comes back with somebody else's bytes.
2802///
2803/// Both of these can come back short, so both loop. A read of zero bytes before the span is filled
2804/// means the file stops earlier than the directory said it does.
2805#[cfg(unix)]
2806fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2807    use std::os::unix::fs::FileExt;
2808    while !bytes.is_empty() {
2809        let read = file.read_at(bytes, offset).map_err(io)?;
2810        if read == 0 {
2811            return Err(invalid("column page ends before its declared length"));
2812        }
2813        offset += read as u64;
2814        bytes = &mut bytes[read..];
2815    }
2816    Ok(())
2817}
2818
2819/// The same read, on the call Windows spells differently.
2820///
2821/// `seek_read` is one `ReadFile` carrying the offset with it, so two of them cannot interleave the
2822/// way a seek and a read can. It does leave the shared cursor somewhere afterwards, which is why
2823/// nothing in this file may read that cursor.
2824#[cfg(windows)]
2825fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2826    use std::os::windows::fs::FileExt;
2827    while !bytes.is_empty() {
2828        let read = file.seek_read(bytes, offset).map_err(io)?;
2829        if read == 0 {
2830            return Err(invalid("column page ends before its declared length"));
2831        }
2832        offset += read as u64;
2833        bytes = &mut bytes[read..];
2834    }
2835    Ok(())
2836}
2837
2838/// Somewhere that is neither, where the cursor is all there is.
2839///
2840/// This one does race, and there is no way to write it so it does not. Nothing we build for runs
2841/// here, so it exists to keep the crate compiling rather than to be correct under threads.
2842#[cfg(not(any(unix, windows)))]
2843fn read_at(file: &File, offset: u64, bytes: &mut [u8]) -> Result<()> {
2844    let mut file = file.try_clone().map_err(io)?;
2845    file.seek(SeekFrom::Start(offset)).map_err(io)?;
2846    file.read_exact(bytes).map_err(io)
2847}
2848
2849fn type_tag(ty: &LogicalType) -> Result<u8> {
2850    match ty {
2851        LogicalType::SmallInt => Ok(1),
2852        LogicalType::Integer => Ok(2),
2853        LogicalType::BigInt => Ok(3),
2854        LogicalType::Varchar => Ok(4),
2855        LogicalType::Date => Ok(5),
2856        LogicalType::Timestamp => Ok(6),
2857        LogicalType::Boolean => Ok(7),
2858        LogicalType::TinyInt => Ok(8),
2859        LogicalType::UTinyInt => Ok(9),
2860        LogicalType::USmallInt => Ok(10),
2861        LogicalType::UInteger => Ok(11),
2862        LogicalType::UBigInt => Ok(12),
2863        _ => Err(Error::not_implemented(format!("native storage for {ty}"))),
2864    }
2865}
2866
2867fn tag_type(tag: u8) -> Result<LogicalType> {
2868    match tag {
2869        1 => Ok(LogicalType::SmallInt),
2870        2 => Ok(LogicalType::Integer),
2871        3 => Ok(LogicalType::BigInt),
2872        4 => Ok(LogicalType::Varchar),
2873        5 => Ok(LogicalType::Date),
2874        6 => Ok(LogicalType::Timestamp),
2875        7 => Ok(LogicalType::Boolean),
2876        8 => Ok(LogicalType::TinyInt),
2877        9 => Ok(LogicalType::UTinyInt),
2878        10 => Ok(LogicalType::USmallInt),
2879        11 => Ok(LogicalType::UInteger),
2880        12 => Ok(LogicalType::UBigInt),
2881        _ => Err(invalid("column type tag is unknown")),
2882    }
2883}
2884
2885fn put_u16(out: &mut Vec<u8>, value: u16) {
2886    out.extend_from_slice(&value.to_le_bytes());
2887}
2888fn put_u32(out: &mut Vec<u8>, value: u32) {
2889    out.extend_from_slice(&value.to_le_bytes());
2890}
2891fn put_u64(out: &mut Vec<u8>, value: u64) {
2892    out.extend_from_slice(&value.to_le_bytes());
2893}
2894fn put_var_u64(out: &mut Vec<u8>, mut value: u64) {
2895    while value >= 0x80 {
2896        out.push((value as u8 & 0x7f) | 0x80);
2897        value >>= 7;
2898    }
2899    out.push(value as u8);
2900}
2901
2902fn frequency_order(left: FrequencyValue, right: FrequencyValue) -> Ordering {
2903    match (left, right) {
2904        (FrequencyValue::Null, FrequencyValue::Null) => Ordering::Equal,
2905        (FrequencyValue::Null, _) => Ordering::Less,
2906        (_, FrequencyValue::Null) => Ordering::Greater,
2907        (FrequencyValue::Integer(left), FrequencyValue::Integer(right)) => left.cmp(&right),
2908        (FrequencyValue::Code(left), FrequencyValue::Code(right)) => left.cmp(&right),
2909        (FrequencyValue::Integer(_), FrequencyValue::Code(_)) => Ordering::Less,
2910        (FrequencyValue::Code(_), FrequencyValue::Integer(_)) => Ordering::Greater,
2911    }
2912}
2913
2914fn code_frequency(dictionary: &GlobalDictionary) -> FrequencySummary {
2915    let mut entries = dictionary
2916        .counts
2917        .iter()
2918        .enumerate()
2919        .filter(|(_, count)| **count != 0)
2920        .map(|(code, &count)| FrequencyEntry { value: FrequencyValue::Code(code as u32), count })
2921        .collect::<Vec<_>>();
2922    if dictionary.nulls != 0 {
2923        entries.push(FrequencyEntry { value: FrequencyValue::Null, count: dictionary.nulls });
2924    }
2925    entries.sort_unstable_by(|left, right| {
2926        right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
2927    });
2928    let omitted_max = entries.get(FREQUENCY_ENTRIES).map_or(0, |entry| entry.count);
2929    entries.truncate(FREQUENCY_ENTRIES);
2930    FrequencySummary { entries, omitted_max, ordinals: Vec::new() }
2931}
2932
2933fn encode_directory(table: &Table) -> Result<Vec<u8>> {
2934    let mut out = DIRECTORY.to_vec();
2935    let name = table.name.as_bytes();
2936    put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
2937    out.extend_from_slice(name);
2938    put_u16(&mut out, u16::try_from(table.fields.len()).map_err(|_| invalid("too many columns"))?);
2939    for field in &table.fields {
2940        let name = field.name.as_bytes();
2941        put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?);
2942        out.extend_from_slice(name);
2943        out.push(type_tag(&field.ty)?);
2944        out.push(u8::from(field.not_null));
2945    }
2946    for dictionary in &table.dictionaries {
2947        match dictionary {
2948            None => out.push(0),
2949            Some(page) => {
2950                out.push(1);
2951                put_u64(&mut out, page.offset);
2952                put_u32(&mut out, page.length);
2953                put_u64(&mut out, page.hash);
2954            }
2955        }
2956    }
2957    for distinct in &table.distincts {
2958        match distinct {
2959            None => out.push(0),
2960            Some(count) => {
2961                out.push(1);
2962                put_u64(&mut out, *count);
2963            }
2964        }
2965    }
2966    put_u64(&mut out, u64::try_from(table.rows).map_err(|_| invalid("row count overflow"))?);
2967    put_u32(&mut out, u32::try_from(table.stripes.len()).map_err(|_| invalid("too many stripes"))?);
2968    for stripe in &table.stripes {
2969        put_u32(
2970            &mut out,
2971            u32::try_from(stripe.parts.len()).map_err(|_| invalid("too many parts in a stripe"))?,
2972        );
2973        for &rows in &stripe.parts {
2974            put_u32(&mut out, rows);
2975        }
2976        put_u64(&mut out, stripe.index.offset);
2977        put_u32(&mut out, stripe.index.length);
2978        for page in &stripe.pages {
2979            put_u64(&mut out, page.offset);
2980            put_u32(&mut out, page.length);
2981        }
2982        for (field, membership) in table.fields.iter().zip(&stripe.memberships) {
2983            if field.ty != LogicalType::Varchar {
2984                continue;
2985            }
2986            let page =
2987                membership.ok_or_else(|| invalid("string page has no code membership index"))?;
2988            put_u64(&mut out, page.offset);
2989            put_u32(&mut out, page.length);
2990            put_u64(&mut out, page.hash);
2991        }
2992        for sieve in &stripe.sieves {
2993            match sieve {
2994                None => out.push(0),
2995                Some(page) => {
2996                    out.push(1);
2997                    put_u64(&mut out, page.offset);
2998                    put_u32(&mut out, page.length);
2999                    put_u64(&mut out, page.hash);
3000                }
3001            }
3002        }
3003        for range in stripe.zone.columns() {
3004            put_bound(&mut out, range.low.as_ref())?;
3005            put_bound(&mut out, range.high.as_ref())?;
3006            put_u32(
3007                &mut out,
3008                u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?,
3009            );
3010            out.push(u8::from(range.exact));
3011            match range.sum {
3012                None => out.push(0),
3013                Some(total) => {
3014                    out.push(1);
3015                    out.extend_from_slice(&total.to_le_bytes());
3016                }
3017            }
3018        }
3019    }
3020    out.extend_from_slice(FREQUENCIES);
3021    put_u16(
3022        &mut out,
3023        u16::try_from(table.frequencies.len())
3024            .map_err(|_| invalid("too many frequency columns"))?,
3025    );
3026    for summary in &table.frequencies {
3027        let Some(summary) = summary else {
3028            out.push(0);
3029            continue;
3030        };
3031        out.push(1);
3032        put_u64(&mut out, summary.omitted_max);
3033        put_u32(
3034            &mut out,
3035            u32::try_from(summary.entries.len())
3036                .map_err(|_| invalid("too many frequency entries"))?,
3037        );
3038        for entry in &summary.entries {
3039            match entry.value {
3040                FrequencyValue::Null => out.push(0),
3041                FrequencyValue::Integer(value) => {
3042                    out.push(1);
3043                    out.extend_from_slice(&value.to_le_bytes());
3044                }
3045                FrequencyValue::Code(value) => {
3046                    out.push(2);
3047                    put_u32(&mut out, value);
3048                }
3049            }
3050            put_u64(&mut out, entry.count);
3051        }
3052        put_u32(
3053            &mut out,
3054            u32::try_from(summary.ordinals.len())
3055                .map_err(|_| invalid("too many frequency ordinals"))?,
3056        );
3057        let mut previous = 0_u64;
3058        for (at, &ordinal) in summary.ordinals.iter().enumerate() {
3059            let delta = if at == 0 {
3060                ordinal
3061            } else {
3062                ordinal
3063                    .checked_sub(previous)
3064                    .ok_or_else(|| invalid("frequency ordinals are not ordered"))?
3065            };
3066            if at != 0 && delta == 0 {
3067                return Err(invalid("frequency ordinals are not unique"));
3068            }
3069            put_var_u64(&mut out, delta);
3070            previous = ordinal;
3071        }
3072    }
3073    Ok(out)
3074}
3075
3076struct Cursor<'a> {
3077    bytes: &'a [u8],
3078    at: usize,
3079}
3080impl<'a> Cursor<'a> {
3081    fn take(&mut self, len: usize) -> Result<&'a [u8]> {
3082        let end = self.at.checked_add(len).ok_or_else(|| invalid("directory offset overflow"))?;
3083        let bytes =
3084            self.bytes.get(self.at..end).ok_or_else(|| invalid("directory is truncated"))?;
3085        self.at = end;
3086        Ok(bytes)
3087    }
3088    fn u8(&mut self) -> Result<u8> {
3089        Ok(self.take(1)?[0])
3090    }
3091    fn u16(&mut self) -> Result<u16> {
3092        Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("two bytes")))
3093    }
3094    fn u32(&mut self) -> Result<u32> {
3095        Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("four bytes")))
3096    }
3097    fn u64(&mut self) -> Result<u64> {
3098        Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("eight bytes")))
3099    }
3100    fn var_u64(&mut self) -> Result<u64> {
3101        let mut value = 0_u64;
3102        for shift in (0..=63).step_by(7) {
3103            let byte = self.u8()?;
3104            let part = u64::from(byte & 0x7f);
3105            if shift == 63 && part > 1 {
3106                return Err(invalid("frequency ordinal varint overflows"));
3107            }
3108            value |= part << shift;
3109            if byte & 0x80 == 0 {
3110                return Ok(value);
3111            }
3112        }
3113        Err(invalid("frequency ordinal varint is too long"))
3114    }
3115    fn bound(&mut self) -> Result<Option<Bound>> {
3116        Ok(match self.u8()? {
3117            0 => None,
3118            1 => Some(Bound::Int(i128::from_le_bytes(
3119                self.take(16)?.try_into().expect("sixteen bytes"),
3120            ))),
3121            2 => Some(Bound::Real(f64::from_le_bytes(
3122                self.take(8)?.try_into().expect("eight bytes"),
3123            ))),
3124            3 => {
3125                let length = self.u32()? as usize;
3126                Some(Bound::Bytes(self.take(length)?.to_vec()))
3127            }
3128            _ => return Err(invalid("bound tag differs")),
3129        })
3130    }
3131    fn text(&mut self) -> Result<String> {
3132        let len = self.u16()? as usize;
3133        String::from_utf8(self.take(len)?.to_vec()).map_err(|_| invalid("name is not UTF-8"))
3134    }
3135}
3136
3137fn decode_directory(bytes: &[u8], size: u64) -> Result<Table> {
3138    let mut cur = Cursor { bytes, at: 0 };
3139    if cur.take(8)? != DIRECTORY {
3140        return Err(invalid("directory magic differs"));
3141    }
3142    let name = cur.text()?;
3143    let width = cur.u16()? as usize;
3144    let mut fields = Vec::with_capacity(width);
3145    for _ in 0..width {
3146        let name = cur.text()?;
3147        let ty = tag_type(cur.u8()?)?;
3148        let not_null = match cur.u8()? {
3149            0 => false,
3150            1 => true,
3151            _ => return Err(invalid("nullability flag differs")),
3152        };
3153        fields.push(Field { name, ty, not_null });
3154    }
3155    let mut dictionaries = Vec::with_capacity(width);
3156    for _ in 0..width {
3157        dictionaries.push(match cur.u8()? {
3158            0 => None,
3159            1 => {
3160                let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3161                let end = page
3162                    .offset
3163                    .checked_add(u64::from(page.length))
3164                    .ok_or_else(|| invalid("dictionary page offset overflow"))?;
3165                // A global dictionary covers a whole column, not one bounded stripe. Its lazy
3166                // payload is intentionally allowed to grow past `MAX_PAGE`; only ordinary column
3167                // pages are capped there. `Writer::finish` has already bounded this length by the
3168                // on-disk `u32`, and the range check below keeps it inside the file.
3169                if page.offset < HEADER || end > size {
3170                    return Err(invalid("dictionary page range is outside the file"));
3171                }
3172                Some(page)
3173            }
3174            _ => return Err(invalid("dictionary page tag differs")),
3175        });
3176    }
3177    let mut distincts = Vec::with_capacity(width);
3178    for _ in 0..width {
3179        distincts.push(match cur.u8()? {
3180            0 => None,
3181            1 => Some(cur.u64()?),
3182            _ => return Err(invalid("distinct count tag differs")),
3183        });
3184    }
3185    let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
3186    let count = cur.u32()? as usize;
3187    let mut stripes = Vec::with_capacity(count);
3188    let mut total = 0_usize;
3189    for _ in 0..count {
3190        let count = cur.u32()? as usize;
3191        if count == 0 || count > STRIPE_PARTS {
3192            return Err(invalid("stripe part count is outside its bound"));
3193        }
3194        let mut parts = Vec::with_capacity(count);
3195        let mut stripe_rows = 0_usize;
3196        for _ in 0..count {
3197            let rows = cur.u32()?;
3198            if rows == 0 {
3199                return Err(invalid("empty part"));
3200            }
3201            parts.push(rows);
3202            stripe_rows = stripe_rows
3203                .checked_add(rows as usize)
3204                .ok_or_else(|| invalid("stripe row count overflow"))?;
3205        }
3206        total =
3207            total.checked_add(stripe_rows).ok_or_else(|| invalid("stripe row count overflow"))?;
3208        let index = Span { offset: cur.u64()?, length: cur.u32()? };
3209        let section = index_section(count)?;
3210        let wanted = section
3211            .checked_mul(width)
3212            .and_then(|bytes| u32::try_from(bytes).ok())
3213            .ok_or_else(|| invalid("index page length overflow"))?;
3214        let end = index
3215            .offset
3216            .checked_add(u64::from(index.length))
3217            .ok_or_else(|| invalid("index page offset overflow"))?;
3218        if index.offset < HEADER || end > size || index.length != wanted {
3219            return Err(invalid("index page range is outside the file"));
3220        }
3221        let mut pages = Vec::with_capacity(width);
3222        for _ in 0..width {
3223            let offset = cur.u64()?;
3224            let length = cur.u32()?;
3225            let end = offset
3226                .checked_add(u64::from(length))
3227                .ok_or_else(|| invalid("page offset overflow"))?;
3228            if offset < HEADER || end > size || length as usize > MAX_PAGE {
3229                return Err(invalid("page range is outside the file"));
3230            }
3231            pages.push(Span { offset, length });
3232        }
3233        let mut memberships = vec![None; width];
3234        for (column, field) in fields.iter().enumerate() {
3235            if field.ty != LogicalType::Varchar {
3236                continue;
3237            }
3238            let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3239            let end = page
3240                .offset
3241                .checked_add(u64::from(page.length))
3242                .ok_or_else(|| invalid("membership page offset overflow"))?;
3243            if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3244                return Err(invalid("membership page range is outside the file"));
3245            }
3246            memberships[column] = Some(page);
3247        }
3248        let mut sieves = vec![None; width];
3249        for sieve in sieves.iter_mut().take(width) {
3250            match cur.u8()? {
3251                0 => continue,
3252                1 => {}
3253                _ => return Err(invalid("a sieve page has an unknown tag")),
3254            }
3255            let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3256            let end = page
3257                .offset
3258                .checked_add(u64::from(page.length))
3259                .ok_or_else(|| invalid("sieve page offset overflow"))?;
3260            if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3261                return Err(invalid("sieve page range is outside the file"));
3262            }
3263            *sieve = Some(page);
3264        }
3265        let mut ranges = Vec::with_capacity(width);
3266        for _ in 0..width {
3267            let low = cur.bound()?;
3268            let high = cur.bound()?;
3269            let nulls = cur.u32()? as usize;
3270            if nulls > stripe_rows {
3271                return Err(invalid("null count exceeds stripe rows"));
3272            }
3273            let exact = cur.u8()? != 0;
3274            let sum = match cur.u8()? {
3275                0 => None,
3276                1 => Some(i128::from_le_bytes(
3277                    cur.take(16)?.try_into().map_err(|_| invalid("a stripe sum is truncated"))?,
3278                )),
3279                _ => return Err(invalid("a stripe sum has an unknown tag")),
3280            };
3281            ranges.push(Range { low, high, nulls, exact, sum });
3282        }
3283        stripes.push(Stripe {
3284            rows: stripe_rows,
3285            parts,
3286            index,
3287            pages,
3288            memberships,
3289            sieves,
3290            zone: Zone::from_ranges(ranges),
3291        });
3292    }
3293    if total != rows {
3294        return Err(invalid("table row count differs from stripes"));
3295    }
3296    let frequencies = if cur.at == bytes.len() {
3297        vec![None; width]
3298    } else {
3299        if cur.take(8)? != FREQUENCIES {
3300            return Err(invalid("directory extension magic differs"));
3301        }
3302        if cur.u16()? as usize != width {
3303            return Err(invalid("frequency column count differs"));
3304        }
3305        let mut frequencies = Vec::with_capacity(width);
3306        for field in &fields {
3307            let summary = match cur.u8()? {
3308                0 => None,
3309                1 => {
3310                    let omitted_max = cur.u64()?;
3311                    let count = cur.u32()? as usize;
3312                    if count > FREQUENCY_ENTRIES {
3313                        return Err(invalid("frequency entry count exceeds its bound"));
3314                    }
3315                    let mut entries = Vec::with_capacity(count);
3316                    // row at a time: directory decoding validates each persisted bounded frequency entry.
3317                    for _ in 0..count {
3318                        let value = match cur.u8()? {
3319                            0 => FrequencyValue::Null,
3320                            1 => FrequencyValue::Integer(i128::from_le_bytes(
3321                                cur.take(16)?.try_into().expect("sixteen bytes"),
3322                            )),
3323                            2 => FrequencyValue::Code(cur.u32()?),
3324                            _ => return Err(invalid("frequency value tag differs")),
3325                        };
3326                        let valid = matches!(
3327                            (&field.ty, value),
3328                            (_, FrequencyValue::Null)
3329                                | (LogicalType::Varchar, FrequencyValue::Code(_))
3330                                | (
3331                                    LogicalType::TinyInt
3332                                        | LogicalType::SmallInt
3333                                        | LogicalType::Integer
3334                                        | LogicalType::BigInt
3335                                        | LogicalType::UTinyInt
3336                                        | LogicalType::USmallInt
3337                                        | LogicalType::UInteger
3338                                        | LogicalType::UBigInt
3339                                        | LogicalType::Date
3340                                        | LogicalType::Timestamp,
3341                                    FrequencyValue::Integer(_),
3342                                )
3343                        );
3344                        if !valid {
3345                            return Err(invalid("frequency value does not match its column"));
3346                        }
3347                        let count = cur.u64()?;
3348                        if count == 0 || count > rows as u64 {
3349                            return Err(invalid("frequency count is outside the table"));
3350                        }
3351                        entries.push(FrequencyEntry { value, count });
3352                    }
3353                    if entries.windows(2).any(|pair| pair[0].count < pair[1].count) {
3354                        return Err(invalid("frequency entries are not descending"));
3355                    }
3356                    let ordinals = {
3357                        let ordinal_count = cur.u32()? as usize;
3358                        if ordinal_count > FREQUENCY_ORDINALS || ordinal_count > rows {
3359                            return Err(invalid("frequency ordinal count exceeds its bound"));
3360                        }
3361                        let mut ordinals = Vec::with_capacity(ordinal_count);
3362                        let mut previous = 0_u64;
3363                        for at in 0..ordinal_count {
3364                            let delta = cur.var_u64()?;
3365                            if at != 0 && delta == 0 {
3366                                return Err(invalid("frequency ordinals are not increasing"));
3367                            }
3368                            let ordinal = if at == 0 {
3369                                delta
3370                            } else {
3371                                previous
3372                                    .checked_add(delta)
3373                                    .ok_or_else(|| invalid("frequency ordinal overflows"))?
3374                            };
3375                            if ordinal >= rows as u64 {
3376                                return Err(invalid("frequency ordinal is outside the table"));
3377                            }
3378                            ordinals.push(ordinal);
3379                            previous = ordinal;
3380                        }
3381                        ordinals
3382                    };
3383                    Some(FrequencySummary { entries, omitted_max, ordinals })
3384                }
3385                _ => return Err(invalid("frequency summary tag differs")),
3386            };
3387            frequencies.push(summary);
3388        }
3389        frequencies
3390    };
3391    if cur.at != bytes.len() {
3392        return Err(invalid("directory has trailing bytes"));
3393    }
3394    Ok(Table { name, fields, stripes, rows, dictionaries, distincts, frequencies })
3395}
3396
3397fn put_bound(out: &mut Vec<u8>, bound: Option<&Bound>) -> Result<()> {
3398    match bound {
3399        None => out.push(0),
3400        Some(Bound::Int(value)) => {
3401            out.push(1);
3402            out.extend_from_slice(&value.to_le_bytes());
3403        }
3404        Some(Bound::Real(value)) => {
3405            out.push(2);
3406            out.extend_from_slice(&value.to_le_bytes());
3407        }
3408        Some(Bound::Bytes(value)) => {
3409            out.push(3);
3410            put_u32(out, u32::try_from(value.len()).map_err(|_| invalid("bound length overflow"))?);
3411            out.extend_from_slice(value);
3412        }
3413    }
3414    Ok(())
3415}
3416
3417/// Which cascades are worth trying on a run of dictionary codes.
3418///
3419/// The exhaustive chooser encodes every candidate at every level of a cascade three deep and keeps
3420/// the smallest, which on a part of 1024 codes is around a hundred full encodes to decide something
3421/// three candidates were always going to win. It is the right default for a crate that does not
3422/// know what it is looking at. Here we do know. Codes are counted from zero in the order the values
3423/// were first seen, so a part of them is one value, or a narrow band, or a few long runs, and those
3424/// are constant, frame of reference and run length. Nothing else has ever come first on this data.
3425///
3426/// A dictionary of dictionary codes is the one candidate that can never pay, because the codes are
3427/// already the dictionary, and it is also the most expensive one to try. Below the top level the
3428/// streams are an RLE's run values and run lengths, which are integers in their own right with no
3429/// runs left in them, so only the two flat candidates go down there.
3430///
3431/// This is size given up for time on purpose, and the ablation is this chooser against
3432/// [`chooser::EXHAUSTIVE`] on the same file.
3433#[derive(Debug)]
3434struct Codes;
3435
3436impl chooser::Chooser for Codes {
3437    fn name(&self) -> &'static str {
3438        "codes"
3439    }
3440
3441    fn narrow_strings(
3442        &self,
3443        _values: &[&[u8]],
3444        offered: &[string::Kind],
3445        _depth: u8,
3446    ) -> Vec<string::Kind> {
3447        // Never reached, because nothing here encodes strings through the cascade. The trait asks
3448        // for it and the honest answer to a question we have no opinion on is the whole list.
3449        offered.to_vec()
3450    }
3451
3452    fn narrow_integers(
3453        &self,
3454        _values: &[i64],
3455        offered: &[integer::Kind],
3456        depth: u8,
3457    ) -> Vec<integer::Kind> {
3458        let keep: &[integer::Kind] = if depth == 0 {
3459            &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Rle]
3460        } else {
3461            &[integer::Kind::Constant, integer::Kind::Packed]
3462        };
3463        let narrowed: Vec<integer::Kind> =
3464            offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3465        // The contract is a non empty subset, and a chunk that offers none of the three is a chunk
3466        // this has no opinion about rather than one that cannot be written.
3467        if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3468    }
3469}
3470
3471/// Which cascades are worth trying on a part of plain integers.
3472///
3473/// Wider than [`Codes`] because the values are not codes and carry whatever shape the column has.
3474/// A timestamp column climbs, so delta is the one that matters and is the reason this exists at
3475/// all: three timestamp columns in ClickBench were coming out at exactly eight bytes a row with
3476/// nothing asked of them. The same three columns are why the stride is here, since a timestamp
3477/// loaded from a source that recorded whole seconds is microseconds with twenty zero bits under
3478/// every value. A column that is one value with a handful of exceptions is sparse. What is still
3479/// left out is the dictionary, for the same reason as in [`Codes`]: it is the most
3480/// expensive candidate to try and this file already puts the columns that want one through a
3481/// dictionary of their own before they ever reach here.
3482#[derive(Debug)]
3483struct Fixed;
3484
3485impl chooser::Chooser for Fixed {
3486    fn name(&self) -> &'static str {
3487        "fixed"
3488    }
3489
3490    fn narrow_strings(
3491        &self,
3492        _values: &[&[u8]],
3493        offered: &[string::Kind],
3494        _depth: u8,
3495    ) -> Vec<string::Kind> {
3496        offered.to_vec()
3497    }
3498
3499    fn narrow_integers(
3500        &self,
3501        _values: &[i64],
3502        offered: &[integer::Kind],
3503        depth: u8,
3504    ) -> Vec<integer::Kind> {
3505        let keep: &[integer::Kind] = if depth == 0 {
3506            &[
3507                integer::Kind::Constant,
3508                integer::Kind::Packed,
3509                integer::Kind::Delta,
3510                integer::Kind::Rle,
3511                integer::Kind::Sparse,
3512                integer::Kind::Strided,
3513            ]
3514        } else {
3515            &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Delta]
3516        };
3517        let narrowed: Vec<integer::Kind> =
3518            offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3519        if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3520    }
3521}
3522
3523/// Every value of an integer part as an `i64`, or `None` for a part this cannot widen without
3524/// losing one.
3525///
3526/// `UBIGINT` is the only integer type left out, because half its range does not fit and a page that
3527/// silently wrapped would be worse than a page that stays plain. Booleans and strings are not
3528/// integers and have their own ways of being small.
3529fn widened(data: &Data) -> Option<Vec<i64>> {
3530    match data {
3531        Data::Int8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3532        Data::UInt8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3533        Data::Int16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3534        Data::UInt16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3535        Data::Int32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3536        Data::UInt32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3537        Data::Int64(values) => Some(values.to_vec()),
3538        _ => None,
3539    }
3540}
3541
3542/// The same values back in the width the column is declared at.
3543///
3544/// A value that does not fit is a page that disagrees with the directory about what the column is,
3545/// which is a damaged file rather than a caller error, so it is refused rather than truncated.
3546fn narrowed(ty: &LogicalType, values: Vec<i64>) -> Result<Data> {
3547    fn fit<T: TryFrom<i64>>(values: &[i64]) -> Result<Vec<T>> {
3548        values
3549            .iter()
3550            .map(|value| T::try_from(*value).map_err(|_| invalid("page value is not of its type")))
3551            .collect()
3552    }
3553    Ok(match ty {
3554        LogicalType::TinyInt => Data::Int8(fit::<i8>(&values)?.into()),
3555        LogicalType::UTinyInt => Data::UInt8(fit::<u8>(&values)?.into()),
3556        LogicalType::SmallInt => Data::Int16(fit::<i16>(&values)?.into()),
3557        LogicalType::USmallInt => Data::UInt16(fit::<u16>(&values)?.into()),
3558        LogicalType::Integer | LogicalType::Date => Data::Int32(fit::<i32>(&values)?.into()),
3559        LogicalType::UInteger => Data::UInt32(fit::<u32>(&values)?.into()),
3560        LogicalType::BigInt | LogicalType::Timestamp => Data::Int64(values.into()),
3561        _ => return Err(invalid("cascade codec belongs to a page that is not integers")),
3562    })
3563}
3564
3565/// How many bytes a part of this type costs written out plainly, which is what the cascade has to
3566/// beat before it is worth the decode.
3567fn plain_width(ty: &LogicalType) -> Option<usize> {
3568    Some(match ty {
3569        LogicalType::TinyInt | LogicalType::UTinyInt => 1,
3570        LogicalType::SmallInt | LogicalType::USmallInt => 2,
3571        LogicalType::Integer | LogicalType::UInteger | LogicalType::Date => 4,
3572        LogicalType::BigInt | LogicalType::Timestamp => 8,
3573        _ => return None,
3574    })
3575}
3576
3577/// A part's plain integers through the cascade, or `None` when nothing it offers is worth it.
3578///
3579/// What it has to beat is whatever the page would otherwise have cost, which is the bit packed form
3580/// where there is one and the plain width where there is not. Both are cheaper to decode than a
3581/// cascade, so a tie goes to them.
3582fn cascaded(
3583    flat: &Vector,
3584    ty: &LogicalType,
3585    packed: Option<&Packed<'_>>,
3586) -> Result<Option<Vec<u8>>> {
3587    let (Some(width), Some(data)) = (plain_width(ty), flat.data()) else { return Ok(None) };
3588    let Some(values) = widened(data) else { return Ok(None) };
3589    let plain = values.len().saturating_mul(width);
3590    let best = match packed {
3591        // The tag, the base, the word count and the words, which is what the codec 2 branch writes.
3592        Some(packed) => plain.min(21 + size_of_val(packed.words())),
3593        None => plain,
3594    };
3595    let out = integer::encode_with(&values, &Fixed)?;
3596    Ok((out.len() < best).then_some(out))
3597}
3598
3599/// A part's dictionary codes through the integer cascade, or `None` when the cascade did not pay.
3600///
3601/// Until now this stream was a `u32` a row with nothing asked of it, and on ClickBench that was
3602/// 400,185,326 bytes for every one of the 28 varchar columns, the same count for `URL` as for a
3603/// column holding the empty string in nearly every row. Codes are dense integers counted from zero
3604/// and a part holds 1024 of them, which is the shape frame of reference is best at, and a column
3605/// with one value everywhere comes back a constant costing nothing per row rather than four bytes.
3606///
3607/// The result is taken only when it is smaller than the plain form. A cascade is allowed to come
3608/// out larger on a part whose codes are genuinely wide, `URL` has about sixty million distinct
3609/// values, and there is no reason to pay for the decode when it does.
3610fn encoded_codes(codes: &[u32]) -> Result<Option<Vec<u8>>> {
3611    let wide: Vec<i64> = codes.iter().map(|code| i64::from(*code)).collect();
3612    let coded = integer::encode_with(&wide, &Codes)?;
3613    let plain = codes.len().saturating_mul(size_of::<u32>());
3614    Ok((coded.len() < plain).then_some(coded))
3615}
3616
3617fn encode(
3618    vector: &Vector,
3619    global: Option<&mut GlobalDictionary>,
3620) -> Result<(Vec<u8>, Option<Vec<u32>>)> {
3621    let ty = vector.logical_type();
3622    // flatten: the file writer needs a uniform scalar page and does it once per loaded chunk.
3623    let flat = vector.flatten()?;
3624    let mut out = Vec::new();
3625    let mut global_codes = None;
3626    if let Some(global) = global {
3627        let mut codes = Vec::with_capacity(flat.len());
3628        for row in 0..flat.len() {
3629            let text = flat.text_at(row).unwrap_or("");
3630            let code = global.code(text)?;
3631            global.observe(code, flat.is_null_at(row))?;
3632            codes.push(code);
3633        }
3634        global_codes = Some(codes);
3635    }
3636    let membership = global_codes.as_deref().map(unique_codes);
3637    let dictionary = if global_codes.is_none() && ty == &LogicalType::Varchar {
3638        string_dictionary(&flat)?
3639    } else {
3640        None
3641    };
3642    let packed_vector = if dictionary.is_none() && global_codes.is_none() {
3643        Some(flat.bit_packed()?)
3644    } else {
3645        None
3646    };
3647    let packed = packed_vector.as_ref().and_then(Vector::packed_parts);
3648    let coded = match global_codes.as_deref() {
3649        Some(codes) => encoded_codes(codes)?,
3650        None => None,
3651    };
3652    // Only where nothing else has claimed the page, which is the plain integer case. A packed part
3653    // is still on the table because the cascade has to beat it too: the bit pack takes a part only
3654    // when it halves it, so a column that shrinks by a third was coming out whole.
3655    let cascade = if dictionary.is_none() && global_codes.is_none() {
3656        cascaded(&flat, ty, packed.as_ref())?
3657    } else {
3658        None
3659    };
3660    out.push(if coded.is_some() {
3661        4
3662    } else if cascade.is_some() {
3663        5
3664    } else if global_codes.is_some() {
3665        3
3666    } else if dictionary.is_some() {
3667        1
3668    } else if packed.is_some() {
3669        2
3670    } else {
3671        0
3672    });
3673    let nulls = flat.validity();
3674    let flag = match nulls {
3675        Validity::AllValid => 0,
3676        Validity::AllInvalid => 1,
3677        Validity::Mask(_) => 2,
3678    };
3679    out.push(flag);
3680    if flag == 2 {
3681        for group in (0..vector.len()).step_by(8) {
3682            let mut bits = 0_u8;
3683            for bit in 0..8 {
3684                if group + bit < vector.len() && !flat.is_null_at(group + bit) {
3685                    bits |= 1 << bit;
3686                }
3687            }
3688            out.push(bits);
3689        }
3690    }
3691    if let Some(coded) = coded {
3692        out.extend_from_slice(&coded);
3693        return Ok((out, membership));
3694    }
3695    if let Some(cascade) = cascade {
3696        out.extend_from_slice(&cascade);
3697        return Ok((out, membership));
3698    }
3699    if let Some(codes) = global_codes {
3700        for code in codes {
3701            put_u32(&mut out, code);
3702        }
3703        return Ok((out, membership));
3704    }
3705    if let Some(dictionary) = dictionary {
3706        out.extend_from_slice(&dictionary);
3707        return Ok((out, membership));
3708    }
3709    if let Some(packed) = packed {
3710        if packed.offset() != 0 {
3711            return Err(invalid("writer received a sliced packed vector"));
3712        }
3713        out.push(u8::try_from(packed.width()).map_err(|_| invalid("packed width overflow"))?);
3714        out.extend_from_slice(&packed.base().to_le_bytes());
3715        put_u32(
3716            &mut out,
3717            u32::try_from(packed.words().len()).map_err(|_| invalid("too many packed words"))?,
3718        );
3719        for word in packed.words() {
3720            put_u64(&mut out, *word);
3721        }
3722        return Ok((out, membership));
3723    }
3724    let data = flat.data().ok_or_else(|| invalid("scalar column did not flatten"))?;
3725    match (ty, data) {
3726        (LogicalType::TinyInt, Data::Int8(values)) => {
3727            for value in &**values {
3728                out.extend_from_slice(&value.to_le_bytes());
3729            }
3730        }
3731        (LogicalType::UTinyInt, Data::UInt8(values)) => {
3732            for value in &**values {
3733                out.extend_from_slice(&value.to_le_bytes());
3734            }
3735        }
3736        (LogicalType::SmallInt, Data::Int16(values)) => {
3737            for value in &**values {
3738                out.extend_from_slice(&value.to_le_bytes());
3739            }
3740        }
3741        (LogicalType::USmallInt, Data::UInt16(values)) => {
3742            for value in &**values {
3743                out.extend_from_slice(&value.to_le_bytes());
3744            }
3745        }
3746        (LogicalType::UInteger, Data::UInt32(values)) => {
3747            for value in &**values {
3748                out.extend_from_slice(&value.to_le_bytes());
3749            }
3750        }
3751        (LogicalType::UBigInt, Data::UInt64(values)) => {
3752            for value in &**values {
3753                out.extend_from_slice(&value.to_le_bytes());
3754            }
3755        }
3756        (LogicalType::Integer | LogicalType::Date, Data::Int32(values)) => {
3757            for value in &**values {
3758                out.extend_from_slice(&value.to_le_bytes());
3759            }
3760        }
3761        (LogicalType::BigInt | LogicalType::Timestamp, Data::Int64(values)) => {
3762            for value in &**values {
3763                out.extend_from_slice(&value.to_le_bytes());
3764            }
3765        }
3766        (LogicalType::Boolean, Data::Bool(values)) => {
3767            for value in &**values {
3768                out.push(u8::from(*value));
3769            }
3770        }
3771        (LogicalType::Varchar, Data::Varlen(values)) => {
3772            let mut bytes = Vec::new();
3773            put_u32(&mut out, 0);
3774            for row in 0..vector.len() {
3775                let value = values.bytes(row).ok_or_else(|| invalid("string view is invalid"))?;
3776                bytes.extend_from_slice(value);
3777                put_u32(
3778                    &mut out,
3779                    u32::try_from(bytes.len())
3780                        .map_err(|_| invalid("string payload exceeds 4GiB"))?,
3781                );
3782            }
3783            out.extend_from_slice(&bytes);
3784        }
3785        _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
3786    }
3787    Ok((out, membership))
3788}
3789
3790fn put_varint(out: &mut Vec<u8>, mut value: u32) {
3791    while value >= 0x80 {
3792        out.push((value as u8 & 0x7f) | 0x80);
3793        value >>= 7;
3794    }
3795    out.push(value as u8);
3796}
3797
3798/// The distinct codes of one part, which is what a stripe's membership index is merged from.
3799fn unique_codes(codes: &[u32]) -> Vec<u32> {
3800    let mut unique = codes.to_vec();
3801    unique.sort_unstable();
3802    unique.dedup();
3803    unique
3804}
3805
3806/// The union of the sorted distinct codes of every part in a stripe.
3807///
3808/// Pairwise up a tree rather than one long list concatenated and sorted. Both are the same order of
3809/// work on paper and the tree is the one that does not sort what is already in order: sixty four
3810/// sorted lists become one in six passes over the values.
3811fn merged_codes(lists: Vec<Vec<u32>>) -> Vec<u32> {
3812    let mut lists = lists;
3813    while lists.len() > 1 {
3814        let mut next = Vec::with_capacity(lists.len().div_ceil(2));
3815        for pair in lists.chunks(2) {
3816            match pair {
3817                [left, right] => next.push(merged_pair(left, right)),
3818                [only] => next.push(only.clone()),
3819                _ => {}
3820            }
3821        }
3822        lists = next;
3823    }
3824    lists.pop().unwrap_or_default()
3825}
3826
3827fn merged_pair(left: &[u32], right: &[u32]) -> Vec<u32> {
3828    let mut out = Vec::with_capacity(left.len().saturating_add(right.len()));
3829    let mut at = 0;
3830    let mut to = 0;
3831    while at < left.len() && to < right.len() {
3832        match left[at].cmp(&right[to]) {
3833            Ordering::Less => {
3834                out.push(left[at]);
3835                at += 1;
3836            }
3837            Ordering::Greater => {
3838                out.push(right[to]);
3839                to += 1;
3840            }
3841            Ordering::Equal => {
3842                out.push(left[at]);
3843                at += 1;
3844                to += 1;
3845            }
3846        }
3847    }
3848    out.extend_from_slice(&left[at..]);
3849    out.extend_from_slice(&right[to..]);
3850    out
3851}
3852
3853/// The widest bounds and the total null count of a stripe, from the bounds of its parts.
3854///
3855/// A bound that is missing from any part is missing from the stripe, because a missing bound means
3856/// nothing is known and a stripe that holds an unknown cannot claim one.
3857fn merged_range(ranges: impl Iterator<Item = Range>) -> Range {
3858    let mut merged = Range::default();
3859    let mut first = true;
3860    for range in ranges {
3861        merged.nulls = merged.nulls.saturating_add(range.nulls);
3862        // Both of these have to survive every part, so one part that could not say anything makes
3863        // the stripe unable to say it either. A sum is dropped on overflow rather than wrapped,
3864        // which leaves the stripe with exact ends and no total, which is a true thing to say.
3865        merged.sum = match (merged.sum.take(), range.sum) {
3866            (Some(held), Some(next)) if !first => held.checked_add(next),
3867            (_, next) if first => next,
3868            _ => None,
3869        };
3870        merged.exact = if first { range.exact } else { merged.exact && range.exact };
3871        if first {
3872            merged.low = range.low;
3873            merged.high = range.high;
3874            first = false;
3875            continue;
3876        }
3877        merged.low = match (merged.low.take(), range.low) {
3878            (Some(held), Some(next)) => Some(held.smaller(next)),
3879            _ => None,
3880        };
3881        merged.high = match (merged.high.take(), range.high) {
3882            (Some(held), Some(next)) => Some(held.larger(next)),
3883            _ => None,
3884        };
3885    }
3886    merged
3887}
3888
3889/// One stripe's sieves for one column: the part count, a length for each part, then their bytes.
3890///
3891/// One page for the whole stripe rather than one per part, because a part's sieve is a few hundred
3892/// bytes and sixty four of those are sixty four directory entries and sixty four reads for something
3893/// a scan walks straight through. A part with no sieve writes a length of zero and costs four bytes.
3894fn encode_sieves<'a>(sieves: impl Iterator<Item = &'a Option<Sieve>>) -> Result<Vec<u8>> {
3895    let held: Vec<&Option<Sieve>> = sieves.collect();
3896    let mut out = Vec::new();
3897    put_u32(
3898        &mut out,
3899        u32::try_from(held.len()).map_err(|_| invalid("too many parts in a stripe"))?,
3900    );
3901    for sieve in &held {
3902        let length = sieve.as_ref().map_or(0, Sieve::len);
3903        put_u32(&mut out, u32::try_from(length).map_err(|_| invalid("sieve length overflow"))?);
3904    }
3905    // flatten: a part with no sieve wrote a length of zero above and contributes no bytes here.
3906    for sieve in held.into_iter().flatten() {
3907        out.extend_from_slice(&sieve.to_bytes());
3908    }
3909    Ok(out)
3910}
3911
3912/// The sieves one encoded page holds, one entry per part of the stripe.
3913///
3914/// A part whose bytes are not a sieve this version understands comes back as `None`, which is a part
3915/// that gets read. That is how a file written by a later version of the sieve stays readable rather
3916/// than being a corrupt page.
3917fn decode_sieves(bytes: &[u8]) -> Result<Vec<Option<Sieve>>> {
3918    let parts = u32::from_le_bytes(
3919        bytes
3920            .get(..4)
3921            .ok_or_else(|| invalid("sieve page is truncated"))?
3922            .try_into()
3923            .map_err(|_| invalid("sieve page is truncated"))?,
3924    ) as usize;
3925    let mut lengths = Vec::with_capacity(parts);
3926    for part in 0..parts {
3927        let at = 4 + part * 4;
3928        let field = bytes.get(at..at + 4).ok_or_else(|| invalid("sieve page is truncated"))?;
3929        lengths.push(u32::from_le_bytes(
3930            field.try_into().map_err(|_| invalid("sieve page is truncated"))?,
3931        ) as usize);
3932    }
3933    let mut at = 4 + parts * 4;
3934    let mut out = Vec::with_capacity(parts);
3935    for length in lengths {
3936        if length == 0 {
3937            out.push(None);
3938            continue;
3939        }
3940        let end = at.checked_add(length).ok_or_else(|| invalid("sieve page is truncated"))?;
3941        let field = bytes.get(at..end).ok_or_else(|| invalid("sieve page is truncated"))?;
3942        out.push(Sieve::from_bytes(field));
3943        at = end;
3944    }
3945    if at != bytes.len() {
3946        return Err(invalid("sieve page has trailing bytes"));
3947    }
3948    Ok(out)
3949}
3950
3951/// One stripe's membership index: the code count and then the codes as ascending deltas.
3952///
3953/// The codes have to be sorted and distinct already, which is what [`unique_codes`] and
3954/// [`merged_codes`] hand over. Anything else decodes as different codes, so neither of those two is
3955/// a step a caller can skip.
3956fn encode_membership(unique: &[u32]) -> Vec<u8> {
3957    let mut out = Vec::with_capacity(unique.len().saturating_mul(2).saturating_add(5));
3958    put_varint(&mut out, u32::try_from(unique.len()).unwrap_or(u32::MAX));
3959    let mut previous = 0;
3960    for (at, &code) in unique.iter().enumerate() {
3961        put_varint(&mut out, if at == 0 { code } else { code - previous });
3962        previous = code;
3963    }
3964    out
3965}
3966
3967fn take_varint(bytes: &[u8], at: &mut usize) -> Result<u32> {
3968    let mut value = 0_u32;
3969    for shift in (0..35).step_by(7) {
3970        let byte = *bytes.get(*at).ok_or_else(|| invalid("membership varint is truncated"))?;
3971        *at += 1;
3972        let part = u32::from(byte & 0x7f);
3973        if shift == 28 && part > 0x0f {
3974            return Err(invalid("membership varint overflow"));
3975        }
3976        value = value
3977            .checked_add(
3978                part.checked_shl(shift).ok_or_else(|| invalid("membership varint overflow"))?,
3979            )
3980            .ok_or_else(|| invalid("membership varint overflow"))?;
3981        if byte & 0x80 == 0 {
3982            return Ok(value);
3983        }
3984    }
3985    Err(invalid("membership varint is too long"))
3986}
3987
3988fn decode_membership(bytes: &[u8]) -> Result<Vec<u32>> {
3989    let mut at = 0;
3990    let count = take_varint(bytes, &mut at)? as usize;
3991    let mut codes = Vec::with_capacity(count);
3992    let mut previous = 0_u32;
3993    for index in 0..count {
3994        let delta = take_varint(bytes, &mut at)?;
3995        let code = if index == 0 {
3996            delta
3997        } else {
3998            previous.checked_add(delta).ok_or_else(|| invalid("membership code overflow"))?
3999        };
4000        if index > 0 && code <= previous {
4001            return Err(invalid("membership codes are not increasing"));
4002        }
4003        codes.push(code);
4004        previous = code;
4005    }
4006    if at != bytes.len() {
4007        return Err(invalid("membership page has trailing bytes"));
4008    }
4009    Ok(codes)
4010}
4011
4012fn string_dictionary(vector: &Vector) -> Result<Option<Vec<u8>>> {
4013    let mut by_text = HashMap::new();
4014    let mut values = Vec::new();
4015    let mut codes = Vec::with_capacity(vector.len());
4016    let mut plain_bytes = 0_usize;
4017    for row in 0..vector.len() {
4018        let text = vector.text_at(row).unwrap_or("");
4019        plain_bytes = plain_bytes.saturating_add(text.len());
4020        let code = match by_text.get(text) {
4021            Some(&code) => code,
4022            None => {
4023                let code = u32::try_from(values.len())
4024                    .map_err(|_| invalid("too many dictionary values"))?;
4025                by_text.insert(text, code);
4026                values.push(text);
4027                code
4028            }
4029        };
4030        codes.push(code);
4031    }
4032    let dictionary_bytes = values.iter().map(|value| value.len()).sum::<usize>();
4033    let encoded = 8_usize
4034        .saturating_add((values.len() + 1).saturating_mul(4))
4035        .saturating_add(dictionary_bytes)
4036        .saturating_add(codes.len().saturating_mul(4));
4037    let plain = (vector.len() + 1).saturating_mul(4).saturating_add(plain_bytes);
4038    if encoded >= plain {
4039        return Ok(None);
4040    }
4041    let mut out = Vec::with_capacity(encoded);
4042    put_u32(
4043        &mut out,
4044        u32::try_from(values.len()).map_err(|_| invalid("too many dictionary values"))?,
4045    );
4046    put_u32(
4047        &mut out,
4048        u32::try_from(dictionary_bytes).map_err(|_| invalid("dictionary payload exceeds 4GiB"))?,
4049    );
4050    let mut offset = 0_u32;
4051    put_u32(&mut out, offset);
4052    for value in &values {
4053        offset = offset
4054            .checked_add(
4055                u32::try_from(value.len()).map_err(|_| invalid("dictionary value is too long"))?,
4056            )
4057            .ok_or_else(|| invalid("dictionary payload exceeds 4GiB"))?;
4058        put_u32(&mut out, offset);
4059    }
4060    for value in values {
4061        out.extend_from_slice(value.as_bytes());
4062    }
4063    for code in codes {
4064        put_u32(&mut out, code);
4065    }
4066    Ok(Some(out))
4067}
4068
4069struct EncodedDictionary {
4070    index: Vec<u8>,
4071    ranks: Vec<u8>,
4072    /// The payload as the blocks it is written as, kept apart rather than joined because joining
4073    /// them is a second copy of a thing that is already gigabytes on the columns that matter.
4074    payload: Vec<Vec<u8>>,
4075}
4076
4077/// The first eight bytes of a value as an integer that sorts the way the bytes sort.
4078fn head(bytes: &[u8]) -> u64 {
4079    let mut word = [0; 8];
4080    let take = bytes.len().min(8);
4081    word[..take].copy_from_slice(&bytes[..take]);
4082    u64::from_be_bytes(word)
4083}
4084
4085/// The sorted order of every global dictionary, one entry per column and empty where there is no
4086/// dictionary.
4087///
4088/// One column's sort has nothing to do with another's, and a table like `hits` has fifteen string
4089/// columns, so this runs across threads the way the numeric synopses above do. It is the only part
4090/// of committing a file that is more than bookkeeping, and doing it serially would show up as a
4091/// pause at the end of a load that thirty two threads had been busy with until then.
4092fn rankings(dictionaries: &[Option<GlobalDictionary>]) -> Result<Vec<Vec<(u64, u32)>>> {
4093    let present =
4094        dictionaries.iter().enumerate().filter(|(_, held)| held.is_some()).map(|(at, _)| at);
4095    let present = present.collect::<Vec<_>>();
4096    let mut orders = vec![Vec::new(); dictionaries.len()];
4097    let workers = std::thread::available_parallelism()
4098        .map_or(1, usize::from)
4099        .min(MAX_FREQUENCY_WORKERS)
4100        .min(present.len());
4101    if workers <= 1 {
4102        for at in present {
4103            if let Some(dictionary) = &dictionaries[at] {
4104                orders[at] = dictionary.ranked();
4105            }
4106        }
4107        return Ok(orders);
4108    }
4109    let width = present.len().div_ceil(workers);
4110    let pieces = std::thread::scope(|scope| {
4111        present
4112            .chunks(width)
4113            .map(|columns| {
4114                scope.spawn(|| {
4115                    columns
4116                        .iter()
4117                        .filter_map(|&at| dictionaries[at].as_ref().map(|held| (at, held.ranked())))
4118                        .collect::<Vec<_>>()
4119                })
4120            })
4121            .collect::<Vec<_>>()
4122            .into_iter()
4123            .map(|handle| {
4124                handle.join().map_err(|_| Error::internal("a dictionary sort worker panicked"))
4125            })
4126            .collect::<Result<Vec<_>>>()
4127    })?;
4128    for piece in pieces {
4129        for (at, order) in piece {
4130            orders[at] = order;
4131        }
4132    }
4133    Ok(orders)
4134}
4135
4136fn encode_global_dictionary(
4137    dictionary: GlobalDictionary,
4138    order: &[(u64, u32)],
4139) -> Result<EncodedDictionary> {
4140    let values = dictionary.offsets.len() - 1;
4141    if order.len() != values {
4142        return Err(invalid("global dictionary order does not cover its values"));
4143    }
4144    let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
4145    let payload = encode_payload(&dictionary)?;
4146    if payload.len() != blocks {
4147        return Err(invalid("global dictionary payload is not the blocks it says it is"));
4148    }
4149    let (ranks, rank_ends) = encode_ranks(order, code_width(values))?;
4150    let rank_blocks = values.div_ceil(TEXT_RANK_BLOCK);
4151    let offset_bits = offset_width(&dictionary.offsets);
4152    let mut index = Vec::with_capacity(
4153        DICTIONARY_HEADER + offset_bytes(values, offset_bits) + (blocks + rank_blocks) * 16,
4154    );
4155    put_u32(
4156        &mut index,
4157        u32::try_from(values).map_err(|_| invalid("global dictionary has too many values"))?,
4158    );
4159    put_u32(&mut index, TEXT_PAYLOAD_VALUES as u32);
4160    put_u32(
4161        &mut index,
4162        u32::try_from(blocks).map_err(|_| invalid("global dictionary has too many blocks"))?,
4163    );
4164    put_u32(&mut index, offset_bits as u32);
4165    encode_offsets(&dictionary.offsets, offset_bits, &mut index)?;
4166    // Where each block ends, so a reader can find one. The stored blocks are shorter than the
4167    // decoded ones and by a different amount each, so this is the one thing the offsets above no
4168    // longer say.
4169    let mut at = 0_u64;
4170    for block in &payload {
4171        at = at
4172            .checked_add(block.len() as u64)
4173            .ok_or_else(|| invalid("global dictionary payload overflow"))?;
4174        put_u64(&mut index, at);
4175    }
4176    for block in &payload {
4177        put_u64(&mut index, checksum(block));
4178    }
4179    // The same two lists for the sorted order. A rank block is packed at whatever width its own
4180    // heads need, so where one ends is no longer arithmetic on the block number.
4181    if rank_ends.len() != rank_blocks {
4182        return Err(invalid("global dictionary order is not the blocks it says it is"));
4183    }
4184    for end in &rank_ends {
4185        put_u64(&mut index, *end);
4186    }
4187    let mut at = 0_usize;
4188    for end in &rank_ends {
4189        let end = usize::try_from(*end).map_err(|_| invalid("global dictionary order overflow"))?;
4190        put_u64(&mut index, checksum(&ranks[at..end]));
4191        at = end;
4192    }
4193    Ok(EncodedDictionary { index, ranks, payload })
4194}
4195
4196/// How many blocks of the payload the shape is settled on.
4197///
4198/// Eight blocks is 8,192 values, which is the sample `chooser::Sampled` draws and is that size for
4199/// the same reason. They are spread across the dictionary rather than taken off the front, because
4200/// a dictionary is in the order values were first seen and the front of it is the first morsel of
4201/// the load.
4202const PAYLOAD_SAMPLE_BLOCKS: usize = 8;
4203
4204/// The shapes the payload encoder picks between.
4205///
4206/// Narrow on purpose. The exhaustive search encodes every candidate at every level and runs at two
4207/// to six megabytes a second on this data, which over the twelve gigabytes of dictionary `hits`
4208/// carries is about an hour of processor time, so it cannot be what a load does. Each of these
4209/// settles the outer level and the one below it, which is where almost all of that hour goes, and
4210/// leaves the levels under them to the exhaustive search where the chunks are small enough for it
4211/// to cost nothing.
4212///
4213/// Measured on the five ClickBench columns that have a dictionary worth the name, at 1,024 values a
4214/// block, against the exhaustive search over the same blocks:
4215///
4216/// | column | exhaustive | FRONT then LZ | LZ then FSST | LZ then PLAIN |
4217/// |---|---|---|---|---|
4218/// | 2 | 2.923 at 4.3 MB/s | 2.587 at 21.2 | 2.593 at 36.1 | 2.538 at 53.6 |
4219/// | 13 | 3.093 at 3.1 | 3.029 at 36.4 | 2.921 at 35.7 | 2.770 at 82.9 |
4220/// | 14 | 2.330 at 2.1 | 2.283 at 24.3 | 2.213 at 23.5 | 2.113 at 67.6 |
4221/// | 39 | 2.459 at 5.3 | 2.147 at 10.6 | 2.145 at 29.3 | 2.088 at 43.1 |
4222/// | 56 | 4.694 at 6.3 | 4.381 at 51.0 | 4.172 at 50.6 | 3.983 at 86.8 |
4223///
4224/// The best of the three per column is 98 percent of the exhaustive ratio for a tenth of the time.
4225/// `FSST` and `PLAIN` on their own are in the list as a floor rather than to win. `FSST` is the
4226/// right answer for text that does not share prefixes with its neighbours, and `PLAIN` is there so
4227/// that a column nothing compresses is found out in the sample and written at a gigabyte a second
4228/// rather than searched for an answer that does not exist.
4229fn payload_shapes() -> Vec<chooser::Settled> {
4230    let integers = vec![integer::Kind::Packed];
4231    [
4232        vec![string::Kind::Front, string::Kind::Lz],
4233        vec![string::Kind::Lz, string::Kind::Fsst],
4234        vec![string::Kind::Lz, string::Kind::Plain],
4235        vec![string::Kind::Fsst],
4236        vec![string::Kind::Plain],
4237    ]
4238    .into_iter()
4239    .map(|strings| chooser::Settled::new(strings, integers.clone()))
4240    .collect()
4241}
4242
4243/// The payload as encoded blocks of [`TEXT_PAYLOAD_VALUES`] values each.
4244///
4245/// Across threads because this is the only part of committing a file that is real work rather than
4246/// bookkeeping. The blocks are the same size and cost about the same, so an index each is enough of
4247/// a queue and there is nothing to weight the way the numeric synopses are weighted.
4248fn encode_payload(dictionary: &GlobalDictionary) -> Result<Vec<Vec<u8>>> {
4249    let values = dictionary.offsets.len() - 1;
4250    let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
4251    let run = |block: usize| {
4252        let first = block * TEXT_PAYLOAD_VALUES;
4253        let last = (first + TEXT_PAYLOAD_VALUES).min(values);
4254        (first..last)
4255            .map(|value| {
4256                let from = dictionary.offsets[value] as usize;
4257                let to = dictionary.offsets[value + 1] as usize;
4258                &dictionary.payload[from..to]
4259            })
4260            .collect::<Vec<_>>()
4261    };
4262    // A dictionary small enough to be the sample is small enough to search in full, and searching
4263    // it costs less than deciding not to.
4264    let shape = (blocks > PAYLOAD_SAMPLE_BLOCKS).then(|| settle_shape(&run, blocks)).transpose()?;
4265    let one = |block: usize| match &shape {
4266        Some(shape) => string::encode_with(&run(block), shape),
4267        None => string::encode(&run(block)),
4268    };
4269    let workers = std::thread::available_parallelism()
4270        .map_or(1, usize::from)
4271        .min(MAX_FREQUENCY_WORKERS)
4272        .min(blocks);
4273    if workers <= 1 {
4274        return (0..blocks).map(one).collect();
4275    }
4276    let next = AtomicUsize::new(0);
4277    let pieces = std::thread::scope(|scope| {
4278        (0..workers)
4279            .map(|_| {
4280                scope.spawn(|| {
4281                    let mut mine = Vec::new();
4282                    loop {
4283                        let block = next.fetch_add(1, Atomic::Relaxed);
4284                        if block >= blocks {
4285                            break;
4286                        }
4287                        mine.push((block, one(block)?));
4288                    }
4289                    Ok(mine)
4290                })
4291            })
4292            .collect::<Vec<_>>()
4293            .into_iter()
4294            .map(|handle| {
4295                handle.join().map_err(|_| Error::internal("a dictionary encode worker panicked"))?
4296            })
4297            .collect::<Result<Vec<_>>>()
4298    })?;
4299    let mut payload = vec![Vec::new(); blocks];
4300    for piece in pieces {
4301        for (block, bytes) in piece {
4302            payload[block] = bytes;
4303        }
4304    }
4305    Ok(payload)
4306}
4307
4308/// Which of [`payload_shapes`] comes out smallest over a sample of the blocks.
4309///
4310/// Every shape is encoded over the same sample and the smallest wins, which is the exhaustive
4311/// search moved up a level: over shapes of a column rather than over candidates of a chunk. The
4312/// sample is spread across the dictionary so that the first and last blocks are both in it, because
4313/// a dictionary written in first seen order has its common values at the front and its long tail at
4314/// the back, and those do not compress alike.
4315fn settle_shape<'a>(
4316    run: &dyn Fn(usize) -> Vec<&'a [u8]>,
4317    blocks: usize,
4318) -> Result<chooser::Settled> {
4319    let last = blocks - 1;
4320    let sample = (0..PAYLOAD_SAMPLE_BLOCKS)
4321        .map(|region| run(region * last / (PAYLOAD_SAMPLE_BLOCKS - 1)))
4322        .collect::<Vec<_>>();
4323    let mut best: Option<(chooser::Settled, usize)> = None;
4324    for shape in payload_shapes() {
4325        let mut size = 0;
4326        for block in &sample {
4327            size += string::encode_with(block, &shape)?.len();
4328        }
4329        if best.as_ref().is_none_or(|(_, smallest)| size < *smallest) {
4330            best = Some((shape, size));
4331        }
4332    }
4333    best.map(|(shape, _)| shape)
4334        .ok_or_else(|| invalid("no shape applies to a global dictionary payload"))
4335}
4336
4337/// The sorted order laid out the way a reader reads it, in blocks of [`TEXT_RANK_BLOCK`] entries.
4338///
4339/// Each block holds its heads first and then its codes, rather than pairing them, because a search
4340/// asks for a head at every probe and for a code about once a search. Keeping the heads together
4341/// means a probe touches eight bytes of a block rather than twelve spread over it, and the last few
4342/// probes of a search, which are the ones that land in the same block, touch the same cache line.
4343fn encode_ranks(order: &[(u64, u32)], code_bits: usize) -> Result<(Vec<u8>, Vec<u64>)> {
4344    let mut out = Vec::with_capacity(order.len() * 4);
4345    let mut ends = Vec::with_capacity(order.len().div_ceil(TEXT_RANK_BLOCK));
4346    let mut heads = Vec::with_capacity(TEXT_RANK_BLOCK);
4347    let mut codes = Vec::with_capacity(TEXT_RANK_BLOCK);
4348    for block in order.chunks(TEXT_RANK_BLOCK) {
4349        // The order is sorted by value and a head is a prefix of a value, so the heads of a block
4350        // rise, the smallest is the first and the largest is the last.
4351        let base = block.first().map_or(0, |&(head, _)| head);
4352        let span = block.last().map_or(0, |&(head, _)| head.wrapping_sub(base));
4353        let width = (u64::BITS - span.leading_zeros()) as usize;
4354        heads.clear();
4355        codes.clear();
4356        for &(head, code) in block {
4357            heads.push(head.wrapping_sub(base));
4358            codes.push(u64::from(code));
4359        }
4360        put_u64(&mut out, base);
4361        out.push(width as u8);
4362        bitpack::pack_tail(&heads, width, &mut out)
4363            .map_err(|_| invalid("global dictionary heads do not pack"))?;
4364        bitpack::pack_tail(&codes, code_bits, &mut out)
4365            .map_err(|_| invalid("global dictionary codes do not pack"))?;
4366        ends.push(out.len() as u64);
4367    }
4368    Ok((out, ends))
4369}
4370
4371fn open_global_dictionary(file: Arc<File>, page: Page, ty: &LogicalType) -> Result<Vector> {
4372    if ty != &LogicalType::Varchar {
4373        return Err(invalid("global dictionary belongs to a non-string column"));
4374    }
4375    let mut header = [0; DICTIONARY_HEADER];
4376    read_at(&file, page.offset, &mut header)?;
4377    let count = u32::from_le_bytes(header[0..4].try_into().expect("four bytes")) as usize;
4378    let per_block = u32::from_le_bytes(header[4..8].try_into().expect("four bytes")) as usize;
4379    let blocks = u32::from_le_bytes(header[8..12].try_into().expect("four bytes")) as usize;
4380    let offset_bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
4381    if per_block != TEXT_PAYLOAD_VALUES {
4382        return Err(invalid("global dictionary block width differs"));
4383    }
4384    if blocks != count.div_ceil(TEXT_PAYLOAD_VALUES) {
4385        return Err(invalid("global dictionary block count differs from its value count"));
4386    }
4387    if offset_bits > u32::BITS as usize {
4388        return Err(invalid("global dictionary packs offsets past a payload"));
4389    }
4390    let offset_len = offset_bytes(count, offset_bits);
4391    // The sorted order is kept out of the index on purpose. The index is read and checksummed in
4392    // full the moment the column is first touched, and the order is half again the size of the
4393    // offsets, so putting it there would make every query that reads a string column pay for a
4394    // search that most of them never make.
4395    let ranks = count;
4396    let rank_blocks = ranks.div_ceil(TEXT_RANK_BLOCK);
4397    // Two words a payload block, one for where it ends in the file and one for its checksum, and the
4398    // same two a rank block.
4399    let hash_len = blocks
4400        .checked_add(rank_blocks)
4401        .and_then(|words| words.checked_mul(16))
4402        .ok_or_else(|| invalid("global dictionary block count overflow"))?;
4403    let index_len = DICTIONARY_HEADER
4404        .checked_add(offset_len)
4405        .and_then(|len| len.checked_add(hash_len))
4406        .ok_or_else(|| invalid("global dictionary header overflow"))?;
4407    if index_len > page.length as usize {
4408        return Err(invalid("global dictionary offset index exceeds its page"));
4409    }
4410    let mut index = vec![0; index_len];
4411    index[..DICTIONARY_HEADER].copy_from_slice(&header);
4412    read_at(&file, page.offset + DICTIONARY_HEADER as u64, &mut index[DICTIONARY_HEADER..])?;
4413    if checksum(&index) != page.hash {
4414        return Err(invalid("global dictionary index checksum differs"));
4415    }
4416    let offsets = index[DICTIONARY_HEADER..DICTIONARY_HEADER + offset_len].to_vec();
4417    let mut words = index[DICTIONARY_HEADER + offset_len..]
4418        .chunks_exact(8)
4419        .map(|part| u64::from_le_bytes(part.try_into().expect("eight bytes")))
4420        .collect::<Vec<_>>();
4421    let mut hashes = words.split_off(blocks);
4422    let mut rank_ends = hashes.split_off(blocks);
4423    let rank_hashes = rank_ends.split_off(rank_blocks);
4424    let ends = words;
4425    // A rank block packs its heads at whatever width its own values need, so its length is no longer
4426    // arithmetic on the block number and the reader has to be told where each one ends.
4427    if rank_ends.windows(2).any(|pair| pair[0] >= pair[1]) {
4428        return Err(invalid("global dictionary order blocks do not rise"));
4429    }
4430    let rank_len = usize::try_from(rank_ends.last().copied().unwrap_or_default())
4431        .map_err(|_| invalid("global dictionary rank overflow"))?;
4432    let body_len = index_len
4433        .checked_add(rank_len)
4434        .ok_or_else(|| invalid("global dictionary header overflow"))?;
4435    if body_len > page.length as usize {
4436        return Err(invalid("global dictionary order exceeds its page"));
4437    }
4438    // What the offsets bound is the decoded payload, and what the page holds is the stored one, so
4439    // the last block end is the only thing that ties the index to the length of the page.
4440    let stored_len = page.length as usize - body_len;
4441    if ends.last().copied().unwrap_or_default() as usize != stored_len
4442        || ends.windows(2).any(|pair| pair[0] > pair[1])
4443    {
4444        return Err(invalid("global dictionary blocks do not bound the payload"));
4445    }
4446    Vector::external_text(
4447        LogicalType::Varchar,
4448        Arc::new(NativeText {
4449            file,
4450            values: count,
4451            offsets,
4452            offset_bits,
4453            ranks,
4454            rank_at: page.offset + index_len as u64,
4455            rank_ends,
4456            rank_hashes,
4457            rank_blocks: (0..rank_blocks).map(|_| OnceLock::new()).collect(),
4458            code_bits: code_width(count),
4459            code_ranks: OnceLock::new(),
4460            payload: page.offset + body_len as u64,
4461            ends,
4462            hashes,
4463            blocks: (0..blocks).map(|_| OnceLock::new()).collect(),
4464        }),
4465    )
4466}
4467
4468fn decode(
4469    ty: &LogicalType,
4470    rows: usize,
4471    bytes: &[u8],
4472    global: Option<Arc<Vector>>,
4473) -> Result<Vector> {
4474    let mut cur = Cursor { bytes, at: 0 };
4475    let codec = cur.u8()?;
4476    let flag = cur.u8()?;
4477    let validity = match flag {
4478        0 => Validity::AllValid,
4479        1 => Validity::AllInvalid,
4480        2 => {
4481            let mask = cur.take(rows.div_ceil(8))?;
4482            Validity::from_iter(rows, |row| mask[row / 8] >> (row % 8) & 1 == 1)
4483        }
4484        _ => return Err(invalid("page validity tag differs")),
4485    };
4486    if codec == 1 {
4487        if ty != &LogicalType::Varchar {
4488            return Err(invalid("dictionary codec belongs to a non-string page"));
4489        }
4490        let count = cur.u32()? as usize;
4491        let payload_len = cur.u32()? as usize;
4492        let offset_bytes = cur.take(
4493            (count + 1)
4494                .checked_mul(4)
4495                .ok_or_else(|| invalid("dictionary offset count overflow"))?,
4496        )?;
4497        let offsets = offset_bytes
4498            .chunks_exact(4)
4499            .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4500            .collect::<Vec<_>>();
4501        let payload = cur.take(payload_len)?.to_vec();
4502        if offsets.first() != Some(&0)
4503            || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4504            || offsets.windows(2).any(|pair| pair[0] > pair[1])
4505        {
4506            return Err(invalid("dictionary offsets do not bound the payload"));
4507        }
4508        let mut strings = StringColumn::over(Buffer::from_vec(payload));
4509        for pair in offsets.windows(2) {
4510            strings.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4511        }
4512        let mut codes = Vec::with_capacity(rows);
4513        for _ in 0..rows {
4514            codes.push(cur.u32()?);
4515        }
4516        if codes.iter().any(|code| *code as usize >= count) {
4517            return Err(invalid("dictionary code is out of range"));
4518        }
4519        if cur.at != bytes.len() {
4520            return Err(invalid("dictionary page has trailing bytes"));
4521        }
4522        let dictionary = Vector::flat(LogicalType::Varchar, Data::Varlen(strings))?;
4523        return Ok(Vector::dictionary(codes, dictionary)?.with_validity(validity));
4524    }
4525    if codec == 3 || codec == 4 {
4526        let dictionary = global.ok_or_else(|| invalid("global code page has no dictionary"))?;
4527        let codes = if codec == 4 {
4528            // The cascade holds the whole tail of the page and says how long it is itself, so the
4529            // check that nothing is left over is the one the decoder already makes.
4530            let wide = integer::decode(&bytes[cur.at..])?;
4531            if wide.len() != rows {
4532                return Err(invalid("encoded code page holds the wrong number of rows"));
4533            }
4534            wide.into_iter()
4535                .map(|code| u32::try_from(code).map_err(|_| invalid("code is not a code")))
4536                .collect::<Result<Vec<u32>>>()?
4537        } else {
4538            let mut codes = Vec::with_capacity(rows);
4539            for _ in 0..rows {
4540                codes.push(cur.u32()?);
4541            }
4542            if cur.at != bytes.len() {
4543                return Err(invalid("global code page has trailing bytes"));
4544            }
4545            codes
4546        };
4547        let highest = codes.iter().copied().max();
4548        return Ok(Vector::stable_dictionary_validated(codes, dictionary, highest)?
4549            .with_validity(validity));
4550    }
4551    if codec == 5 {
4552        // The cascade holds the whole tail of the page and says how long it is itself.
4553        let values = integer::decode(&bytes[cur.at..])?;
4554        if values.len() != rows {
4555            return Err(invalid("cascade page holds the wrong number of rows"));
4556        }
4557        let data = narrowed(ty, values)?;
4558        return Ok(Vector::flat(ty.clone(), data)?.with_validity(validity));
4559    }
4560    if codec == 2 {
4561        let width = u32::from(cur.u8()?);
4562        let base = i128::from_le_bytes(cur.take(16)?.try_into().expect("sixteen bytes"));
4563        let count = cur.u32()? as usize;
4564        let mut words = Vec::with_capacity(count);
4565        for _ in 0..count {
4566            words.push(cur.u64()?);
4567        }
4568        if cur.at != bytes.len() {
4569            return Err(invalid("packed page has trailing bytes"));
4570        }
4571        return Ok(Vector::packed(ty.clone(), words, width, base, rows)?.with_validity(validity));
4572    }
4573    if codec != 0 {
4574        return Err(invalid("page codec is unknown"));
4575    }
4576    let data = match ty {
4577        LogicalType::TinyInt => {
4578            let values = cur.take(rows)?;
4579            Data::Int8(values.iter().map(|item| *item as i8).collect::<Vec<_>>().into())
4580        }
4581        LogicalType::UTinyInt => Data::UInt8(cur.take(rows)?.to_vec().into()),
4582        LogicalType::SmallInt => {
4583            let values =
4584                cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4585            Data::Int16(
4586                values
4587                    .chunks_exact(2)
4588                    .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
4589                    .collect::<Vec<_>>()
4590                    .into(),
4591            )
4592        }
4593        LogicalType::USmallInt => {
4594            let values =
4595                cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4596            Data::UInt16(
4597                values
4598                    .chunks_exact(2)
4599                    .map(|item| u16::from_le_bytes(item.try_into().expect("two bytes")))
4600                    .collect::<Vec<_>>()
4601                    .into(),
4602            )
4603        }
4604        LogicalType::UInteger => {
4605            let values =
4606                cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4607            Data::UInt32(
4608                values
4609                    .chunks_exact(4)
4610                    .map(|item| u32::from_le_bytes(item.try_into().expect("four bytes")))
4611                    .collect::<Vec<_>>()
4612                    .into(),
4613            )
4614        }
4615        LogicalType::UBigInt => {
4616            let values =
4617                cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4618            Data::UInt64(
4619                values
4620                    .chunks_exact(8)
4621                    .map(|item| u64::from_le_bytes(item.try_into().expect("eight bytes")))
4622                    .collect::<Vec<_>>()
4623                    .into(),
4624            )
4625        }
4626        LogicalType::Integer | LogicalType::Date => {
4627            let values =
4628                cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4629            Data::Int32(
4630                values
4631                    .chunks_exact(4)
4632                    .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
4633                    .collect::<Vec<_>>()
4634                    .into(),
4635            )
4636        }
4637        LogicalType::BigInt | LogicalType::Timestamp => {
4638            let values =
4639                cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4640            Data::Int64(
4641                values
4642                    .chunks_exact(8)
4643                    .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
4644                    .collect::<Vec<_>>()
4645                    .into(),
4646            )
4647        }
4648        LogicalType::Boolean => {
4649            let values = cur.take(rows)?;
4650            if values.iter().any(|value| *value > 1) {
4651                return Err(invalid("boolean page has another value"));
4652            }
4653            Data::Bool(values.iter().map(|value| *value == 1).collect::<Vec<_>>().into())
4654        }
4655        LogicalType::Varchar => {
4656            let offset_bytes = cur
4657                .take((rows + 1).checked_mul(4).ok_or_else(|| invalid("offset count overflow"))?)?;
4658            let offsets = offset_bytes
4659                .chunks_exact(4)
4660                .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4661                .collect::<Vec<_>>();
4662            let payload = cur.take(bytes.len() - cur.at)?.to_vec();
4663            if offsets.first() != Some(&0)
4664                || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4665                || offsets.windows(2).any(|pair| pair[0] > pair[1])
4666            {
4667                return Err(invalid("string offsets do not bound the payload"));
4668            }
4669            let mut values = StringColumn::over(Buffer::from_vec(payload));
4670            for pair in offsets.windows(2) {
4671                values.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4672            }
4673            Data::Varlen(values)
4674        }
4675        _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
4676    };
4677    if cur.at != bytes.len() {
4678        return Err(invalid("page has trailing bytes"));
4679    }
4680    Ok(Vector::flat(ty.clone(), data)?.with_validity(validity))
4681}
4682
4683#[cfg(test)]
4684mod tests {
4685    use std::fs;
4686    use std::io::{Seek, SeekFrom, Write};
4687    use std::path::PathBuf;
4688    use std::time::{SystemTime, UNIX_EPOCH};
4689
4690    use rudb_common::Value;
4691    use rudb_common::bounds::Op;
4692
4693    use super::*;
4694
4695    #[test]
4696    fn checksum_matches_fixed_vectors() {
4697        assert_eq!(checksum(b""), 0xef46_db37_51d8_e999);
4698        assert_eq!(checksum(b"a"), 0xd24e_c4f1_a98c_6e5b);
4699        assert_eq!(checksum(b"abc"), 0x44bc_2cf5_ad77_0999);
4700    }
4701
4702    fn path(label: &str) -> PathBuf {
4703        let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
4704        std::env::temp_dir().join(format!("rudb-native-{label}-{}-{stamp}.rdb", std::process::id()))
4705    }
4706
4707    /// A read names the offset it wants, so a cursor somebody else moved cannot reach it.
4708    #[test]
4709    fn a_read_at_an_offset_ignores_where_another_thread_left_the_cursor() {
4710        const SPANS: usize = 64;
4711        const SPAN: usize = 512;
4712        let path = path("positional");
4713        let content: Vec<u8> =
4714            (0..SPANS).flat_map(|span| std::iter::repeat_n(span as u8, SPAN)).collect();
4715        fs::write(&path, &content).expect("the file is written");
4716        let file = Arc::new(File::open(&path).expect("the file opens"));
4717        std::thread::scope(|scope| {
4718            for _ in 0..8 {
4719                let file = Arc::clone(&file);
4720                scope.spawn(move || {
4721                    for _ in 0..64 {
4722                        for span in 0..SPANS {
4723                            let mut bytes = [0_u8; SPAN];
4724                            read_at(&file, (span * SPAN) as u64, &mut bytes)
4725                                .expect("the span reads");
4726                            assert!(
4727                                bytes.iter().all(|byte| *byte == span as u8),
4728                                "span {span} came back as {}",
4729                                bytes[0],
4730                            );
4731                        }
4732                    }
4733                });
4734            }
4735        });
4736        let mut past = [0_u8; SPAN];
4737        let end = (SPANS * SPAN) as u64;
4738        let error = read_at(&file, end, &mut past).expect_err("a read past the end is refused");
4739        assert!(error.message().contains("ends before its declared length"), "{error}");
4740        drop(file);
4741        let _ = fs::remove_file(&path);
4742    }
4743
4744    /// The writer records where it put a page and puts it there, whatever the cursor is doing.
4745    ///
4746    /// The cursor is moved between the steps that record an offset, which is what reading the pages
4747    /// back to build the frequencies does on a platform with no `pread`. Without the fix the
4748    /// directory lands on top of a page and the file fails to reopen.
4749    #[test]
4750    fn a_writer_puts_a_page_where_it_said_it_did_wherever_the_cursor_has_got_to() {
4751        let path = path("cursor");
4752        let mut writer = Writer::create(
4753            &path,
4754            "items",
4755            vec![
4756                Field::required("id", LogicalType::Integer),
4757                Field::new("text", LogicalType::Varchar),
4758            ],
4759        )
4760        .expect("new file");
4761        writer.append(&sample()).expect("first part");
4762        writer.file.seek(SeekFrom::Start(0)).expect("the cursor goes back to the header");
4763        writer.append(&sample()).expect("second part");
4764        writer.file.seek(SeekFrom::Start(1)).expect("and somewhere useless again");
4765        writer.finish().expect("commit");
4766        let reader = Reader::open(&path).expect("reopen from disk");
4767        assert_eq!(reader.table().rows(), 6);
4768        let ids = reader.read(0, &[0]).expect("the integer page reads back");
4769        assert_eq!(ids.value_at(0, 0), Value::Integer(4));
4770        assert_eq!(ids.value_at(2, 0), Value::Integer(-2));
4771        let text = reader.read(1, &[1]).expect("the text page reads back");
4772        assert_eq!(text.value_at(1, 0), Value::Null);
4773        assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4774        // Nothing the directory points at may run past the end of the file, which is the shape the
4775        // failure took: a page recorded at an offset the directory had already been written over.
4776        let end = reader.table().stripes().iter().flat_map(|stripe| {
4777            stripe
4778                .pages
4779                .iter()
4780                .map(|page| page.offset + u64::from(page.length))
4781                .chain(std::iter::once(stripe.index.offset + u64::from(stripe.index.length)))
4782        });
4783        let last = end.fold(HEADER, u64::max);
4784        let directory = fs::metadata(&path).expect("the file is there").len();
4785        assert!(last <= directory, "a page runs to {last} in a file of {directory} bytes");
4786        fs::remove_file(path).expect("remove scratch file");
4787    }
4788
4789    /// How long a global dictionary index is, read out of the page's own header.
4790    ///
4791    /// The tests below damage a byte of the order or of the payload, so they need to know where each
4792    /// one starts, and working it out here rather than writing a number down means adding something
4793    /// to the index does not quietly turn one of them into a test that damages the index instead.
4794    fn dictionary_index_len(header: &[u8; DICTIONARY_HEADER]) -> u64 {
4795        let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
4796        let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
4797        let bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
4798        let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
4799        DICTIONARY_HEADER as u64
4800            + offset_bytes(count as usize, bits) as u64
4801            + (blocks + rank_blocks) * 16
4802    }
4803
4804    /// How long the sorted order is, which is where its last block ends.
4805    fn last_rank_end(file: &File, offset: u64, header: &[u8; DICTIONARY_HEADER]) -> u64 {
4806        let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
4807        let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
4808        let bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
4809        let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
4810        let at = offset
4811            + DICTIONARY_HEADER as u64
4812            + offset_bytes(count as usize, bits) as u64
4813            + blocks * 16
4814            + (rank_blocks - 1) * 8;
4815        let mut end = [0; 8];
4816        read_at(file, at, &mut end).expect("the last rank block end");
4817        u64::from_le_bytes(end)
4818    }
4819
4820    fn sample() -> Chunk {
4821        Chunk::new(vec![
4822            Vector::from_values(
4823                LogicalType::Integer,
4824                &[Value::Integer(4), Value::Integer(9), Value::Integer(-2)],
4825            )
4826            .expect("integers"),
4827            Vector::from_values(
4828                LogicalType::Varchar,
4829                &[
4830                    Value::Varchar("alpha".into()),
4831                    Value::Null,
4832                    Value::Varchar("long text after a slash".into()),
4833                ],
4834            )
4835            .expect("strings"),
4836        ])
4837        .expect("matching rows")
4838    }
4839
4840    fn sample_ids() -> Chunk {
4841        Chunk::new(vec![
4842            Vector::flat(LogicalType::Integer, Data::Int32(vec![7, 8, 9].into()))
4843                .expect("integers"),
4844        ])
4845        .expect("one column")
4846    }
4847
4848    #[test]
4849    fn committed_file_reopens_and_reads_only_requested_columns() {
4850        let path = path("reopen");
4851        let mut writer = Writer::create(
4852            &path,
4853            "items",
4854            vec![
4855                Field::required("id", LogicalType::Integer),
4856                Field::new("text", LogicalType::Varchar),
4857            ],
4858        )
4859        .expect("new file");
4860        writer.append(&sample()).expect("first part");
4861        writer.append(&sample()).expect("second part");
4862        writer.finish().expect("commit");
4863        let reader = Reader::open(&path).expect("reopen from disk");
4864        assert_eq!(reader.table().rows(), 6);
4865        // Two appends below the stripe bound are two parts of one stripe, which is the whole point
4866        // of the split: the directory describes the stripe and the scan still reads a part.
4867        assert_eq!(reader.table().stripes().len(), 1);
4868        assert_eq!(reader.parts(), 2);
4869        assert_eq!(reader.part_rows(0), 3);
4870        assert_eq!(reader.part_rows(1), 3);
4871        let text = reader.read(1, &[1]).expect("only text page");
4872        assert_eq!(text.width(), 1);
4873        assert_eq!(text.value_at(1, 0), Value::Null);
4874        assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4875        let sparse = reader.read_sparse(1, &[1]).expect("one part without its whole page");
4876        assert_eq!(sparse.width(), 1);
4877        assert_eq!(sparse.value_at(1, 0), Value::Null);
4878        assert_eq!(sparse.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4879        assert!(!reader.skips_codes(0, 1, &[0]).expect("alpha is in the stripe"));
4880        assert!(!reader.skips_codes(0, 1, &[2]).expect("long text is in the stripe"));
4881        assert!(reader.skips_codes(0, 1, &[3]).expect("unknown code is absent"));
4882        let count = reader.read(0, &[]).expect("no page is needed for count");
4883        assert_eq!(count.len(), 3);
4884        assert!(reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }]));
4885        assert!(!reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(0) }]));
4886        let integers = reader.top_frequencies(0, 1).expect("valid integer synopsis").expect("kept");
4887        assert_eq!(
4888            integers,
4889            vec![(Value::Integer(-2), 2), (Value::Integer(4), 2), (Value::Integer(9), 2),]
4890        );
4891        let strings = reader.top_frequencies(1, 1).expect("valid string synopsis").expect("kept");
4892        assert_eq!(strings.len(), 3);
4893        assert!(strings.contains(&(Value::Null, 2)));
4894        assert!(strings.contains(&(Value::Varchar("alpha".into()), 2)));
4895        assert!(strings.contains(&(Value::Varchar("long text after a slash".into()), 2)));
4896        fs::remove_file(path).expect("remove scratch file");
4897    }
4898
4899    /// Two pipeline instances handing over whole runs, which is what makes the native sink safe to
4900    /// instance.
4901    ///
4902    /// The runs arrive in the order the instances finished reading them rather than in source
4903    /// order, and the second one to finish is the one that read the earlier rows. Each run is still
4904    /// a stripe of its own and the table still reads back in source order, which is the whole of
4905    /// what the writer promises about ordering.
4906    #[test]
4907    fn runs_handed_over_out_of_order_still_read_back_in_source_order() {
4908        let path = path("interleaved-runs");
4909        let mut writer =
4910            Writer::create(&path, "interleaved", vec![Field::new("v", LogicalType::BigInt)])
4911                .expect("new file");
4912        for morsel in [2_u64, 0, 3, 1] {
4913            let parts = (0..4_u64)
4914                .map(|chunk| {
4915                    let first = i64::try_from(morsel * 32 + chunk * 8).expect("small");
4916                    let values =
4917                        (0..8_i64).map(|row| Value::BigInt(first + row)).collect::<Vec<_>>();
4918                    let column =
4919                        Vector::from_values(LogicalType::BigInt, &values).expect("a column");
4920                    ((morsel, chunk), Chunk::new(vec![column]).expect("one column"))
4921                })
4922                .collect::<Vec<_>>();
4923            writer.append_stripe(parts).expect("a stripe");
4924        }
4925        writer.finish().expect("commit");
4926
4927        let reader = Reader::open(&path).expect("valid directory");
4928        assert_eq!(reader.table().stripes().len(), 4, "a run is a stripe of its own");
4929        assert_eq!(reader.table().rows(), 128);
4930        for part in 0..16_usize {
4931            let read = reader.read(part, &[0]).expect("a part back");
4932            for row in 0..8_usize {
4933                let want = i64::try_from(part * 8 + row).expect("small");
4934                assert_eq!(read.value_at(row, 0), Value::BigInt(want), "part {part} row {row}");
4935            }
4936        }
4937        fs::remove_file(path).expect("remove scratch file");
4938    }
4939
4940    /// Runs from different callers may interleave and may not overlap, and the commit is what
4941    /// catches an overlap.
4942    #[test]
4943    fn runs_that_overlap_each_other_are_refused_at_commit() {
4944        let path = path("overlapping-runs");
4945        let mut writer =
4946            Writer::create(&path, "overlapping", vec![Field::new("v", LogicalType::BigInt)])
4947                .expect("new file");
4948        let one = |order: (u64, u64)| {
4949            let column =
4950                Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)]).expect("a column");
4951            (order, Chunk::new(vec![column]).expect("one column"))
4952        };
4953        // The second run sits inside the first rather than after it, which is a thing no instance
4954        // holding its own contiguous run can produce and a thing the file cannot represent.
4955        writer.append_stripe(vec![one((0, 0)), one((0, 2))]).expect("a stripe");
4956        writer.append_stripe(vec![one((0, 1))]).expect("a stripe");
4957        let error = writer.finish().expect_err("the runs overlap");
4958        assert!(error.message().contains("source order"), "{error}");
4959        fs::remove_file(path).expect("remove scratch file");
4960    }
4961
4962    /// A stripe holds [`STRIPE_PARTS`] parts, so a run longer than that is a caller bug rather than
4963    /// something to split, and the writer says so at the door instead of quietly cutting it in two.
4964    #[test]
4965    fn a_run_longer_than_a_stripe_is_refused() {
4966        let path = path("overlong-run");
4967        let mut writer =
4968            Writer::create(&path, "overlong", vec![Field::new("v", LogicalType::BigInt)])
4969                .expect("new file");
4970        let parts = (0..=STRIPE_PARTS)
4971            .map(|at| {
4972                let column = Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)])
4973                    .expect("a column");
4974                let chunk = Chunk::new(vec![column]).expect("one column");
4975                ((0, u64::try_from(at).expect("small")), chunk)
4976            })
4977            .collect::<Vec<_>>();
4978        let error = writer.append_stripe(parts).expect_err("one part too many");
4979        assert!(error.message().contains("more parts than it holds"), "{error}");
4980        fs::remove_file(path).expect("remove scratch file");
4981    }
4982
4983    /// Parts past the stripe bound start a new stripe, and every part stays addressable on its own.
4984    ///
4985    /// This is the shape the format exists for, so both ends of the split are checked here. The
4986    /// directory holds three stripes rather than a hundred and thirty one, and a read of any one
4987    /// part still answers with that part's rows rather than with its whole stripe's.
4988    #[test]
4989    fn parts_past_the_stripe_bound_start_a_new_stripe() {
4990        let path = path("stripe-bound");
4991        let mut writer = Writer::create(
4992            &path,
4993            "items",
4994            vec![
4995                Field::required("id", LogicalType::Integer),
4996                Field::new("text", LogicalType::Varchar),
4997            ],
4998        )
4999        .expect("new file");
5000        let parts = STRIPE_PARTS * 2 + 3;
5001        for part in 0..parts {
5002            let id = part as i32;
5003            let chunk = Chunk::new(vec![
5004                Vector::from_values(
5005                    LogicalType::Integer,
5006                    &[Value::Integer(id), Value::Integer(-id)],
5007                )
5008                .expect("integers"),
5009                Vector::from_values(
5010                    LogicalType::Varchar,
5011                    &[Value::Varchar(format!("value {part}")), Value::Null],
5012                )
5013                .expect("strings"),
5014            ])
5015            .expect("matching rows");
5016            writer.append(&chunk).expect("one part");
5017        }
5018        writer.finish().expect("commit");
5019
5020        let reader = Reader::open(&path).expect("reopen from disk");
5021        assert_eq!(reader.parts(), parts);
5022        assert_eq!(reader.table().rows(), parts * 2);
5023        assert_eq!(reader.table().stripes().len(), parts.div_ceil(STRIPE_PARTS));
5024        assert_eq!(reader.table().stripes()[0].parts(), STRIPE_PARTS);
5025        assert_eq!(reader.table().stripes()[0].rows(), STRIPE_PARTS * 2);
5026        assert_eq!(reader.table().stripes()[2].parts(), 3);
5027        // Backwards on purpose. The reader keeps four stripes a column, so a scan that walks the
5028        // table the other way is what catches a cache that only ever holds what it just read.
5029        for part in (0..parts).rev() {
5030            let dense = reader.read(part, &[0, 1]).expect("a whole page read");
5031            let sparse = reader.read_sparse(part, &[0, 1]).expect("one part read");
5032            for chunk in [&dense, &sparse] {
5033                assert_eq!(chunk.len(), 2, "part {part} has its own row count");
5034                assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5035                assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
5036                assert_eq!(chunk.value_at(0, 1), Value::Varchar(format!("value {part}")));
5037                assert_eq!(chunk.value_at(1, 1), Value::Null);
5038            }
5039        }
5040        // The bounds are merged over the stripe, so they answer for the range the whole stripe
5041        // covers and not for the part that was asked about.
5042        let above = [Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }];
5043        assert!(reader.skips(0, &above), "the first stripe stops at 63");
5044        assert!(!reader.skips(STRIPE_PARTS * 2, &above), "the third stripe reaches 130");
5045        fs::remove_file(path).expect("remove scratch file");
5046    }
5047
5048    /// A scattered value in the column that decides `WHERE UserID = ?`.
5049    fn scattered(n: i64) -> i64 {
5050        n.wrapping_mul(-7_046_029_254_386_353_131)
5051    }
5052
5053    /// A part whose sieve does not hold the constant is skipped, and a range would skip none of them.
5054    ///
5055    /// This is ClickBench query 19 in miniature. The values are spread over the whole of `BIGINT`, so
5056    /// every stripe's bounds cover nearly all of it and rule out nothing, and the part that really
5057    /// holds the value is the only one a scan has to read.
5058    #[test]
5059    fn a_part_is_skipped_when_its_sieve_does_not_hold_the_constant() {
5060        let path = path("sieve-skip");
5061        let mut writer =
5062            Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
5063                .expect("new file");
5064        let parts = STRIPE_PARTS + 3;
5065        // Big enough that the filter is worth its bytes. A part of eight numbers packs to under a
5066        // hundred bytes and the smallest filter there is is sixty nine, so a filter over a part
5067        // that small costs about as much to read as the rows do and is no longer written.
5068        let per_part = 128;
5069        for part in 0..parts {
5070            let held: Vec<Value> = (0..per_part)
5071                .map(|row| Value::BigInt(scattered((part * per_part + row) as i64)))
5072                .collect();
5073            let chunk =
5074                Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
5075                    .expect("one column");
5076            writer.append(&chunk).expect("one part");
5077        }
5078        writer.finish().expect("commit");
5079
5080        let reader = Reader::open(&path).expect("reopen from disk");
5081        let probe = |value: i64| Probe {
5082            column: 0,
5083            op: Op::Equal,
5084            value: Bound::Int(i128::from(scattered(value))),
5085        };
5086        for wanted in [0_i64, (per_part + 1) as i64, (parts * per_part - 1) as i64] {
5087            let tests = [probe(wanted)];
5088            let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &tests)).collect();
5089            let home = wanted as usize / per_part;
5090            assert!(kept.contains(&home), "the part holding {wanted} is read");
5091            // A filter answers maybe, so a part it keeps need not hold the value. Sixty seven parts
5092            // of a hundred and twenty eight numbers each, at a dozen bits a value, is about one
5093            // stray part across the whole file and that is what this leaves room for.
5094            assert!(kept.len() <= 2, "{wanted} keeps {kept:?}, which is more than one stray part");
5095        }
5096        let absent = [probe((parts * per_part) as i64 + 1)];
5097        let kept = (0..parts).filter(|&part| !reader.skips(part, &absent)).count();
5098        assert!(kept <= 1, "{kept} parts of {parts} kept a value no part holds");
5099        // The same probes against the bounds alone, which is what this replaces. A column of
5100        // scattered numbers has a range per stripe that covers nearly the whole type.
5101        let tests = [probe(0)];
5102        assert!(
5103            reader.table().stripes().iter().all(|stripe| !stripe.zone.skips(&tests)),
5104            "the bounds rule out no stripe at all"
5105        );
5106        fs::remove_file(path).expect("remove scratch file");
5107    }
5108
5109    /// A sieve bigger than the part it indexes is not written, and one smaller than it still is.
5110    ///
5111    /// Both columns hold values spread over the whole of `BIGINT`, so neither gets a bitmap and both
5112    /// reach the filter. They differ in what the part costs to read. `spread` is a thousand distinct
5113    /// numbers and packs to eight kilobytes, so a filter of about thirteen hundred bytes is a good
5114    /// trade. `repeated` is the same thousand rows over four numbers in runs and encodes to
5115    /// almost nothing, but the filter is sized for the rows rather than the values it turns out to
5116    /// hold, so it comes out larger than the data. Reading it to decide whether to read the part spends more than
5117    /// the part, every time, and that is the case this drops.
5118    #[test]
5119    fn a_sieve_larger_than_the_part_it_indexes_is_not_written() {
5120        let path = path("sieve-pays");
5121        let fields = vec![
5122            Field::required("spread", LogicalType::BigInt),
5123            Field::required("repeated", LogicalType::BigInt),
5124        ];
5125        let mut writer = Writer::create(&path, "hits", fields).expect("new file");
5126        let parts = 3;
5127        let per_part = 1024;
5128        for part in 0..parts {
5129            let base = (part * per_part) as i64;
5130            let spread: Vec<Value> =
5131                (0..per_part).map(|row| Value::BigInt(scattered(base + row as i64))).collect();
5132            let repeated: Vec<Value> =
5133                (0..per_part).map(|row| Value::BigInt(scattered((row / 256) as i64))).collect();
5134            let chunk = Chunk::new(vec![
5135                Vector::from_values(LogicalType::BigInt, &spread).expect("numbers"),
5136                Vector::from_values(LogicalType::BigInt, &repeated).expect("numbers"),
5137            ])
5138            .expect("two columns");
5139            writer.append(&chunk).expect("one part");
5140        }
5141        writer.finish().expect("commit");
5142
5143        let reader = Reader::open(&path).expect("reopen from disk");
5144        let layout = reader.layout();
5145        let spread = &layout.columns[0];
5146        let repeated = &layout.columns[1];
5147        assert!(spread.sieves > 0, "a column whose parts are worth a filter keeps one");
5148        assert_eq!(
5149            repeated.sieves, 0,
5150            "a column whose filter costs more than its parts keeps none"
5151        );
5152        // Per part this is the rule itself, so it holds over the column as well: a part without a
5153        // sieve adds to one side of this and to nothing on the other.
5154        for column in &layout.columns {
5155            assert!(
5156                column.sieves < column.pages,
5157                "{} spends {} on sieves over {} of data",
5158                column.name,
5159                column.sieves,
5160                column.pages
5161            );
5162        }
5163        // The filter that was kept still does what it is for.
5164        let absent = [Probe {
5165            column: 0,
5166            op: Op::Equal,
5167            value: Bound::Int(i128::from(scattered((parts * per_part) as i64 + 1))),
5168        }];
5169        assert!((0..parts).all(|part| reader.skips(part, &absent)), "no part holds it");
5170        fs::remove_file(path).expect("remove scratch file");
5171    }
5172
5173    /// A damaged sieve page is a part that gets read, not a query that fails.
5174    ///
5175    /// A sieve is an index over rows that are still there and still correct, so losing one costs
5176    /// time and costs no answers. That is the opposite of the membership index beside it, which is
5177    /// the only thing standing between a string page and a wrong answer.
5178    #[test]
5179    fn a_damaged_sieve_page_is_read_through_rather_than_refused() {
5180        let path = path("sieve-damaged");
5181        let mut writer =
5182            Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
5183                .expect("new file");
5184        let rows = 128;
5185        let held: Vec<Value> = (0..rows).map(|row| Value::BigInt(scattered(row))).collect();
5186        let chunk =
5187            Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
5188                .expect("one column");
5189        writer.append(&chunk).expect("one part");
5190        writer.finish().expect("commit");
5191
5192        let page =
5193            Reader::open(&path).expect("reopen").table.stripes[0].sieves[0].expect("a sieve page");
5194        let mut file = OpenOptions::new().write(true).open(&path).expect("open the sieve page");
5195        file.seek(SeekFrom::Start(page.offset + u64::from(page.length) - 1)).expect("seek");
5196        file.write_all(&[0xff]).expect("damage one byte");
5197        drop(file);
5198
5199        let reader = Reader::open(&path).expect("reopen the damaged file");
5200        let absent =
5201            [Probe { column: 0, op: Op::Equal, value: Bound::Int(i128::from(scattered(99))) }];
5202        assert!(!reader.skips(0, &absent), "a sieve that cannot be read skips nothing");
5203        assert_eq!(
5204            reader.read(0, &[0]).expect("the rows are untouched").len(),
5205            usize::try_from(rows).expect("a small count")
5206        );
5207        fs::remove_file(path).expect("remove scratch file");
5208    }
5209
5210    /// Eight workers over one stripe read it once between them.
5211    ///
5212    /// This is the shape a scan actually has. Parts are handed out in order, so every worker on a
5213    /// column crosses into a stripe within a few parts of the others, and before [`Reader::held`]
5214    /// started sharing the read every one of them read the whole page. On the full ClickBench file
5215    /// that was a `MIN(EventDate), MAX(EventDate)` moving 3.2 GB off the disk to look at 400 MB of
5216    /// column, which is most of what a first touch costs.
5217    ///
5218    /// The workers that lose the race still answer, out of the part reads they do instead, which is
5219    /// what the values below are checking.
5220    #[test]
5221    fn workers_that_want_the_same_stripe_read_it_once() {
5222        let path = path("single-flight");
5223        let mut writer =
5224            Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5225                .expect("new file");
5226        for part in 0..STRIPE_PARTS {
5227            let id = part as i32;
5228            let chunk = Chunk::new(vec![
5229                Vector::from_values(
5230                    LogicalType::Integer,
5231                    &[Value::Integer(id), Value::Integer(-id)],
5232                )
5233                .expect("integers"),
5234            ])
5235            .expect("matching rows");
5236            writer.append(&chunk).expect("one part");
5237        }
5238        writer.finish().expect("commit");
5239
5240        let reader = Reader::open(&path).expect("reopen from disk");
5241        assert_eq!(reader.table().stripes().len(), 1, "one stripe is the point of the test");
5242        let barrier = std::sync::Barrier::new(8);
5243        std::thread::scope(|scope| {
5244            for worker in 0..8 {
5245                let reader = &reader;
5246                let barrier = &barrier;
5247                scope.spawn(move || {
5248                    barrier.wait();
5249                    for part in (worker..STRIPE_PARTS).step_by(8) {
5250                        let chunk = reader.read(part, &[0]).expect("a whole page read");
5251                        assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5252                        assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
5253                    }
5254                });
5255            }
5256        });
5257        assert_eq!(reader.pages.load(Atomic::Relaxed), 1, "one stripe, one page read, whoever won");
5258        fs::remove_file(path).expect("remove scratch file");
5259    }
5260
5261    /// Opening a file reads the header and the directory, and nothing that depends on the rows.
5262    ///
5263    /// `spec/stats/04-in-memory.md` section 4.2. There are no statistics in the file yet, so this
5264    /// holds today by not having anything to load, and that is exactly why it is worth pinning now.
5265    /// The change that breaks it is the reasonable looking one: summaries are a few hundred bytes,
5266    /// the next query will want them, so read them on the way past. A process that opened the
5267    /// database to run one trivial query pays for all of it and gets nothing.
5268    ///
5269    /// Two files of the same shape and a thousand times the rows in one of them, opened, and the
5270    /// two openings cost the same. The stripe count is held equal so that the directory is the same
5271    /// size in both, which leaves the rows as the only thing that changed. Anything read out of the
5272    /// data would show up here.
5273    #[test]
5274    fn opening_costs_the_same_over_a_thousand_times_the_rows() {
5275        let opened = |label: &str, rows_per_part: i32| {
5276            let path = path(label);
5277            let mut writer =
5278                Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5279                    .expect("new file");
5280            for part in 0..STRIPE_PARTS * 3 {
5281                // Scrambled rather than sequential, so that the fat file is actually fatter. A run
5282                // of consecutive integers encodes to almost nothing and would leave the two files
5283                // the same size, which would make this test pass for the wrong reason.
5284                let values = (0..rows_per_part)
5285                    .map(|row| {
5286                        Value::Integer((part as i32 * rows_per_part + row).wrapping_mul(2_654_435))
5287                    })
5288                    .collect::<Vec<_>>();
5289                let chunk = Chunk::new(vec![
5290                    Vector::from_values(LogicalType::Integer, &values).expect("integers"),
5291                ])
5292                .expect("matching rows");
5293                writer.append(&chunk).expect("one part");
5294            }
5295            writer.finish().expect("commit");
5296            let reader = Reader::open(&path).expect("reopen from disk");
5297            let size = fs::metadata(&path).expect("the file is there").len();
5298            let out = (reader.reads(), reader.table().stripes().len(), size);
5299            fs::remove_file(path).expect("remove scratch file");
5300            out
5301        };
5302
5303        let (thin, thin_stripes, thin_size) = opened("open-thin", 1);
5304        let (fat, fat_stripes, fat_size) = opened("open-fat", 1000);
5305        assert_eq!(
5306            thin_stripes, fat_stripes,
5307            "the same stripe count is what makes this a fair ask"
5308        );
5309        assert!(
5310            fat_size > thin_size * 50,
5311            "the fat file has to actually be larger, and it is {fat_size} against {thin_size}"
5312        );
5313
5314        assert_eq!(thin.opening.reads, fat.opening.reads, "the same reads either way");
5315        assert_eq!(thin.pages, 0, "opening read a page");
5316        assert_eq!(fat.pages, 0, "opening read a page");
5317        assert_eq!(thin.indexes, 0, "opening read an index");
5318        assert_eq!(fat.indexes, 0, "opening read an index");
5319        // Not exactly equal, because a directory holds offsets and a larger file has larger ones,
5320        // and a handful of bytes of varint is not somebody loading statistics. A factor is.
5321        assert!(
5322            fat.opening.bytes < thin.opening.bytes * 2,
5323            "opening the thin file read {} bytes and the fat one read {}",
5324            thin.opening.bytes,
5325            fat.opening.bytes
5326        );
5327    }
5328
5329    /// The reads a file costs to open are fixed by its shape and not by what ran before.
5330    ///
5331    /// `spec/stats/04-in-memory.md` section 4.3, which is the rule that keeps a plan reproducible:
5332    /// the plan is a function of the data, the generation and the settings, and never of what
5333    /// happened to be in cache. Opening the same file twice in the same process has to cost the
5334    /// same, because a second open that read less would be an open that was about to plan
5335    /// differently.
5336    #[test]
5337    fn two_opens_of_one_file_cost_the_same_and_the_second_is_not_cheaper() {
5338        let path = path("open-twice");
5339        let mut writer =
5340            Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5341                .expect("new file");
5342        for part in 0..STRIPE_PARTS * 3 {
5343            let chunk = Chunk::new(vec![
5344                Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
5345                    .expect("integers"),
5346            ])
5347            .expect("matching rows");
5348            writer.append(&chunk).expect("one part");
5349        }
5350        writer.finish().expect("commit");
5351
5352        let first = Reader::open(&path).expect("open");
5353        // A whole scan in between, so the operating system's page cache is as warm as it gets and
5354        // anything that consulted it would show up in the second open.
5355        for part in 0..first.parts() {
5356            first.read(part, &[0]).expect("a part");
5357        }
5358        assert!(first.reads().pages > 0, "the scan has to have read something");
5359        let second = Reader::open(&path).expect("open again");
5360
5361        assert_eq!(first.reads().opening, second.reads().opening);
5362        assert_eq!(
5363            second.reads().pages,
5364            0,
5365            "the second open read a page off the back of the first"
5366        );
5367        assert_eq!(second.reads().indexes, 0, "the second open read an index it inherited");
5368        fs::remove_file(path).expect("remove scratch file");
5369    }
5370
5371    /// A scan reads a stripe's index once for the whole scan, not once per part that misses.
5372    ///
5373    /// The page cache holds four stripes and an index used to ride inside it, so a table with more
5374    /// stripes than that read the index again every time a stripe came back around. The index is a
5375    /// few hundred bytes and the page is a quarter of a megabyte, which is why they are now under
5376    /// different budgets. This is the test that keeps them there, since the saving is small enough
5377    /// that nothing in a benchmark would notice it going away again.
5378    #[test]
5379    fn an_index_is_read_once_per_stripe_however_often_the_page_is_evicted() {
5380        let path = path("index-cache");
5381        let mut writer =
5382            Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5383                .expect("new file");
5384        let parts = STRIPE_PARTS * (CACHED_STRIPES_PER_COLUMN + 2);
5385        for part in 0..parts {
5386            let id = part as i32;
5387            let chunk = Chunk::new(vec![
5388                Vector::from_values(LogicalType::Integer, &[Value::Integer(id)]).expect("integers"),
5389            ])
5390            .expect("matching rows");
5391            writer.append(&chunk).expect("one part");
5392        }
5393        writer.finish().expect("commit");
5394
5395        let reader = Reader::open(&path).expect("reopen from disk");
5396        let stripes = reader.table().stripes().len();
5397        assert!(stripes > CACHED_STRIPES_PER_COLUMN, "the page cache has to be too small for this");
5398        // Twice over, so that the second pass finds every page evicted and every index kept.
5399        for _ in 0..2 {
5400            for part in 0..parts {
5401                let chunk = reader.read(part, &[0]).expect("a part");
5402                assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5403            }
5404        }
5405        assert_eq!(reader.indexes.load(Atomic::Relaxed), stripes, "one index read per stripe");
5406        assert!(
5407            reader.pages.load(Atomic::Relaxed) > stripes,
5408            "the pages are the ones that get read again, which is what makes the index count mean \
5409             something"
5410        );
5411        fs::remove_file(path).expect("remove scratch file");
5412    }
5413
5414    /// A worker per stripe reads its stripe once, once the cache has been told how many there are.
5415    ///
5416    /// This is the shape a scan has when it hands out a whole stripe per morsel rather than a part.
5417    /// Nobody races for a page any more, but every worker holds a different one for the length of a
5418    /// stripe, so a cache that keeps four pages while eight workers are in eight stripes evicts
5419    /// every one of them before its owner has finished with it, and the owner reads a quarter of a
5420    /// megabyte again for the next part. The barrier is what makes that certain rather than likely:
5421    /// without it a worker can run a whole stripe before the next one starts and never collide.
5422    #[test]
5423    fn a_worker_per_stripe_reads_its_page_once_when_the_cache_was_told_to_expect_it() {
5424        let workers = CACHED_STRIPES_PER_COLUMN + 4;
5425        let path = path("stripe-per-worker");
5426        let mut writer =
5427            Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5428                .expect("new file");
5429        for part in 0..STRIPE_PARTS * workers {
5430            let chunk = Chunk::new(vec![
5431                Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
5432                    .expect("integers"),
5433            ])
5434            .expect("matching rows");
5435            writer.append(&chunk).expect("one part");
5436        }
5437        writer.finish().expect("commit");
5438
5439        let read = |told: bool| {
5440            let reader = Reader::open(&path).expect("reopen from disk");
5441            assert_eq!(reader.table().stripes().len(), workers, "a stripe per worker");
5442            if told {
5443                reader.keep_stripes(workers);
5444            }
5445            let barrier = std::sync::Barrier::new(workers);
5446            std::thread::scope(|scope| {
5447                for (worker, run) in reader.stripe_parts().into_iter().enumerate() {
5448                    let reader = &reader;
5449                    let barrier = &barrier;
5450                    scope.spawn(move || {
5451                        for part in run {
5452                            barrier.wait();
5453                            let chunk = reader.read(part, &[0]).expect("a part of my own stripe");
5454                            assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5455                        }
5456                        assert!(worker < workers);
5457                    });
5458                }
5459            });
5460            reader.pages.load(Atomic::Relaxed)
5461        };
5462
5463        assert_eq!(read(true), workers, "one page read per stripe and no more");
5464        assert!(read(false) > workers, "a cache that small is read again on every part");
5465        fs::remove_file(path).expect("remove scratch file");
5466    }
5467
5468    /// A damaged index page is caught before anything decodes a part out of it.
5469    ///
5470    /// The index is the one structure a reader trusts to find bytes with, so it carries a checksum
5471    /// per column section rather than one for the page, and this is what says that check runs.
5472    #[test]
5473    fn a_damaged_index_page_is_an_error() {
5474        let path = path("damaged-index");
5475        let mut writer =
5476            Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5477                .expect("new file");
5478        writer.append(&sample_ids()).expect("first part");
5479        writer.append(&sample_ids()).expect("second part");
5480        writer.finish().expect("commit");
5481
5482        let reader = Reader::open(&path).expect("valid directory");
5483        let index = reader.table.stripes[0].index;
5484        let mut byte = [0; 1];
5485        read_at(&reader.file, index.offset, &mut byte).expect("the first part length");
5486        let mut file = OpenOptions::new().write(true).open(&path).expect("open index page");
5487        file.seek(SeekFrom::Start(index.offset)).expect("index start");
5488        file.write_all(&[!byte[0]]).expect("damage the first part length");
5489        let error = reader.read(1, &[0]).expect_err("a damaged index must not be used");
5490        assert!(error.message().contains("index page section checksum differs"), "{error}");
5491        fs::remove_file(path).expect("remove scratch file");
5492    }
5493
5494    /// Every integer width the format knows about, written and read back.
5495    ///
5496    /// The unsigned ones are the reason ClickBench can be stored at all: `hits` types `EventDate`
5497    /// as `USMALLINT`, and one unsupported column meant the whole table was refused. The extremes
5498    /// are in here on purpose, because a width that round trips through the wrong signedness only
5499    /// goes wrong at the end of its range.
5500    #[test]
5501    fn every_integer_width_round_trips_through_a_page() {
5502        let path = path("integer-widths");
5503        let columns = [
5504            (LogicalType::TinyInt, vec![Value::TinyInt(i8::MIN), Value::TinyInt(i8::MAX)]),
5505            (LogicalType::UTinyInt, vec![Value::UTinyInt(0), Value::UTinyInt(u8::MAX)]),
5506            (LogicalType::SmallInt, vec![Value::SmallInt(i16::MIN), Value::SmallInt(i16::MAX)]),
5507            (LogicalType::USmallInt, vec![Value::USmallInt(0), Value::USmallInt(u16::MAX)]),
5508            (LogicalType::Integer, vec![Value::Integer(i32::MIN), Value::Integer(i32::MAX)]),
5509            (LogicalType::UInteger, vec![Value::UInteger(0), Value::UInteger(u32::MAX)]),
5510            (LogicalType::BigInt, vec![Value::BigInt(i64::MIN), Value::BigInt(i64::MAX)]),
5511            (LogicalType::UBigInt, vec![Value::UBigInt(0), Value::UBigInt(u64::MAX)]),
5512        ];
5513        let fields = columns
5514            .iter()
5515            .enumerate()
5516            .map(|(at, (ty, _))| Field::required(format!("c{at}"), ty.clone()))
5517            .collect::<Vec<_>>();
5518        let vectors = columns
5519            .iter()
5520            .map(|(ty, values)| Vector::from_values(ty.clone(), values).expect("a vector"))
5521            .collect::<Vec<_>>();
5522        let mut writer = Writer::create(&path, "widths", fields).expect("new file");
5523        writer.append(&Chunk::new(vectors).expect("matching rows")).expect("one stripe");
5524        writer.finish().expect("commit");
5525
5526        let reader = Reader::open(&path).expect("reopen from disk");
5527        let wanted = (0..columns.len()).collect::<Vec<_>>();
5528        let read = reader.read(0, &wanted).expect("every column");
5529        assert_eq!(read.len(), 2);
5530        // row at a time: each column has its own type and its own pair of extremes.
5531        for (at, (ty, values)) in columns.iter().enumerate() {
5532            assert_eq!(read.value_at(0, at), values[0], "the low end of {ty}");
5533            assert_eq!(read.value_at(1, at), values[1], "the high end of {ty}");
5534        }
5535        fs::remove_file(path).expect("remove scratch file");
5536    }
5537
5538    #[test]
5539    fn numeric_frequency_candidates_keep_bounded_row_ordinals() {
5540        let path = path("frequency-ordinals");
5541        let mut writer =
5542            Writer::create(&path, "items", vec![Field::required("id", LogicalType::BigInt)])
5543                .expect("new file");
5544        let mut values = Vec::new();
5545        for leader in 0..10_i64 {
5546            values.extend(std::iter::repeat_n(leader, 100));
5547        }
5548        values.extend(1_000_i64..41_000);
5549        for part in values.chunks(1_024) {
5550            let vector = Vector::flat(LogicalType::BigInt, Data::Int64(part.to_vec().into()))
5551                .expect("big integers");
5552            writer.append(&Chunk::new(vec![vector]).expect("one column")).expect("one stripe");
5553        }
5554        writer.finish().expect("commit");
5555
5556        let reader = Reader::open(&path).expect("reopen from disk");
5557        let occurrences =
5558            reader.frequency_occurrences(0).expect("valid metadata").expect("bounded ordinals");
5559        assert!(occurrences.omitted_max < 100);
5560        assert!(occurrences.ordinals.len() <= FREQUENCY_ORDINALS);
5561        assert!(occurrences.ordinals.windows(2).all(|pair| pair[0] < pair[1]));
5562        assert_eq!(&occurrences.ordinals[..1_000], &(0_u64..1_000).collect::<Vec<_>>());
5563        fs::remove_file(path).expect("remove scratch file");
5564    }
5565
5566    /// The bug this is here for cost a 43 GB ClickBench table and an hour of reloading it. The
5567    /// format went from 11 to 12, every binary built after that said "magic or major version is
5568    /// unsupported" about the file, and there was no way to tell from the message whether the path
5569    /// was wrong, the file was truncated, or it was ours and simply older. The number this build
5570    /// wants is the whole answer and it was the one thing the message did not carry.
5571    #[test]
5572    fn a_file_from_another_format_says_which_format_it_is() {
5573        let older = path("older-format");
5574        let mut writer =
5575            Writer::create(&older, "items", vec![Field::new("id", LogicalType::Integer)])
5576                .expect("new file");
5577        let chunk = Chunk::new(vec![
5578            Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
5579                .expect("integers"),
5580        ])
5581        .expect("chunk");
5582        writer.append(&chunk).expect("page written");
5583        writer.finish().expect("commit");
5584
5585        let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
5586        file.seek(SeekFrom::Start(8)).expect("the version follows the magic");
5587        file.write_all(&(FORMAT - 1).to_le_bytes()).expect("write an older version");
5588        drop(file);
5589        let complaint = Reader::open(&older).expect_err("an older format is refused").to_string();
5590        assert!(complaint.contains(&format!("format {}", FORMAT - 1)), "{complaint}");
5591        assert!(complaint.contains(&format!("format {FORMAT}")), "{complaint}");
5592
5593        let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
5594        file.seek(SeekFrom::Start(0)).expect("the magic is first");
5595        file.write_all(b"NOTRUDB!").expect("write another engine's magic");
5596        drop(file);
5597        let complaint = Reader::open(&older).expect_err("a foreign file is refused").to_string();
5598        assert!(complaint.contains("magic"), "{complaint}");
5599        assert!(!complaint.contains("format"), "a version has nothing to do with it: {complaint}");
5600        fs::remove_file(older).expect("remove scratch file");
5601    }
5602
5603    #[test]
5604    fn an_unfinished_or_damaged_file_does_not_answer_with_partial_rows() {
5605        let unfinished = path("unfinished");
5606        let mut writer =
5607            Writer::create(&unfinished, "items", vec![Field::new("id", LogicalType::Integer)])
5608                .expect("new file");
5609        let chunk = Chunk::new(vec![
5610            Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
5611                .expect("integers"),
5612        ])
5613        .expect("chunk");
5614        writer.append(&chunk).expect("page written");
5615        drop(writer);
5616        assert!(Reader::open(&unfinished).is_err(), "no directory was committed");
5617        fs::remove_file(unfinished).expect("remove scratch file");
5618
5619        let damaged = path("damaged");
5620        let mut writer =
5621            Writer::create(&damaged, "items", vec![Field::new("id", LogicalType::Integer)])
5622                .expect("new file");
5623        writer.append(&chunk).expect("page written");
5624        writer.finish().expect("commit");
5625        let reader = Reader::open(&damaged).expect("valid directory");
5626        let mut file =
5627            OpenOptions::new().write(true).open(&damaged).expect("open for a damaged page");
5628        file.seek(SeekFrom::Start(HEADER + 1)).expect("inside first page");
5629        file.write_all(&[255]).expect("damage one byte");
5630        assert!(reader.read(0, &[0]).is_err(), "page checksum rejects corruption");
5631        fs::remove_file(damaged).expect("remove scratch file");
5632    }
5633
5634    #[test]
5635    fn damaged_lazy_dictionary_payload_is_an_error() {
5636        let path = path("damaged-dictionary");
5637        let mut writer = Writer::create(
5638            &path,
5639            "items",
5640            vec![
5641                Field::required("id", LogicalType::Integer),
5642                Field::new("text", LogicalType::Varchar),
5643            ],
5644        )
5645        .expect("new file");
5646        writer.append(&sample()).expect("stripe written");
5647        writer.finish().expect("commit");
5648
5649        let reader = Reader::open(&path).expect("valid directory");
5650        let dictionary = reader.table.dictionaries[1].expect("string dictionary page");
5651        // Read the count out of the page rather than writing it here, so that adding something
5652        // else to the index does not silently turn this into a test that damages the index.
5653        let mut header = [0; DICTIONARY_HEADER];
5654        read_at(&reader.file, dictionary.offset, &mut header).expect("dictionary header");
5655        let index_len = dictionary_index_len(&header);
5656        let rank_len = last_rank_end(&reader.file, dictionary.offset, &header);
5657        let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5658        file.seek(SeekFrom::Start(dictionary.offset + index_len + rank_len))
5659            .expect("inside dictionary payload");
5660        file.write_all(&[255]).expect("damage dictionary payload");
5661
5662        let chunk = reader.read(0, &[1]).expect("code page and dictionary index remain valid");
5663        let error =
5664            chunk.validate_external().expect_err("payload corruption must reach the caller");
5665        assert!(error.message().contains("payload checksum differs"), "{error}");
5666        fs::remove_file(path).expect("remove scratch file");
5667    }
5668
5669    /// A payload of many blocks reads and checks every block of it.
5670    ///
5671    /// The test above has a dictionary of three values, which is one block, so it says nothing
5672    /// about a reader finding the right block among many. This one has thirty thousand values,
5673    /// which is thirty blocks, and it reads a value out of the first block and a value out of the
5674    /// last and then damages the last and asks for it again.
5675    #[test]
5676    fn a_dictionary_over_many_blocks_checks_every_block_of_it() {
5677        let path = path("dictionary-blocks");
5678        let value =
5679            |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5680        let parts = 30;
5681        let per_part = 1000;
5682        let mut writer =
5683            Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5684                .expect("new file");
5685        for part in 0..parts {
5686            let values = (0..per_part)
5687                .map(|row| Value::Varchar(value(part * per_part + row)))
5688                .collect::<Vec<_>>();
5689            let chunk = Chunk::new(vec![
5690                Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5691            ])
5692            .expect("matching rows");
5693            writer.append(&chunk).expect("a part");
5694        }
5695        writer.finish().expect("commit");
5696
5697        let reader = Reader::open(&path).expect("reopen from disk");
5698        let dictionary = reader.table.dictionaries[0].expect("string dictionary page");
5699        assert!(
5700            parts * per_part > TEXT_PAYLOAD_VALUES * 4,
5701            "the dictionary has to be several blocks for this to be testing anything"
5702        );
5703        for part in [0, parts - 1] {
5704            let chunk = reader.read(part, &[0]).expect("a part");
5705            chunk.validate_external().expect("every payload block checks out");
5706            assert_eq!(chunk.value_at(0, 0), Value::Varchar(value(part * per_part)));
5707        }
5708
5709        let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5710        file.seek(SeekFrom::Start(dictionary.offset + u64::from(dictionary.length) - 4))
5711            .expect("the last bytes of the page are payload");
5712        file.write_all(&[255]).expect("damage the last payload block");
5713        let reader = Reader::open(&path).expect("the directory and the index are untouched");
5714        let chunk = reader.read(parts - 1, &[0]).expect("the code page remains valid");
5715        let error = chunk.validate_external().expect_err("the damage must reach the caller");
5716        assert!(error.message().contains("payload checksum differs"), "{error}");
5717        fs::remove_file(path).expect("remove scratch file");
5718    }
5719
5720    /// Values of different lengths read back where the offsets say they do.
5721    ///
5722    /// The offsets are packed at one width for the column, they are relative to the payload block a
5723    /// value lands in, and they go in runs of half a block, so there are two boundaries where the
5724    /// arithmetic could be off by one and neither shows up on values that are all the same length.
5725    /// This writes 5,000 values whose lengths cycle through a wide range and reads every one back,
5726    /// so the first value of a block, the last value of a run and the last value of a block are all
5727    /// covered several times over. An empty value is in the cycle because a zero length span is the
5728    /// case the reader short circuits.
5729    #[test]
5730    fn values_of_different_lengths_read_back_out_of_packed_offsets() {
5731        let path = path("dictionary-offsets");
5732        let value = |row: usize| {
5733            if row % 511 == 3 { String::new() } else { "x".repeat(row % 97) + &format!("{row:05}") }
5734        };
5735        let rows = 5_000;
5736        let mut writer =
5737            Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5738                .expect("new file");
5739        let values = (0..rows).map(|row| Value::Varchar(value(row))).collect::<Vec<_>>();
5740        for part in values.chunks(1_000) {
5741            let chunk =
5742                Chunk::new(vec![Vector::from_values(LogicalType::Varchar, part).expect("strings")])
5743                    .expect("matching rows");
5744            writer.append(&chunk).expect("a part");
5745        }
5746        writer.finish().expect("commit");
5747
5748        let reader = Reader::open(&path).expect("reopen from disk");
5749        assert!(
5750            rows > TEXT_PAYLOAD_VALUES * 4,
5751            "the dictionary has to be several blocks for this to be testing anything"
5752        );
5753        for part in 0..rows / 1_000 {
5754            let chunk = reader.read(part, &[0]).expect("a part");
5755            for row in 0..1_000 {
5756                let row = part * 1_000 + row;
5757                assert_eq!(
5758                    chunk.value_at(row % 1_000, 0),
5759                    Value::Varchar(value(row)),
5760                    "value {row}"
5761                );
5762            }
5763        }
5764        fs::remove_file(path).expect("remove scratch file");
5765    }
5766
5767    /// Every worker of a scan wants the dictionary at the same moment and one of them fetches it.
5768    ///
5769    /// Asking a `OnceLock` whether it holds something answers the question a worker that already has
5770    /// the dictionary is asking and not the one a worker without it is asking, which is whether
5771    /// somebody is already on their way with it. Sixteen workers that all miss will all read the
5772    /// page, all verify it and all decode it, and fifteen will drop the result. Nothing about that
5773    /// is incorrect, which is why it went unnoticed, and it showed up as ClickBench 38 getting
5774    /// slower when the scan in front of it got faster and stopped staggering the arrivals.
5775    ///
5776    /// The barrier is what makes the test about that rather than about luck. Without it the first
5777    /// thread is usually finished before the last one starts and the count is one either way.
5778    #[test]
5779    fn a_global_dictionary_is_opened_once_however_many_workers_ask_at_once() {
5780        let path = path("dictionary-once");
5781        let parts = 8;
5782        let per_part = 500;
5783        let value =
5784            |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5785        let mut writer =
5786            Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5787                .expect("new file");
5788        for part in 0..parts {
5789            let values = (0..per_part)
5790                .map(|row| Value::Varchar(value(part * per_part + row)))
5791                .collect::<Vec<_>>();
5792            let chunk = Chunk::new(vec![
5793                Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5794            ])
5795            .expect("matching rows");
5796            writer.append(&chunk).expect("a part");
5797        }
5798        writer.finish().expect("commit");
5799
5800        let reader = Reader::open(&path).expect("reopen from disk");
5801        assert!(reader.table.dictionaries[0].is_some(), "the column has to have one to share");
5802        assert_eq!(reader.reads().dictionaries, 0, "opening the file does not open a dictionary");
5803
5804        let workers = 16;
5805        let gate = std::sync::Barrier::new(workers);
5806        std::thread::scope(|scope| {
5807            for worker in 0..workers {
5808                let reader = reader.clone();
5809                let gate = &gate;
5810                scope.spawn(move || {
5811                    gate.wait();
5812                    let chunk = reader.read(worker % parts, &[0]).expect("a part");
5813                    assert_eq!(
5814                        chunk.value_at(0, 0),
5815                        Value::Varchar(value((worker % parts) * per_part))
5816                    );
5817                });
5818            }
5819        });
5820
5821        assert_eq!(reader.reads().dictionaries, 1, "sixteen workers, one dictionary, one open");
5822        fs::remove_file(path).expect("remove scratch file");
5823    }
5824
5825    /// The sorted order sits outside the index the page checksum covers, because a query that
5826    /// never searches a dictionary should not read it, so it carries its own checksums and this is
5827    /// what says they are checked. A search that trusted a damaged order would give a wrong answer
5828    /// rather than a slow one.
5829    #[test]
5830    fn a_damaged_sorted_order_is_an_error() {
5831        let path = path("damaged-order");
5832        let mut writer = Writer::create(
5833            &path,
5834            "items",
5835            vec![
5836                Field::required("id", LogicalType::Integer),
5837                Field::new("text", LogicalType::Varchar),
5838            ],
5839        )
5840        .expect("new file");
5841        writer.append(&sample()).expect("stripe written");
5842        writer.finish().expect("commit");
5843
5844        let reader = Reader::open(&path).expect("valid directory");
5845        let page = reader.table.dictionaries[1].expect("string dictionary page");
5846        let mut header = [0; DICTIONARY_HEADER];
5847        read_at(&reader.file, page.offset, &mut header).expect("dictionary header");
5848        let index_len = dictionary_index_len(&header);
5849        let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5850        file.seek(SeekFrom::Start(page.offset + index_len)).expect("the first head");
5851        file.write_all(&[255]).expect("damage the order");
5852
5853        let dictionary = reader.dictionary(1).expect("read").expect("a string column has one");
5854        let error = dictionary.compare_rank(0, b"anything").expect_err("a damaged order is caught");
5855        assert!(error.message().contains("rank checksum differs"), "{error}");
5856        fs::remove_file(path).expect("remove scratch file");
5857    }
5858
5859    /// Codes stay in first appearance order and the sorted order is written beside them, so a
5860    /// reader can put the values back in order without the writer having had to know them all
5861    /// before it handed out the first code.
5862    #[test]
5863    fn a_global_dictionary_carries_the_sorted_order_of_its_values() {
5864        // Chosen so the sort cannot be decided on the first eight bytes alone. Three values share
5865        // a nine byte prefix, one is a prefix of another, and one is empty.
5866        let spellings = ["overlong1z", "b", "", "overlong1a", "overlong", "ab", "a", "overlong1"];
5867        let path = path("dictionary-order");
5868        let mut writer =
5869            Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
5870                .expect("new file");
5871        writer
5872            .append(
5873                &Chunk::new(vec![
5874                    Vector::from_values(
5875                        LogicalType::Varchar,
5876                        &spellings.map(|text| Value::Varchar(text.into())),
5877                    )
5878                    .expect("strings"),
5879                ])
5880                .expect("one column"),
5881            )
5882            .expect("stripe written");
5883        writer.finish().expect("commit");
5884
5885        let reader = Reader::open(&path).expect("valid directory");
5886        let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
5887        let count = dictionary.ranks().expect("a v10 file stores one");
5888        assert_eq!(count, spellings.len(), "every distinct value has a rank");
5889        let order = (0..count)
5890            .map(|rank| dictionary.code_at_rank(rank).expect("a code"))
5891            .collect::<Vec<_>>();
5892        let mut seen = order.clone();
5893        seen.sort_unstable();
5894        assert_eq!(seen, (0..spellings.len() as u32).collect::<Vec<_>>(), "a permutation of codes");
5895
5896        let ranked = order
5897            .iter()
5898            .map(|&code| {
5899                dictionary.try_bytes_at(code as usize).expect("read").expect("a value").to_vec()
5900            })
5901            .collect::<Vec<_>>();
5902        let mut expected = spellings.map(|text| text.as_bytes().to_vec()).to_vec();
5903        expected.sort();
5904        assert_eq!(ranked, expected, "rank order is value order");
5905
5906        // What a search asks, on the values themselves rather than through a kernel, so that a
5907        // file whose heads disagree with its bytes is caught here rather than as a wrong answer.
5908        for (rank, value) in expected.iter().enumerate() {
5909            assert_eq!(
5910                dictionary.compare_rank(rank, value).expect("compare"),
5911                Ordering::Equal,
5912                "rank {rank} is its own value"
5913            );
5914            if rank > 0 {
5915                assert_eq!(
5916                    dictionary.compare_rank(rank - 1, value).expect("compare"),
5917                    Ordering::Less,
5918                    "rank {rank} follows the one before it"
5919                );
5920            }
5921        }
5922        fs::remove_file(path).expect("remove scratch file");
5923    }
5924
5925    #[test]
5926    fn damaged_membership_cannot_skip_a_string_page() {
5927        let path = path("damaged-membership");
5928        let mut writer = Writer::create(
5929            &path,
5930            "items",
5931            vec![
5932                Field::required("id", LogicalType::Integer),
5933                Field::new("text", LogicalType::Varchar),
5934            ],
5935        )
5936        .expect("new file");
5937        writer.append(&sample()).expect("stripe written");
5938        writer.finish().expect("commit");
5939
5940        let reader = Reader::open(&path).expect("valid directory");
5941        let membership = reader.table.stripes[0].memberships[1].expect("string membership");
5942        let mut file = OpenOptions::new().write(true).open(&path).expect("open membership page");
5943        file.seek(SeekFrom::Start(membership.offset)).expect("membership start");
5944        file.write_all(&[255]).expect("damage membership");
5945        let error = reader.skips_codes(0, 1, &[3]).expect_err("corruption must not skip rows");
5946        assert!(error.message().contains("membership page checksum differs"), "{error}");
5947        fs::remove_file(path).expect("remove scratch file");
5948    }
5949
5950    #[test]
5951    fn membership_delta_stream_is_sorted_exact_and_bounded() {
5952        let unique = unique_codes(&[900, 4, 4, 72, 9, u32::MAX]);
5953        assert_eq!(unique, [4, 9, 72, 900, u32::MAX]);
5954        let encoded = encode_membership(&unique);
5955        assert_eq!(
5956            decode_membership(&encoded).expect("valid membership"),
5957            [4, 9, 72, 900, u32::MAX]
5958        );
5959        // A stripe's index is the union of its parts', so a code in two of them is in it once and
5960        // the result is still one ascending run of deltas.
5961        let merged = merged_codes(vec![vec![4, 900], vec![9, 900, u32::MAX], vec![72]]);
5962        assert_eq!(merged, [4, 9, 72, 900, u32::MAX]);
5963        assert_eq!(
5964            decode_membership(&encode_membership(&merged)).expect("valid membership"),
5965            unique
5966        );
5967        assert!(decode_membership(&[1, 0x80]).is_err(), "a truncated varint is invalid");
5968        assert!(
5969            decode_membership(&[1, 0xff, 0xff, 0xff, 0xff, 0x10]).is_err(),
5970            "a value past u32 is invalid"
5971        );
5972    }
5973
5974    #[test]
5975    fn a_global_dictionary_may_be_larger_than_one_column_page() {
5976        let dictionary = Page {
5977            offset: HEADER,
5978            length: u32::try_from(MAX_PAGE + 1).expect("the page bound fits on disk"),
5979            hash: 0,
5980        };
5981        let table = Table {
5982            name: "items".to_owned(),
5983            fields: vec![Field::new("text", LogicalType::Varchar)],
5984            stripes: Vec::new(),
5985            rows: 0,
5986            dictionaries: vec![Some(dictionary)],
5987            distincts: vec![None],
5988            frequencies: vec![None],
5989        };
5990        let directory = encode_directory(&table).expect("directory");
5991        let file_size = dictionary.offset + u64::from(dictionary.length) + 1;
5992
5993        let decoded = decode_directory(&directory, file_size).expect("large lazy dictionary");
5994        assert_eq!(decoded.dictionaries[0].expect("dictionary").length, dictionary.length);
5995    }
5996
5997    #[test]
5998    fn a_column_with_one_value_everywhere_costs_almost_nothing_a_row() {
5999        let path = path("constant-codes");
6000        let mut writer =
6001            Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
6002                .expect("new file");
6003        let empty = vec![Value::Varchar(String::new()); 1024];
6004        for _ in 0..4 {
6005            let column = Vector::from_values(LogicalType::Varchar, &empty).expect("strings");
6006            writer.append(&Chunk::new(vec![column]).expect("one column")).expect("a part");
6007        }
6008        writer.finish().expect("commit");
6009
6010        let reader = Reader::open(&path).expect("valid directory");
6011        let pages = reader.layout().columns.first().expect("one column").pages;
6012        // This column used to cost four bytes a row, 16,384 of them, the same as a column of four
6013        // thousand distinct URLs would. The cascade calls each part a constant, so what is left is
6014        // a tag, a count and the value, and the row count stops being what drives the number.
6015        assert!(pages < 256, "{pages} bytes of pages for 4,096 rows of one value");
6016        let read = reader.read(3, &[0]).expect("the last part back");
6017        assert_eq!(read.value_at(0, 0), Value::Varchar(String::new()));
6018        assert_eq!(read.value_at(1023, 0), Value::Varchar(String::new()));
6019        fs::remove_file(path).expect("remove scratch file");
6020    }
6021
6022    #[test]
6023    fn a_cascade_value_too_wide_for_its_column_is_refused_rather_than_cut() {
6024        // What a damaged page looks like from here: the cascade decoded, so the bytes are not
6025        // truncated, but the values do not belong to the column the directory says they do.
6026        let over = vec![i64::from(i32::MAX) + 1];
6027        let error = narrowed(&LogicalType::Integer, over).expect_err("a page that disagrees");
6028        assert!(format!("{error}").contains("not of its type"), "{error}");
6029        assert!(narrowed(&LogicalType::BigInt, vec![i64::MIN]).is_ok(), "bigint holds all of i64");
6030        assert!(narrowed(&LogicalType::Varchar, vec![0]).is_err(), "strings are not integers");
6031    }
6032
6033    #[test]
6034    fn a_code_stream_the_cascade_cannot_shrink_is_left_alone() {
6035        // A shift register rather than a run, because an arithmetic run is the one wide shape the
6036        // cascade does shrink. This is what a column with tens of millions of distinct values hands
6037        // over: full width codes with no order to them.
6038        let mut state: u32 = 0x9e37_79b9;
6039        let spread: Vec<u32> = (0..1024)
6040            .map(|_| {
6041                state ^= state << 13;
6042                state ^= state >> 17;
6043                state ^= state << 5;
6044                state
6045            })
6046            .collect();
6047        assert_eq!(encoded_codes(&spread).expect("no failure"), None);
6048        let near: Vec<u32> = (0..1024).collect();
6049        let coded = encoded_codes(&near).expect("no failure").expect("counting up is packable");
6050        assert!(coded.len() < near.len() * 4, "{} bytes for a run of 1,024", coded.len());
6051    }
6052
6053    /// The columns of a stripe are encoded on whichever thread got to them, so the one thing that
6054    /// must not depend on which thread that was is the file. Two writes of the same rows are
6055    /// compared byte for byte rather than value for value, because a dictionary that two columns
6056    /// somehow shared would still read back correctly and would hand out its codes in the order the
6057    /// threads happened to run in, which is exactly what this is here to catch.
6058    #[test]
6059    fn two_writes_of_the_same_rows_give_the_same_bytes() {
6060        fn written(path: &PathBuf) {
6061            let fields = (0..40)
6062                .map(|column| {
6063                    let ty =
6064                        if column % 4 == 0 { LogicalType::Varchar } else { LogicalType::BigInt };
6065                    Field::new(format!("c{column}"), ty)
6066                })
6067                .collect::<Vec<_>>();
6068            let mut writer = Writer::create(path, "wide", fields).expect("new file");
6069            for part in 0..70_u64 {
6070                let columns = (0..40)
6071                    .map(|column| {
6072                        let values = (0..64_u64)
6073                            .map(|row| {
6074                                let seed = part.wrapping_mul(31).wrapping_add(row);
6075                                if column % 4 == 0 {
6076                                    Value::Varchar(format!("v{}", seed % 17))
6077                                } else {
6078                                    Value::BigInt(i64::try_from(seed % 97).expect("small"))
6079                                }
6080                            })
6081                            .collect::<Vec<_>>();
6082                        let ty = if column % 4 == 0 {
6083                            LogicalType::Varchar
6084                        } else {
6085                            LogicalType::BigInt
6086                        };
6087                        Vector::from_values(ty, &values).expect("a column")
6088                    })
6089                    .collect::<Vec<_>>();
6090                writer.append(&Chunk::new(columns).expect("forty columns")).expect("a part");
6091            }
6092            writer.finish().expect("commit");
6093        }
6094
6095        let first = path("repeatable-one");
6096        let second = path("repeatable-two");
6097        written(&first);
6098        written(&second);
6099        let left = fs::read(&first).expect("the first file");
6100        let right = fs::read(&second).expect("the second file");
6101        assert_eq!(left.len(), right.len(), "two writes of the same rows differ in length");
6102        assert!(left == right, "two writes of the same rows differ in their bytes");
6103
6104        // And the rows are still there, since a pair of identically wrong files would pass the
6105        // comparison above on its own.
6106        let reader = Reader::open(&first).expect("valid directory");
6107        assert_eq!(reader.table().rows(), 70 * 64);
6108        let read = reader.read(0, &[0, 1]).expect("the first part back");
6109        assert_eq!(read.value_at(0, 0), Value::Varchar("v0".to_owned()));
6110        assert_eq!(read.value_at(0, 1), Value::BigInt(0));
6111        fs::remove_file(first).expect("remove scratch file");
6112        fs::remove_file(second).expect("remove scratch file");
6113    }
6114}