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