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