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