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