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