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