1#![forbid(unsafe_code)]
29
30use std::cmp::Ordering;
31use std::collections::{HashMap, VecDeque};
32use std::fs::{File, OpenOptions};
33use std::io::{Read, Seek, SeekFrom};
34use std::mem::size_of;
35use std::path::Path;
36use std::sync::atomic::{AtomicUsize, Ordering as Atomic};
37use std::sync::{Arc, Mutex, OnceLock};
38
39use rudb_common::bounds::{Bound, Op};
40use rudb_common::{Error, Field, LogicalType, Result, Value};
41use rudb_encoding::{chooser, integer, string};
42use rudb_storage::sieve::Sieve;
43use rudb_storage::{Probe, Range, Zone};
44use rudb_vector::string::StringColumn;
45use rudb_vector::validity::Validity;
46use rudb_vector::{Buffer, Chunk, Data, TextSource, Vector};
47
48const MAGIC: &[u8; 8] = b"RUDBNV10";
49const DIRECTORY: &[u8; 8] = b"RUDBDI10";
50const FORMAT: u32 = 13;
51const HEADER: u64 = 80;
52const SLOT_BYTES: usize = 28;
53const MAX_PAGE: usize = 256 * 1024 * 1024;
54const MAX_DIRECTORY: usize = 128 * 1024 * 1024;
55const FREQUENCIES: &[u8; 8] = b"RUDBFQ2\0";
56const FREQUENCY_CANDIDATES: usize = 32_768;
57const FREQUENCY_ENTRIES: usize = 512;
58const FREQUENCY_BUILD_RANK: usize = 10;
59const FREQUENCY_ORDINALS: usize = 65_536;
60const MAX_FREQUENCY_WORKERS: usize = 16;
61
62const SIEVE_BUDGET: usize = 8 * 1024;
68
69fn io(error: std::io::Error) -> Error {
70 Error::io(error.to_string())
71}
72
73fn invalid(message: &str) -> Error {
74 Error::invalid_input(format!("invalid rudb native file: {message}"))
75}
76
77fn sum(counts: impl Iterator<Item = u64>) -> u64 {
79 counts.fold(0, u64::saturating_add)
80}
81
82fn span_bytes(spans: &[Span], at: usize) -> u64 {
84 spans.get(at).map_or(0, |span| u64::from(span.length))
85}
86
87fn page_bytes(pages: &[Option<Page>], at: usize) -> u64 {
89 pages.get(at).and_then(Option::as_ref).map_or(0, Page::bytes)
90}
91
92fn checksum(bytes: &[u8]) -> u64 {
93 const P1: u64 = 11_400_714_785_074_694_791;
94 const P2: u64 = 14_029_467_366_897_019_727;
95 const P3: u64 = 1_609_587_929_392_839_161;
96 const P4: u64 = 9_650_029_242_287_828_579;
97 const P5: u64 = 2_870_177_450_012_600_261;
98 let round = |state: u64, word: u64| {
99 state.wrapping_add(word.wrapping_mul(P2)).rotate_left(31).wrapping_mul(P1)
100 };
101 let merge = |state: u64, lane: u64| (state ^ round(0, lane)).wrapping_mul(P1).wrapping_add(P4);
102 let word =
103 |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("eight checksum bytes"));
104
105 let mut at = 0;
106 let mut hash = if bytes.len() >= 32 {
107 let mut one = P1.wrapping_add(P2);
108 let mut two = P2;
109 let mut three = 0;
110 let mut four = 0_u64.wrapping_sub(P1);
111 while at + 32 <= bytes.len() {
112 one = round(one, word(at));
113 two = round(two, word(at + 8));
114 three = round(three, word(at + 16));
115 four = round(four, word(at + 24));
116 at += 32;
117 }
118 let combined = one
119 .rotate_left(1)
120 .wrapping_add(two.rotate_left(7))
121 .wrapping_add(three.rotate_left(12))
122 .wrapping_add(four.rotate_left(18));
123 merge(merge(merge(merge(combined, one), two), three), four)
124 } else {
125 P5
126 };
127 hash = hash.wrapping_add(bytes.len() as u64);
128 while at + 8 <= bytes.len() {
129 hash ^= round(0, word(at));
130 hash = hash.rotate_left(27).wrapping_mul(P1).wrapping_add(P4);
131 at += 8;
132 }
133 if at + 4 <= bytes.len() {
134 let tail = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four checksum bytes"));
135 hash ^= u64::from(tail).wrapping_mul(P1);
136 hash = hash.rotate_left(23).wrapping_mul(P2).wrapping_add(P3);
137 at += 4;
138 }
139 while at < bytes.len() {
140 hash ^= u64::from(bytes[at]).wrapping_mul(P5);
141 hash = hash.rotate_left(11).wrapping_mul(P1);
142 at += 1;
143 }
144 hash ^= hash >> 33;
145 hash = hash.wrapping_mul(P2);
146 hash ^= hash >> 29;
147 hash = hash.wrapping_mul(P3);
148 hash ^ (hash >> 32)
149}
150
151#[derive(Debug, Clone, Copy)]
152struct Slot {
153 offset: u64,
154 length: u32,
155 generation: u64,
156 hash: u64,
157}
158
159impl Slot {
160 fn bytes(self) -> [u8; SLOT_BYTES] {
161 let mut result = [0; SLOT_BYTES];
162 result[..8].copy_from_slice(&self.offset.to_le_bytes());
163 result[8..12].copy_from_slice(&self.length.to_le_bytes());
164 result[12..20].copy_from_slice(&self.generation.to_le_bytes());
165 result[20..28].copy_from_slice(&self.hash.to_le_bytes());
166 result
167 }
168
169 fn read(bytes: &[u8]) -> Self {
170 Self {
171 offset: u64::from_le_bytes(bytes[..8].try_into().expect("eight bytes")),
172 length: u32::from_le_bytes(bytes[8..12].try_into().expect("four bytes")),
173 generation: u64::from_le_bytes(bytes[12..20].try_into().expect("eight bytes")),
174 hash: u64::from_le_bytes(bytes[20..28].try_into().expect("eight bytes")),
175 }
176 }
177}
178
179#[derive(Debug, Clone, Copy)]
180struct Page {
181 offset: u64,
182 length: u32,
183 hash: u64,
184}
185
186impl Page {
187 fn bytes(&self) -> u64 {
189 u64::from(self.length)
190 }
191}
192
193#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
194enum FrequencyValue {
195 Null,
196 Integer(i128),
197 Code(u32),
198}
199
200#[derive(Debug, Clone)]
201struct FrequencyEntry {
202 value: FrequencyValue,
203 count: u64,
204}
205
206#[derive(Debug, Clone)]
211struct FrequencySummary {
212 entries: Vec<FrequencyEntry>,
213 omitted_max: u64,
214 ordinals: Vec<u64>,
215}
216
217#[derive(Debug, Clone, PartialEq, Eq)]
219pub struct FrequencyOccurrences {
220 pub omitted_max: u64,
222 pub ordinals: Vec<u64>,
224}
225
226#[derive(Debug, Clone, Copy, Default)]
233struct Span {
234 offset: u64,
235 length: u32,
236}
237
238#[derive(Debug, Clone)]
240pub struct Stripe {
241 rows: usize,
242 parts: Vec<u32>,
245 index: Span,
249 pages: Vec<Span>,
250 memberships: Vec<Option<Page>>,
251 sieves: Vec<Option<Page>>,
254 zone: Zone,
255}
256
257impl Stripe {
258 #[must_use]
260 pub fn rows(&self) -> usize {
261 self.rows
262 }
263
264 #[must_use]
266 pub fn parts(&self) -> usize {
267 self.parts.len()
268 }
269}
270
271#[derive(Debug, Clone)]
273pub struct Table {
274 name: String,
275 fields: Vec<Field>,
276 stripes: Vec<Stripe>,
277 rows: usize,
278 dictionaries: Vec<Option<Page>>,
279 frequencies: Vec<Option<FrequencySummary>>,
280}
281
282impl Table {
283 #[must_use]
285 pub fn name(&self) -> &str {
286 &self.name
287 }
288
289 #[must_use]
291 pub fn fields(&self) -> &[Field] {
292 &self.fields
293 }
294
295 #[must_use]
297 pub fn rows(&self) -> usize {
298 self.rows
299 }
300
301 #[must_use]
303 pub fn stripes(&self) -> &[Stripe] {
304 &self.stripes
305 }
306}
307
308#[derive(Debug, Clone)]
310pub struct ColumnLayout {
311 pub name: String,
313 pub kind: String,
315 pub pages: u64,
317 pub memberships: u64,
319 pub sieves: u64,
321 pub dictionary: u64,
323}
324
325impl ColumnLayout {
326 #[must_use]
328 pub fn total(&self) -> u64 {
329 self.pages
330 .saturating_add(self.memberships)
331 .saturating_add(self.sieves)
332 .saturating_add(self.dictionary)
333 }
334}
335
336#[derive(Debug, Clone)]
347pub struct Layout {
348 pub file: u64,
350 pub rows: usize,
352 pub stripes: usize,
354 pub parts: usize,
356 pub columns: Vec<ColumnLayout>,
358 pub indexes: u64,
361 pub directory: u64,
363 pub header: u64,
365}
366
367impl Layout {
368 #[must_use]
370 pub fn columns_total(&self) -> u64 {
371 self.columns.iter().map(ColumnLayout::total).fold(0, u64::saturating_add)
372 }
373
374 #[must_use]
380 pub fn unaccounted(&self) -> u64 {
381 self.file
382 .saturating_sub(self.columns_total())
383 .saturating_sub(self.indexes)
384 .saturating_sub(self.directory)
385 .saturating_sub(self.header)
386 }
387}
388
389#[derive(Debug)]
391struct GlobalDictionary {
392 primary: HashMap<u64, u32>,
393 collisions: HashMap<u64, Vec<u32>>,
394 offsets: Vec<u32>,
395 payload: Vec<u8>,
396 counts: Vec<u64>,
397 nulls: u64,
398}
399
400impl GlobalDictionary {
401 fn new() -> Self {
402 Self {
403 primary: HashMap::new(),
404 collisions: HashMap::new(),
405 offsets: vec![0],
406 payload: Vec::new(),
407 counts: Vec::new(),
408 nulls: 0,
409 }
410 }
411
412 fn bytes(&self, code: u32) -> Option<&[u8]> {
413 let start = *self.offsets.get(code as usize)? as usize;
414 let end = *self.offsets.get(code as usize + 1)? as usize;
415 self.payload.get(start..end)
416 }
417
418 fn code(&mut self, text: &str) -> Result<u32> {
419 let hash = checksum(text.as_bytes());
420 if let Some(&code) = self.primary.get(&hash) {
421 if self.bytes(code) == Some(text.as_bytes()) {
422 return Ok(code);
423 }
424 if let Some(codes) = self.collisions.get(&hash) {
425 if let Some(code) =
426 codes.iter().copied().find(|&code| self.bytes(code) == Some(text.as_bytes()))
427 {
428 return Ok(code);
429 }
430 }
431 let code = self.insert(text)?;
432 self.collisions.entry(hash).or_default().push(code);
433 return Ok(code);
434 }
435 let code = self.insert(text)?;
436 self.primary.insert(hash, code);
437 Ok(code)
438 }
439
440 fn insert(&mut self, text: &str) -> Result<u32> {
441 let code = u32::try_from(self.offsets.len() - 1)
442 .map_err(|_| invalid("global dictionary has too many values"))?;
443 self.payload.extend_from_slice(text.as_bytes());
444 self.offsets.push(
445 u32::try_from(self.payload.len())
446 .map_err(|_| invalid("global dictionary payload exceeds 4 GiB"))?,
447 );
448 self.counts.push(0);
449 Ok(code)
450 }
451
452 fn ranked(&self) -> Vec<(u64, u32)> {
472 let count = self.offsets.len() - 1;
473 let mut ranked = (0..count)
474 .map(|code| {
475 let code = code as u32;
476 (head(self.bytes(code).unwrap_or_default()), code)
477 })
478 .collect::<Vec<_>>();
479 ranked.sort_unstable_by(|left, right| {
480 left.0.cmp(&right.0).then_with(|| self.bytes(left.1).cmp(&self.bytes(right.1)))
481 });
482 ranked
483 }
484
485 fn observe(&mut self, code: u32, null: bool) -> Result<()> {
486 if null {
487 self.nulls = self.nulls.saturating_add(1);
488 return Ok(());
489 }
490 let count = self
491 .counts
492 .get_mut(code as usize)
493 .ok_or_else(|| invalid("global dictionary count code is out of range"))?;
494 *count = count.saturating_add(1);
495 Ok(())
496 }
497}
498
499#[derive(Debug)]
501pub struct Writer {
502 file: File,
503 at: u64,
511 table: Table,
512 generation: u64,
513 order: Vec<((u64, u64), (u64, u64))>,
516 next_order: u64,
517 dictionaries: Vec<Option<GlobalDictionary>>,
518 pending: Vec<PendingPart>,
519}
520
521#[derive(Debug)]
522struct PendingPart {
523 order: (u64, u64),
524 rows: usize,
525 pages: Vec<Vec<u8>>,
526 codes: Vec<Option<Vec<u32>>>,
527 zone: Zone,
528 sieves: Vec<Option<Sieve>>,
529}
530
531const STRIPE_PARTS: usize = 64;
538
539const INDEX_ENTRY: usize = size_of::<u32>() + size_of::<u64>();
541
542fn index_section(parts: usize) -> Result<usize> {
544 parts
545 .checked_mul(INDEX_ENTRY)
546 .and_then(|bytes| bytes.checked_add(size_of::<u64>()))
547 .ok_or_else(|| invalid("index page length overflow"))
548}
549
550impl Writer {
551 pub fn create(
557 path: impl AsRef<Path>,
558 name: impl Into<String>,
559 fields: Vec<Field>,
560 ) -> Result<Self> {
561 for field in &fields {
562 type_tag(&field.ty)?;
563 }
564 let file =
565 OpenOptions::new().write(true).read(true).create_new(true).open(path).map_err(io)?;
566 let mut header = [0; HEADER as usize];
567 header[..8].copy_from_slice(MAGIC);
568 header[8..12].copy_from_slice(&FORMAT.to_le_bytes());
569 write_at(&file, 0, &header)?;
570 Ok(Self {
571 file,
572 at: HEADER,
573 dictionaries: fields
574 .iter()
575 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
576 .collect(),
577 table: Table {
578 name: name.into(),
579 dictionaries: vec![None; fields.len()],
580 fields,
581 stripes: Vec::new(),
582 rows: 0,
583 frequencies: Vec::new(),
584 },
585 generation: 1,
586 order: Vec::new(),
587 next_order: 0,
588 pending: Vec::with_capacity(STRIPE_PARTS),
589 })
590 }
591
592 fn put(&mut self, bytes: &[u8]) -> Result<()> {
597 write_at(&self.file, self.at, bytes)?;
598 self.at = self
599 .at
600 .checked_add(bytes.len() as u64)
601 .ok_or_else(|| invalid("native file length overflow"))?;
602 Ok(())
603 }
604
605 pub fn append(&mut self, chunk: &Chunk) -> Result<()> {
611 let order = (self.next_order, 0);
612 self.next_order = self.next_order.saturating_add(1);
613 self.append_at(order, chunk)
614 }
615
616 pub fn append_at(&mut self, order: (u64, u64), chunk: &Chunk) -> Result<()> {
627 if chunk.is_empty() {
628 return Ok(());
629 }
630 if chunk.width() != self.table.fields.len() {
631 return Err(invalid("chunk width differs from table schema"));
632 }
633 let mut pages = Vec::with_capacity(chunk.width());
634 let mut codes = Vec::with_capacity(chunk.width());
635 for (index, field) in self.table.fields.iter().enumerate() {
636 let column = chunk.column(index)?;
637 if column.logical_type() != &field.ty {
638 return Err(invalid("chunk type differs from table schema"));
639 }
640 let (bytes, unique) = encode(column, self.dictionaries[index].as_mut())?;
641 if bytes.len() > MAX_PAGE {
642 return Err(invalid("column page exceeds the configured bound"));
643 }
644 pages.push(bytes);
645 codes.push(unique);
646 }
647 self.table.rows = self
648 .table
649 .rows
650 .checked_add(chunk.len())
651 .ok_or_else(|| invalid("row count overflow"))?;
652 if self.pending.last().is_some_and(|last| last.order > order) {
653 self.flush_pending()?;
654 }
655 let zone = Zone::of(chunk);
658 let mut sieves = Vec::with_capacity(chunk.width());
659 for (index, column) in chunk.columns().iter().enumerate() {
660 let range =
661 zone.column(index).ok_or_else(|| invalid("a zone is narrower than its chunk"))?;
662 if self.dictionaries[index].is_some() {
667 sieves.push(None);
668 continue;
669 }
670 sieves.push(Sieve::of(column, range, SIEVE_BUDGET));
671 }
672 self.pending.push(PendingPart { order, rows: chunk.len(), pages, codes, zone, sieves });
673 if self.pending.len() == STRIPE_PARTS {
674 self.flush_pending()?;
675 }
676 Ok(())
677 }
678
679 fn flush_pending(&mut self) -> Result<()> {
681 if self.pending.is_empty() {
682 return Ok(());
683 }
684 let width = self.table.fields.len();
685 let mut held = std::mem::take(&mut self.pending);
688 let parts = held.len();
689 let mut pages = Vec::with_capacity(width);
690 let mut memberships = vec![None; width];
691 let mut ranges = Vec::with_capacity(width);
692 let mut index = Vec::with_capacity(width.saturating_mul(index_section(parts)?));
693 for column in 0..width {
694 let offset = self.at;
695 let section = index.len();
696 let mut length = 0_usize;
697 for pending in &held {
698 let bytes = &pending.pages[column];
699 write_at(&self.file, self.at + length as u64, bytes)?;
700 put_u32(
701 &mut index,
702 u32::try_from(bytes.len()).map_err(|_| invalid("part length overflow"))?,
703 );
704 put_u64(&mut index, checksum(bytes));
705 length = length
706 .checked_add(bytes.len())
707 .ok_or_else(|| invalid("column page length overflow"))?;
708 }
709 let hash = checksum(&index[section..]);
710 put_u64(&mut index, hash);
711 if length > MAX_PAGE {
712 return Err(invalid("column page exceeds the configured bound"));
713 }
714 self.at = self
715 .at
716 .checked_add(length as u64)
717 .ok_or_else(|| invalid("native file length overflow"))?;
718 pages.push(Span {
719 offset,
720 length: u32::try_from(length).map_err(|_| invalid("page length overflow"))?,
721 });
722 ranges.push(merged_range(
723 held.iter().map(|pending| pending.zone.column(column).cloned().unwrap_or_default()),
724 ));
725 }
726 for (column, membership) in memberships.iter_mut().enumerate() {
727 if held.iter().all(|pending| pending.codes[column].is_none()) {
728 continue;
729 }
730 let lists = held
731 .iter()
732 .map(|pending| pending.codes[column].clone().unwrap_or_default())
733 .collect::<Vec<_>>();
734 let bytes = encode_membership(&merged_codes(lists));
735 let offset = self.at;
736 self.put(&bytes)?;
737 *membership = Some(Page {
738 offset,
739 length: u32::try_from(bytes.len())
740 .map_err(|_| invalid("membership page length overflow"))?,
741 hash: checksum(&bytes),
742 });
743 }
744 let mut sieves = vec![None; width];
745 for (column, page) in sieves.iter_mut().enumerate() {
746 if held.iter().all(|pending| pending.sieves[column].is_none()) {
747 continue;
748 }
749 let bytes = encode_sieves(held.iter().map(|pending| &pending.sieves[column]))?;
750 let offset = self.at;
751 self.put(&bytes)?;
752 *page = Some(Page {
753 offset,
754 length: u32::try_from(bytes.len())
755 .map_err(|_| invalid("sieve page length overflow"))?,
756 hash: checksum(&bytes),
757 });
758 }
759 let offset = self.at;
760 self.put(&index)?;
761 let index = Span {
762 offset,
763 length: u32::try_from(index.len())
764 .map_err(|_| invalid("index page length overflow"))?,
765 };
766 let mut rows = 0_usize;
767 let mut lengths = Vec::with_capacity(parts);
768 let mut span = None;
769 for pending in held.drain(..) {
770 rows = rows.checked_add(pending.rows).ok_or_else(|| invalid("row count overflow"))?;
771 lengths
772 .push(u32::try_from(pending.rows).map_err(|_| invalid("part row count overflow"))?);
773 span = Some(
774 span.map_or((pending.order, pending.order), |(first, _)| (first, pending.order)),
775 );
776 }
777 self.order.push(span.ok_or_else(|| invalid("a stripe was flushed with no parts"))?);
778 self.table.stripes.push(Stripe {
779 rows,
780 parts: lengths,
781 index,
782 pages,
783 memberships,
784 sieves,
785 zone: Zone::from_ranges(ranges),
786 });
787 self.pending = held;
789 Ok(())
790 }
791
792 fn numeric_frequency(&self, column: usize) -> Result<Option<FrequencySummary>> {
796 let ty = &self.table.fields[column].ty;
797 if !matches!(
798 ty,
799 LogicalType::TinyInt
800 | LogicalType::SmallInt
801 | LogicalType::Integer
802 | LogicalType::BigInt
803 | LogicalType::UTinyInt
804 | LogicalType::USmallInt
805 | LogicalType::UInteger
806 | LogicalType::UBigInt
807 | LogicalType::Date
808 | LogicalType::Timestamp
809 ) {
810 return Ok(None);
811 }
812 let mut candidates: HashMap<FrequencyValue, u32> = HashMap::new();
813 let mut decrements = 0_u64;
814 self.visit_numeric(column, |_, value| {
815 if let Some(count) = candidates.get_mut(&value) {
816 *count = count.saturating_add(1);
817 } else if candidates.len() < FREQUENCY_CANDIDATES {
818 candidates.insert(value, 1);
819 } else {
820 candidates.retain(|_, count| {
821 *count -= 1;
822 *count != 0
823 });
824 decrements = decrements.saturating_add(1);
825 }
826 })?;
827 let (exact, ordinals) = if decrements == 0 {
828 (
829 candidates
830 .into_iter()
831 .map(|(value, count)| (value, u64::from(count)))
832 .collect::<HashMap<_, _>>(),
833 Vec::new(),
834 )
835 } else {
836 let mut lower = candidates.values().copied().collect::<Vec<_>>();
837 lower.sort_unstable_by(|left, right| right.cmp(left));
838 if lower.len() < FREQUENCY_BUILD_RANK
839 || u64::from(lower[FREQUENCY_BUILD_RANK - 1]) <= decrements
840 {
841 return Ok(None);
842 }
843 let mut exact =
844 candidates.into_keys().map(|value| (value, 0_u64)).collect::<HashMap<_, _>>();
845 let mut ordinals = Vec::new();
846 let mut exceeded = false;
847 self.visit_numeric(column, |ordinal, value| {
848 if let Some(count) = exact.get_mut(&value) {
849 *count = count.saturating_add(1);
850 if !exceeded {
851 if ordinals.len() < FREQUENCY_ORDINALS {
852 ordinals.push(ordinal);
853 } else {
854 ordinals.clear();
855 exceeded = true;
856 }
857 }
858 }
859 })?;
860 (exact, ordinals)
861 };
862 let mut entries = exact
863 .into_iter()
864 .map(|(value, count)| FrequencyEntry { value, count })
865 .collect::<Vec<_>>();
866 entries.sort_unstable_by(|left, right| {
867 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
868 });
869 let omitted_max =
870 entries.get(FREQUENCY_ENTRIES).map_or(decrements, |entry| decrements.max(entry.count));
871 entries.truncate(FREQUENCY_ENTRIES);
872 Ok(Some(FrequencySummary { entries, omitted_max, ordinals }))
873 }
874
875 fn visit_numeric(
876 &self,
877 column: usize,
878 mut visit: impl FnMut(u64, FrequencyValue),
879 ) -> Result<()> {
880 let ty = &self.table.fields[column].ty;
881 let mut start = 0_u64;
882 for stripe in &self.table.stripes {
883 let spans = read_index(&self.file, stripe, column)?;
884 let page = stripe.pages[column];
885 let mut bytes = vec![0; page.length as usize];
886 read_at(&self.file, page.offset, &mut bytes)?;
887 for (span, &rows) in spans.iter().zip(&stripe.parts) {
888 let part = part_bytes(&bytes, *span)?;
889 if checksum(part) != span.hash {
890 return Err(invalid("column page checksum differs while building frequencies"));
891 }
892 let rows = rows as usize;
893 let vector = decode(ty, rows, part, None)?;
894 for row in 0..rows {
896 let value = if vector.is_null_at(row) {
897 FrequencyValue::Null
898 } else {
899 let widened = match vector.signed_at(row) {
903 Some(value) => Some(value),
904 None => match vector.value_at(row) {
905 Value::UTinyInt(value) => Some(i128::from(value)),
906 Value::USmallInt(value) => Some(i128::from(value)),
907 Value::UInteger(value) => Some(i128::from(value)),
908 Value::UBigInt(value) => Some(i128::from(value)),
909 _ => None,
910 },
911 };
912 FrequencyValue::Integer(widened.ok_or_else(|| {
913 invalid("numeric frequency page did not contain an integer value")
914 })?)
915 };
916 visit(start.saturating_add(row as u64), value);
917 }
918 start = start.saturating_add(rows as u64);
919 }
920 }
921 Ok(())
922 }
923
924 fn numeric_frequencies(&self) -> Result<Vec<Option<FrequencySummary>>> {
926 let columns = self
927 .table
928 .fields
929 .iter()
930 .enumerate()
931 .filter_map(|(column, field)| {
932 matches!(
933 field.ty,
934 LogicalType::TinyInt
935 | LogicalType::SmallInt
936 | LogicalType::Integer
937 | LogicalType::BigInt
938 | LogicalType::UTinyInt
939 | LogicalType::USmallInt
940 | LogicalType::UInteger
941 | LogicalType::UBigInt
942 | LogicalType::Date
943 | LogicalType::Timestamp
944 )
945 .then_some(column)
946 })
947 .collect::<Vec<_>>();
948 let workers = std::thread::available_parallelism()
949 .map_or(1, usize::from)
950 .min(MAX_FREQUENCY_WORKERS)
951 .min(columns.len());
952 if workers <= 1 {
953 let mut frequencies = vec![None; self.table.fields.len()];
954 for column in columns {
955 frequencies[column] = self.numeric_frequency(column)?;
956 }
957 return Ok(frequencies);
958 }
959 let width = columns.len().div_ceil(workers);
960 let pieces = std::thread::scope(|scope| {
961 columns
962 .chunks(width)
963 .map(|columns| {
964 scope.spawn(|| {
965 columns
966 .iter()
967 .map(|&column| Ok((column, self.numeric_frequency(column)?)))
968 .collect::<Result<Vec<_>>>()
969 })
970 })
971 .collect::<Vec<_>>()
972 .into_iter()
973 .map(|handle| {
974 handle
975 .join()
976 .map_err(|_| Error::internal("a native frequency worker panicked"))?
977 })
978 .collect::<Result<Vec<_>>>()
979 })?;
980 let mut frequencies = vec![None; self.table.fields.len()];
981 for piece in pieces {
982 for (column, summary) in piece {
983 frequencies[column] = summary;
984 }
985 }
986 Ok(frequencies)
987 }
988
989 pub fn finish(mut self) -> Result<Table> {
995 self.flush_pending()?;
996 let mut stripes = std::mem::take(&mut self.order)
997 .into_iter()
998 .zip(std::mem::take(&mut self.table.stripes))
999 .collect::<Vec<_>>();
1000 stripes.sort_by_key(|(order, _)| order.0);
1001 let mut previous: Option<(u64, u64)> = None;
1002 for ((first, last), _) in &stripes {
1003 if previous.is_some_and(|previous| previous >= *first) {
1004 return Err(invalid("chunks did not arrive in source order"));
1005 }
1006 previous = Some(*last);
1007 }
1008 self.table.stripes = stripes.into_iter().map(|(_, stripe)| stripe).collect();
1009 self.table.frequencies = self.numeric_frequencies()?;
1010 let dictionaries = std::mem::take(&mut self.dictionaries);
1011 let orders = rankings(&dictionaries)?;
1012 for (index, (dictionary, order)) in dictionaries.into_iter().zip(orders).enumerate() {
1013 let Some(dictionary) = dictionary else { continue };
1014 self.table.frequencies[index] = Some(code_frequency(&dictionary));
1015 let encoded = encode_global_dictionary(dictionary, &order)?;
1016 let offset = self.at;
1017 self.put(&encoded.index)?;
1018 self.put(&encoded.ranks)?;
1019 self.put(&encoded.payload)?;
1020 let length = encoded
1021 .index
1022 .len()
1023 .checked_add(encoded.ranks.len())
1024 .and_then(|len| len.checked_add(encoded.payload.len()))
1025 .ok_or_else(|| invalid("dictionary page length overflow"))?;
1026 self.table.dictionaries[index] = Some(Page {
1027 offset,
1028 length: u32::try_from(length)
1029 .map_err(|_| invalid("dictionary page length overflow"))?,
1030 hash: checksum(&encoded.index),
1031 });
1032 }
1033 let directory = encode_directory(&self.table)?;
1034 if directory.len() > MAX_DIRECTORY {
1035 return Err(invalid("directory exceeds the configured bound"));
1036 }
1037 let offset = self.at;
1038 self.put(&directory)?;
1039 self.file.sync_all().map_err(io)?;
1040 let slot = Slot {
1041 offset,
1042 length: u32::try_from(directory.len())
1043 .map_err(|_| invalid("directory length overflow"))?,
1044 generation: self.generation,
1045 hash: checksum(&directory),
1046 };
1047 write_at(&self.file, 16, &slot.bytes())?;
1050 self.file.sync_all().map_err(io)?;
1051 Ok(self.table)
1052 }
1053}
1054
1055#[derive(Debug, Clone)]
1057pub struct Reader {
1058 file: Arc<File>,
1059 table: Arc<Table>,
1060 dictionaries: Arc<Vec<OnceLock<Arc<Vector>>>>,
1061 sieves: Arc<Vec<Vec<SieveSlot>>>,
1065 places: Arc<Vec<Place>>,
1067 cache: Arc<Vec<Mutex<Cached>>>,
1068 pages: Arc<AtomicUsize>,
1071 indexes: Arc<AtomicUsize>,
1074 kept: Arc<AtomicUsize>,
1077 size: u64,
1079 directory: u64,
1081}
1082
1083#[derive(Debug, Clone, Copy)]
1085struct Place {
1086 stripe: u32,
1087 part: u32,
1088 rows: u32,
1089}
1090
1091#[derive(Debug, Clone, Copy)]
1093struct PartSpan {
1094 start: usize,
1095 length: usize,
1096 hash: u64,
1097}
1098
1099#[derive(Debug, Clone)]
1105struct CachedColumn {
1106 stripe: usize,
1107 index: Arc<Vec<PartSpan>>,
1108 page: Option<Arc<Vec<u8>>>,
1109}
1110
1111#[derive(Debug, Default)]
1131struct Cached {
1132 pages: Vec<Option<Arc<Vec<u8>>>>,
1133 order: VecDeque<usize>,
1134 loading: Vec<usize>,
1135 index: Vec<Option<Arc<Vec<PartSpan>>>>,
1136}
1137
1138const CACHED_STRIPES_PER_COLUMN: usize = 4;
1150
1151type SieveSlot = OnceLock<Arc<Vec<Option<Sieve>>>>;
1153
1154type CrossingCache = OnceLock<Box<[OnceLock<Result<Vec<u8>>>]>>;
1155
1156#[derive(Debug)]
1157struct NativeText {
1158 file: Arc<File>,
1159 offsets: Vec<u32>,
1160 ranks: usize,
1162 rank_at: u64,
1166 rank_hashes: Vec<u64>,
1167 rank_blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1168 payload: u64,
1169 payload_len: usize,
1170 hashes: Vec<u64>,
1171 payload_extents: Vec<OnceLock<Result<Vec<u8>>>>,
1173 crossing: Vec<CrossingCache>,
1174}
1175
1176const TEXT_PAYLOAD_BLOCK: usize = 64 * 1024;
1177
1178const TEXT_PAYLOAD_EXTENT: usize = 8;
1199const TEXT_CROSSING_BLOCK: usize = 1024;
1200
1201const TEXT_RANK_BLOCK: usize = 512;
1211
1212const RANK_ENTRY: usize = size_of::<u64>() + size_of::<u32>();
1214
1215impl NativeText {
1216 fn payload_block(&self, block: usize) -> Result<Option<&[u8]>> {
1217 if block >= self.hashes.len() {
1218 return Ok(None);
1219 }
1220 let extent = block / TEXT_PAYLOAD_EXTENT;
1221 let Some(slot) = self.payload_extents.get(extent) else { return Ok(None) };
1222 let bytes = slot
1223 .get_or_init(|| {
1224 let start = extent
1225 .checked_mul(TEXT_PAYLOAD_EXTENT * TEXT_PAYLOAD_BLOCK)
1226 .ok_or_else(|| invalid("global dictionary block offset overflow"))?;
1227 let len = (TEXT_PAYLOAD_EXTENT * TEXT_PAYLOAD_BLOCK).min(
1228 self.payload_len
1229 .checked_sub(start)
1230 .ok_or_else(|| invalid("global dictionary block starts past payload"))?,
1231 );
1232 let mut bytes = vec![0; len];
1233 read_at(&self.file, self.payload + start as u64, &mut bytes)?;
1234 for (within, piece) in bytes.chunks(TEXT_PAYLOAD_BLOCK).enumerate() {
1237 if checksum(piece)
1238 != *self
1239 .hashes
1240 .get(extent * TEXT_PAYLOAD_EXTENT + within)
1241 .ok_or_else(|| invalid("global dictionary block has no checksum"))?
1242 {
1243 return Err(invalid("global dictionary payload checksum differs"));
1244 }
1245 }
1246 Ok(bytes)
1247 })
1248 .as_ref()
1249 .map_err(Clone::clone)?;
1250 let within = (block % TEXT_PAYLOAD_EXTENT) * TEXT_PAYLOAD_BLOCK;
1251 let end = (within + TEXT_PAYLOAD_BLOCK).min(bytes.len());
1252 Ok(bytes.get(within..end))
1253 }
1254
1255 fn rank_parts(&self, rank: usize) -> Result<(&[u8], usize)> {
1262 let slot = self
1263 .rank_blocks
1264 .get(rank / TEXT_RANK_BLOCK)
1265 .ok_or_else(|| invalid("global dictionary rank is past the order"))?;
1266 let block = slot
1267 .get_or_init(|| {
1268 let first = rank / TEXT_RANK_BLOCK * TEXT_RANK_BLOCK;
1269 let len = TEXT_RANK_BLOCK.min(self.ranks - first) * RANK_ENTRY;
1270 let mut bytes = vec![0; len];
1271 read_at(&self.file, self.rank_at + (first * RANK_ENTRY) as u64, &mut bytes)?;
1272 if checksum(&bytes)
1273 != *self
1274 .rank_hashes
1275 .get(rank / TEXT_RANK_BLOCK)
1276 .ok_or_else(|| invalid("global dictionary rank block has no checksum"))?
1277 {
1278 return Err(invalid("global dictionary rank checksum differs"));
1279 }
1280 Ok(bytes)
1281 })
1282 .as_ref()
1283 .map_err(Clone::clone)?;
1284 Ok((block.as_slice(), rank % TEXT_RANK_BLOCK))
1285 }
1286
1287 fn head_at(&self, rank: usize) -> Result<u64> {
1289 let (block, within) = self.rank_parts(rank)?;
1290 let at = within * size_of::<u64>();
1291 let bytes = block
1292 .get(at..at + size_of::<u64>())
1293 .ok_or_else(|| invalid("global dictionary rank block is short of heads"))?;
1294 Ok(u64::from_le_bytes(bytes.try_into().expect("eight bytes")))
1295 }
1296}
1297
1298impl TextSource for NativeText {
1299 fn len(&self) -> usize {
1300 self.offsets.len().saturating_sub(1)
1301 }
1302
1303 fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
1304 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1305 else {
1306 return Ok(None);
1307 };
1308 if start == end {
1309 return Ok(Some(&[]));
1310 }
1311 let first = start as usize / TEXT_PAYLOAD_BLOCK;
1312 let last = (end as usize - 1) / TEXT_PAYLOAD_BLOCK;
1313 if first == last {
1314 let Some(block) = self.payload_block(first)? else { return Ok(None) };
1315 let within = start as usize % TEXT_PAYLOAD_BLOCK;
1316 return Ok(block.get(within..within + (end - start) as usize));
1317 }
1318 let Some(crossing) = self.crossing.get(index / TEXT_CROSSING_BLOCK) else {
1319 return Ok(None);
1320 };
1321 let block = crossing.get_or_init(|| {
1322 (0..TEXT_CROSSING_BLOCK).map(|_| OnceLock::new()).collect::<Vec<_>>().into_boxed_slice()
1323 });
1324 block[index % TEXT_CROSSING_BLOCK]
1325 .get_or_init(|| {
1326 let mut bytes = Vec::with_capacity((end - start) as usize);
1327 for part in first..=last {
1328 let source = self
1329 .payload_block(part)?
1330 .ok_or_else(|| invalid("global dictionary block is missing"))?;
1331 let from = if part == first { start as usize % TEXT_PAYLOAD_BLOCK } else { 0 };
1332 let to = if part == last {
1333 (end as usize - 1) % TEXT_PAYLOAD_BLOCK + 1
1334 } else {
1335 source.len()
1336 };
1337 bytes.extend_from_slice(source.get(from..to).ok_or_else(|| {
1338 invalid("global dictionary value exceeds its payload block")
1339 })?);
1340 }
1341 Ok(bytes)
1342 })
1343 .as_ref()
1344 .map(|bytes| Some(bytes.as_slice()))
1345 .map_err(Clone::clone)
1346 }
1347
1348 fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
1349 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1350 else {
1351 return Ok(None);
1352 };
1353 Ok(Some((end - start) as usize))
1354 }
1355
1356 fn ranks(&self) -> Option<usize> {
1357 (self.ranks > 0).then_some(self.ranks)
1358 }
1359
1360 fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
1361 let settled = self.head_at(rank)?.cmp(&head(wanted));
1365 if settled != Ordering::Equal {
1366 return Ok(settled);
1367 }
1368 let code = self.code_at_rank(rank)?;
1369 let bytes = self
1370 .bytes_at(code as usize)?
1371 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
1372 Ok(bytes.cmp(wanted))
1373 }
1374
1375 fn code_at_rank(&self, rank: usize) -> Result<u32> {
1376 let (block, within) = self.rank_parts(rank)?;
1377 let heads = block.len() / RANK_ENTRY * size_of::<u64>();
1378 let at = heads + within * size_of::<u32>();
1379 let bytes = block
1380 .get(at..at + size_of::<u32>())
1381 .ok_or_else(|| invalid("global dictionary rank block is short of codes"))?;
1382 let code = u32::from_le_bytes(bytes.try_into().expect("four bytes"));
1383 if code as usize >= self.len() {
1384 return Err(invalid("global dictionary order names a code it does not have"));
1385 }
1386 Ok(code)
1387 }
1388
1389 fn footprint(&self) -> usize {
1390 self.offsets.capacity() * size_of::<u32>()
1391 + self.rank_hashes.capacity() * size_of::<u64>()
1392 + self.rank_blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1393 + self
1394 .rank_blocks
1395 .iter()
1396 .filter_map(OnceLock::get)
1397 .filter_map(|result| result.as_ref().ok())
1398 .map(Vec::capacity)
1399 .sum::<usize>()
1400 + self.payload_extents.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1401 + self.hashes.capacity() * size_of::<u64>()
1402 + self
1403 .payload_extents
1404 .iter()
1405 .filter_map(OnceLock::get)
1406 .filter_map(|result| result.as_ref().ok())
1407 .map(Vec::capacity)
1408 .sum::<usize>()
1409 + self.crossing.capacity() * size_of::<CrossingCache>()
1410 + self
1411 .crossing
1412 .iter()
1413 .filter_map(OnceLock::get)
1414 .map(|block| {
1415 block.len() * size_of::<OnceLock<Result<Vec<u8>>>>()
1416 + block
1417 .iter()
1418 .filter_map(OnceLock::get)
1419 .filter_map(|result| result.as_ref().ok())
1420 .map(Vec::capacity)
1421 .sum::<usize>()
1422 })
1423 .sum::<usize>()
1424 }
1425}
1426
1427fn places(table: &Table) -> Result<Vec<Place>> {
1429 let mut places = Vec::with_capacity(table.stripes.len().saturating_mul(STRIPE_PARTS));
1430 for (at, stripe) in table.stripes.iter().enumerate() {
1431 let index = u32::try_from(at).map_err(|_| invalid("too many stripes"))?;
1432 for (part, &rows) in stripe.parts.iter().enumerate() {
1433 places.push(Place {
1434 stripe: index,
1435 part: u32::try_from(part).map_err(|_| invalid("too many parts in a stripe"))?,
1436 rows,
1437 });
1438 }
1439 }
1440 Ok(places)
1441}
1442
1443fn read_index(file: &File, stripe: &Stripe, column: usize) -> Result<Vec<PartSpan>> {
1448 let parts = stripe.parts.len();
1449 let section = index_section(parts)?;
1450 let at = column.checked_mul(section).ok_or_else(|| invalid("index page offset overflow"))?;
1451 let end = at.checked_add(section).ok_or_else(|| invalid("index page offset overflow"))?;
1452 if end > stripe.index.length as usize {
1453 return Err(invalid("index page is shorter than its columns"));
1454 }
1455 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1456 let mut bytes = vec![0; section];
1457 let offset = stripe
1458 .index
1459 .offset
1460 .checked_add(at as u64)
1461 .ok_or_else(|| invalid("index page offset overflow"))?;
1462 read_at(file, offset, &mut bytes)?;
1463 let entries = section - size_of::<u64>();
1464 let stored = u64::from_le_bytes(bytes[entries..].try_into().expect("eight bytes"));
1465 if checksum(&bytes[..entries]) != stored {
1466 return Err(invalid(&format!(
1469 "index page section checksum differs, column {column} of {parts} parts at {offset}, \
1470 wanted {stored:016x} and got {:016x}",
1471 checksum(&bytes[..entries]),
1472 )));
1473 }
1474 let mut spans = Vec::with_capacity(parts);
1475 let mut start = 0_usize;
1476 for part in 0..parts {
1477 let at = part * INDEX_ENTRY;
1478 let length = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four bytes")) as usize;
1479 let hash = u64::from_le_bytes(bytes[at + 4..at + 12].try_into().expect("eight bytes"));
1480 spans.push(PartSpan { start, length, hash });
1481 start = start.checked_add(length).ok_or_else(|| invalid("column page length overflow"))?;
1482 }
1483 if start != page.length as usize {
1484 return Err(invalid("column page length differs from its index"));
1485 }
1486 Ok(spans)
1487}
1488
1489fn part_bytes(page: &[u8], span: PartSpan) -> Result<&[u8]> {
1491 let end = span.start.checked_add(span.length).ok_or_else(|| invalid("part range overflow"))?;
1492 page.get(span.start..end).ok_or_else(|| invalid("part exceeds its column page"))
1493}
1494
1495fn remember(cached: &mut Cached, held: &CachedColumn, kept: usize) {
1500 if let Some(slot) = cached.index.get_mut(held.stripe) {
1501 if slot.is_none() {
1502 *slot = Some(Arc::clone(&held.index));
1503 }
1504 }
1505 let Some(page) = held.page.clone() else { return };
1506 let Some(slot) = cached.pages.get_mut(held.stripe) else { return };
1507 if slot.is_none() {
1508 cached.order.push_back(held.stripe);
1509 }
1510 *slot = Some(page);
1511 while cached.order.len() > kept.max(1) {
1512 let Some(oldest) = cached.order.pop_front() else { break };
1513 if let Some(slot) = cached.pages.get_mut(oldest) {
1514 *slot = None;
1515 }
1516 }
1517}
1518
1519impl Reader {
1520 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1526 let mut file = File::open(path).map_err(io)?;
1527 let size = file.metadata().map_err(io)?.len();
1528 if size < HEADER {
1529 return Err(invalid("file is shorter than its header"));
1530 }
1531 let mut header = [0; HEADER as usize];
1532 file.read_exact(&mut header).map_err(io)?;
1533 let version = u32::from_le_bytes([header[8], header[9], header[10], header[11]]);
1534 if &header[..8] != MAGIC {
1539 return Err(invalid("the header does not begin with a rudb native magic"));
1540 }
1541 if version != FORMAT {
1542 return Err(invalid(&format!(
1543 "the file is format {version} and this build reads format {FORMAT}, so it has to \
1544 be written again"
1545 )));
1546 }
1547 let mut selected = None;
1548 for start in [16, 16 + SLOT_BYTES] {
1549 let slot = Slot::read(&header[start..start + SLOT_BYTES]);
1550 if slot.generation == 0 || slot.length == 0 || slot.length as usize > MAX_DIRECTORY {
1551 continue;
1552 }
1553 let Some(end) = slot.offset.checked_add(u64::from(slot.length)) else { continue };
1554 if slot.offset < HEADER || end > size {
1555 continue;
1556 }
1557 let mut bytes = vec![0; slot.length as usize];
1558 file.seek(SeekFrom::Start(slot.offset)).map_err(io)?;
1559 file.read_exact(&mut bytes).map_err(io)?;
1560 if checksum(&bytes) == slot.hash
1561 && selected
1562 .as_ref()
1563 .is_none_or(|(old, _): &(Slot, Vec<u8>)| old.generation < slot.generation)
1564 {
1565 selected = Some((slot, bytes));
1566 }
1567 }
1568 let (slot, bytes) =
1569 selected.ok_or_else(|| invalid("no committed directory slot is valid"))?;
1570 let table = decode_directory(&bytes, size)?;
1571 let places = places(&table)?;
1572 let dictionaries = (0..table.fields.len()).map(|_| OnceLock::new()).collect();
1573 let stripes = table.stripes.len();
1574 let cache = (0..table.fields.len())
1575 .map(|_| {
1576 Mutex::new(Cached {
1577 pages: (0..stripes).map(|_| None).collect(),
1578 index: (0..stripes).map(|_| None).collect(),
1579 ..Cached::default()
1580 })
1581 })
1582 .collect::<Vec<_>>();
1583 let sieves = (0..table.fields.len())
1584 .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
1585 .collect();
1586 Ok(Self {
1587 file: Arc::new(file),
1588 table: Arc::new(table),
1589 dictionaries: Arc::new(dictionaries),
1590 sieves: Arc::new(sieves),
1591 places: Arc::new(places),
1592 cache: Arc::new(cache),
1593 pages: Arc::new(AtomicUsize::new(0)),
1594 indexes: Arc::new(AtomicUsize::new(0)),
1595 kept: Arc::new(AtomicUsize::new(CACHED_STRIPES_PER_COLUMN)),
1596 size,
1597 directory: u64::from(slot.length),
1598 })
1599 }
1600
1601 #[must_use]
1606 pub fn layout(&self) -> Layout {
1607 let table = &self.table;
1608 let stripes = table.stripes.as_slice();
1609 let columns = table
1610 .fields
1611 .iter()
1612 .enumerate()
1613 .map(|(at, field)| ColumnLayout {
1614 name: field.name.clone(),
1615 kind: field.ty.to_string(),
1616 pages: sum(stripes.iter().map(|stripe| span_bytes(&stripe.pages, at))),
1617 memberships: sum(stripes.iter().map(|stripe| page_bytes(&stripe.memberships, at))),
1618 sieves: sum(stripes.iter().map(|stripe| page_bytes(&stripe.sieves, at))),
1619 dictionary: page_bytes(&table.dictionaries, at),
1620 })
1621 .collect();
1622 Layout {
1623 file: self.size,
1624 rows: table.rows,
1625 stripes: stripes.len(),
1626 parts: self.places.len(),
1627 columns,
1628 indexes: sum(stripes.iter().map(|stripe| u64::from(stripe.index.length))),
1629 directory: self.directory,
1630 header: HEADER,
1631 }
1632 }
1633
1634 #[must_use]
1636 pub fn parts(&self) -> usize {
1637 self.places.len()
1638 }
1639
1640 #[must_use]
1647 pub fn stripe_parts(&self) -> Vec<std::ops::Range<usize>> {
1648 let mut runs = Vec::with_capacity(self.table.stripes.len());
1649 let mut start = 0;
1650 for stripe in &self.table.stripes {
1651 let end = start + stripe.parts.len();
1652 runs.push(start..end);
1653 start = end;
1654 }
1655 runs
1656 }
1657
1658 pub fn keep_stripes(&self, stripes: usize) {
1665 self.kept.fetch_max(stripes, Atomic::Relaxed);
1666 }
1667
1668 #[must_use]
1670 pub fn part_rows(&self, at: usize) -> usize {
1671 self.places.get(at).map_or(0, |place| place.rows as usize)
1672 }
1673
1674 #[must_use]
1676 pub fn table(&self) -> &Table {
1677 &self.table
1678 }
1679
1680 pub fn top_frequencies(&self, column: usize, top: usize) -> Result<Option<Vec<(Value, u64)>>> {
1689 let field = self
1690 .table
1691 .fields
1692 .get(column)
1693 .ok_or_else(|| invalid("frequency column index out of range"))?;
1694 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1695 return Ok(None);
1696 };
1697 if top == 0 || summary.entries.len() < top {
1698 return Ok(None);
1699 }
1700 let boundary = summary.entries[top - 1].count;
1701 if boundary <= summary.omitted_max {
1702 return Ok(None);
1703 }
1704 let dictionary =
1705 if field.ty == LogicalType::Varchar { self.dictionary(column)? } else { None };
1706 let mut out = Vec::with_capacity(summary.entries.len());
1707 for entry in &summary.entries {
1708 let value = match entry.value {
1709 FrequencyValue::Null => Value::Null,
1710 FrequencyValue::Integer(value) => match field.ty {
1711 LogicalType::TinyInt => Value::TinyInt(
1712 i8::try_from(value)
1713 .map_err(|_| invalid("frequency TINYINT is out of range"))?,
1714 ),
1715 LogicalType::UTinyInt => Value::UTinyInt(
1716 u8::try_from(value)
1717 .map_err(|_| invalid("frequency UTINYINT is out of range"))?,
1718 ),
1719 LogicalType::USmallInt => Value::USmallInt(
1720 u16::try_from(value)
1721 .map_err(|_| invalid("frequency USMALLINT is out of range"))?,
1722 ),
1723 LogicalType::UInteger => Value::UInteger(
1724 u32::try_from(value)
1725 .map_err(|_| invalid("frequency UINTEGER is out of range"))?,
1726 ),
1727 LogicalType::UBigInt => Value::UBigInt(
1728 u64::try_from(value)
1729 .map_err(|_| invalid("frequency UBIGINT is out of range"))?,
1730 ),
1731 LogicalType::SmallInt => Value::SmallInt(
1732 i16::try_from(value)
1733 .map_err(|_| invalid("frequency SMALLINT is out of range"))?,
1734 ),
1735 LogicalType::Integer => Value::Integer(
1736 i32::try_from(value)
1737 .map_err(|_| invalid("frequency INTEGER is out of range"))?,
1738 ),
1739 LogicalType::BigInt => Value::BigInt(
1740 i64::try_from(value)
1741 .map_err(|_| invalid("frequency BIGINT is out of range"))?,
1742 ),
1743 LogicalType::Date => Value::Date(
1744 i32::try_from(value)
1745 .map_err(|_| invalid("frequency DATE is out of range"))?,
1746 ),
1747 LogicalType::Timestamp => Value::Timestamp(
1748 i64::try_from(value)
1749 .map_err(|_| invalid("frequency TIMESTAMP is out of range"))?,
1750 ),
1751 _ => return Err(invalid("integer frequency belongs to another type")),
1752 },
1753 FrequencyValue::Code(code) => dictionary
1754 .as_ref()
1755 .ok_or_else(|| invalid("frequency code has no dictionary"))?
1756 .try_value_at(code as usize)?,
1757 };
1758 out.push((value, entry.count));
1759 }
1760 Ok(Some(out))
1761 }
1762
1763 pub fn frequency_occurrences(&self, column: usize) -> Result<Option<FrequencyOccurrences>> {
1773 self.table
1774 .fields
1775 .get(column)
1776 .ok_or_else(|| invalid("frequency column index out of range"))?;
1777 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1778 return Ok(None);
1779 };
1780 if summary.ordinals.is_empty() {
1781 return Ok(None);
1782 }
1783 Ok(Some(FrequencyOccurrences {
1784 omitted_max: summary.omitted_max,
1785 ordinals: summary.ordinals.clone(),
1786 }))
1787 }
1788
1789 pub fn distinct_values(&self, column: usize) -> Result<Option<u64>> {
1809 if self.null_count(column)? > 0 {
1810 return Ok(None);
1811 }
1812 Ok(self.dictionary(column)?.map(|dictionary| dictionary.len() as u64))
1813 }
1814
1815 pub fn null_count(&self, column: usize) -> Result<u64> {
1826 if column >= self.table.fields.len() {
1827 return Err(invalid("null count column index out of range"));
1828 }
1829 let mut nulls = 0_u64;
1830 for stripe in &self.table.stripes {
1831 let range = stripe
1832 .zone
1833 .column(column)
1834 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
1835 nulls = nulls
1836 .checked_add(range.nulls as u64)
1837 .ok_or_else(|| invalid("null count overflow"))?;
1838 }
1839 Ok(nulls)
1840 }
1841
1842 pub fn text_extremes(&self, column: usize) -> Result<Option<(Value, Value)>> {
1857 if self.null_count(column)? > 0 {
1858 return Ok(None);
1859 }
1860 let Some(dictionary) = self.dictionary(column)? else { return Ok(None) };
1861 let Some(ranks) = dictionary.ranks() else { return Ok(None) };
1862 if ranks == 0 {
1863 return Ok(None);
1864 }
1865 let low = text_at_rank(&dictionary, 0)?;
1866 let high = text_at_rank(&dictionary, ranks - 1)?;
1867 Ok(Some((low, high)))
1868 }
1869
1870 pub fn exact_extremes(&self, column: usize) -> Result<Option<(Bound, Bound)>> {
1893 if column >= self.table.fields.len() {
1894 return Err(invalid("extremes column index out of range"));
1895 }
1896 let mut low: Option<Bound> = None;
1897 let mut high: Option<Bound> = None;
1898 for stripe in &self.table.stripes {
1899 let range = stripe
1900 .zone
1901 .column(column)
1902 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
1903 if !range.exact {
1904 return Ok(None);
1905 }
1906 let (Some(small), Some(large)) = (range.low.as_ref(), range.high.as_ref()) else {
1911 if stripe.rows > range.nulls {
1912 return Ok(None);
1913 }
1914 continue;
1915 };
1916 low = Some(low.map_or_else(|| small.clone(), |held| held.smaller(small.clone())));
1917 high = Some(high.map_or_else(|| large.clone(), |held| held.larger(large.clone())));
1918 }
1919 Ok(low.zip(high))
1920 }
1921
1922 pub fn exact_sum(&self, column: usize) -> Result<Option<(i128, u64)>> {
1935 if column >= self.table.fields.len() {
1936 return Err(invalid("sum column index out of range"));
1937 }
1938 let mut total = 0_i128;
1939 let mut rows = 0_u64;
1940 for stripe in &self.table.stripes {
1941 let range = stripe
1942 .zone
1943 .column(column)
1944 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
1945 let Some(part) = range.sum else { return Ok(None) };
1946 let Some(sum) = total.checked_add(part) else { return Ok(None) };
1947 total = sum;
1948 rows = rows.saturating_add(stripe.rows as u64 - range.nulls as u64);
1949 }
1950 Ok(Some((total, rows)))
1951 }
1952
1953 fn dictionary(&self, column: usize) -> Result<Option<Arc<Vector>>> {
1954 let Some(page) = self.table.dictionaries[column] else { return Ok(None) };
1955 if let Some(dictionary) = self.dictionaries[column].get() {
1956 return Ok(Some(Arc::clone(dictionary)));
1957 }
1958 let dictionary = Arc::new(open_global_dictionary(
1959 Arc::clone(&self.file),
1960 page,
1961 &self.table.fields[column].ty,
1962 )?);
1963 let _ = self.dictionaries[column].set(Arc::clone(&dictionary));
1964 Ok(Some(self.dictionaries[column].get().map_or(dictionary, Arc::clone)))
1965 }
1966
1967 pub fn read(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
1976 self.read_impl(part, columns, true)
1977 }
1978
1979 pub fn read_sparse(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
1989 self.read_impl(part, columns, false)
1990 }
1991
1992 pub fn skips_codes(&self, part: usize, column: usize, candidates: &[u32]) -> Result<bool> {
1999 if candidates.is_empty() {
2000 return Ok(true);
2001 }
2002 if candidates.windows(2).any(|pair| pair[0] >= pair[1]) {
2003 return Err(Error::internal("native code candidates are not sorted and unique"));
2004 }
2005 let stripe = self.stripe_of(part)?;
2006 let Some(page) = stripe.memberships.get(column).copied().flatten() else {
2007 return Ok(false);
2008 };
2009 let mut bytes = vec![0; page.length as usize];
2010 read_at(&self.file, page.offset, &mut bytes)?;
2011 if checksum(&bytes) != page.hash {
2012 return Err(invalid("membership page checksum differs"));
2013 }
2014 let codes = decode_membership(&bytes)?;
2015 let mut left = 0;
2016 let mut right = 0;
2017 while left < codes.len() && right < candidates.len() {
2018 match codes[left].cmp(&candidates[right]) {
2019 Ordering::Less => left += 1,
2020 Ordering::Greater => right += 1,
2021 Ordering::Equal => return Ok(false),
2022 }
2023 }
2024 Ok(true)
2025 }
2026
2027 fn stripe_of(&self, part: usize) -> Result<&Stripe> {
2028 let place = self.places.get(part).ok_or_else(|| invalid("part index out of range"))?;
2029 self.table
2030 .stripes
2031 .get(place.stripe as usize)
2032 .ok_or_else(|| invalid("stripe index out of range"))
2033 }
2034
2035 fn held(&self, at: usize, stripe: &Stripe, column: usize, whole: bool) -> Result<CachedColumn> {
2052 let cache = self.cache.get(column).ok_or_else(|| invalid("column index out of range"))?;
2053 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2054 let known = cached.index.get(at).and_then(Clone::clone);
2055 let page = cached.pages.get(at).and_then(Clone::clone);
2056 if let Some(index) = known.clone() {
2057 if !whole || page.is_some() {
2058 return Ok(CachedColumn { stripe: at, index, page });
2059 }
2060 }
2061 if cached.loading.contains(&at) {
2062 drop(cached);
2063 if let Some(index) = known {
2067 return Ok(CachedColumn { stripe: at, index, page: None });
2068 }
2069 let held = self.page_of(stripe, column, at, false, None)?;
2070 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2071 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2072 return Ok(held);
2073 }
2074 cached.loading.push(at);
2075 drop(cached);
2076
2077 let read = self.page_of(stripe, column, at, whole, known);
2078
2079 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2083 if let Some(position) = cached.loading.iter().position(|loading| *loading == at) {
2084 cached.loading.remove(position);
2085 }
2086 let held = read?;
2087 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2088 Ok(held)
2089 }
2090
2091 fn page_of(
2097 &self,
2098 stripe: &Stripe,
2099 column: usize,
2100 at: usize,
2101 whole: bool,
2102 known: Option<Arc<Vec<PartSpan>>>,
2103 ) -> Result<CachedColumn> {
2104 let index = match known {
2105 Some(index) => index,
2106 None => {
2107 self.indexes.fetch_add(1, Atomic::Relaxed);
2108 Arc::new(read_index(&self.file, stripe, column)?)
2109 }
2110 };
2111 let page = if whole {
2112 self.pages.fetch_add(1, Atomic::Relaxed);
2113 let span = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2114 let mut bytes = vec![0; span.length as usize];
2115 read_at(&self.file, span.offset, &mut bytes)?;
2116 Some(Arc::new(bytes))
2117 } else {
2118 None
2119 };
2120 Ok(CachedColumn { stripe: at, index, page })
2121 }
2122
2123 fn read_impl(&self, at: usize, columns: &[usize], whole: bool) -> Result<Chunk> {
2124 let place = *self.places.get(at).ok_or_else(|| invalid("part index out of range"))?;
2125 let index = place.stripe as usize;
2126 let stripe =
2127 self.table.stripes.get(index).ok_or_else(|| invalid("stripe index out of range"))?;
2128 let rows = place.rows as usize;
2129 let mut picked = Vec::with_capacity(columns.len());
2130 for &column in columns {
2131 let field = self
2132 .table
2133 .fields
2134 .get(column)
2135 .ok_or_else(|| invalid("column index out of range"))?;
2136 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2137 let held = self.held(index, stripe, column, whole)?;
2138 let span = *held
2139 .index
2140 .get(place.part as usize)
2141 .ok_or_else(|| invalid("part index out of range"))?;
2142 let owned;
2143 let bytes = match &held.page {
2144 Some(held) => part_bytes(held, span)?,
2145 None => {
2146 let offset = page
2147 .offset
2148 .checked_add(span.start as u64)
2149 .ok_or_else(|| invalid("part range overflow"))?;
2150 let mut bytes = vec![0; span.length];
2151 read_at(&self.file, offset, &mut bytes)?;
2152 owned = bytes;
2153 &owned
2154 }
2155 };
2156 if checksum(bytes) != span.hash {
2157 return Err(invalid(&format!(
2158 "column page checksum differs, column {column} part {} at {}+{} of {} bytes, \
2159 wanted {:016x} and got {:016x}",
2160 place.part,
2161 page.offset,
2162 span.start,
2163 span.length,
2164 span.hash,
2165 checksum(bytes),
2166 )));
2167 }
2168 let dictionary = self.dictionary(column)?;
2169 picked.push(decode(&field.ty, rows, bytes, dictionary)?);
2170 }
2171 Chunk::with_rows(picked, rows)
2172 }
2173
2174 #[must_use]
2184 pub fn skips(&self, part: usize, probes: &[Probe]) -> bool {
2185 let Some(place) = self.places.get(part).copied() else { return false };
2186 let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
2187 if stripe.zone.skips(probes) {
2188 return true;
2189 }
2190 probes.iter().any(|probe| self.sifted(place, probe))
2191 }
2192
2193 fn sifted(&self, place: Place, probe: &Probe) -> bool {
2199 if probe.op != Op::Equal {
2200 return false;
2201 }
2202 match self.stripe_sieves(place.stripe as usize, probe.column) {
2203 Some(sieves) => sieves
2204 .get(place.part as usize)
2205 .and_then(Option::as_ref)
2206 .is_some_and(|sieve| sieve.excludes(&probe.value)),
2207 None => false,
2208 }
2209 }
2210
2211 fn stripe_sieves(&self, stripe: usize, column: usize) -> Option<&[Option<Sieve>]> {
2218 let slot = self.sieves.get(column)?.get(stripe)?;
2219 if let Some(held) = slot.get() {
2220 return Some(held);
2221 }
2222 let page = self.table.stripes.get(stripe)?.sieves.get(column).copied().flatten()?;
2223 let mut bytes = vec![0; page.length as usize];
2224 read_at(&self.file, page.offset, &mut bytes).ok()?;
2225 if checksum(&bytes) != page.hash {
2226 return None;
2227 }
2228 let sieves = Arc::new(decode_sieves(&bytes).ok()?);
2229 let _ = slot.set(sieves);
2230 slot.get().map(|held| held.as_slice())
2231 }
2232}
2233
2234fn text_at_rank(dictionary: &Vector, rank: usize) -> Result<Value> {
2236 let code = dictionary.code_at_rank(rank)? as usize;
2237 let text = dictionary
2238 .try_text_at(code)?
2239 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
2240 Ok(Value::Varchar(text.into()))
2241}
2242
2243#[cfg(unix)]
2248fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2249 use std::os::unix::fs::FileExt;
2250 while !bytes.is_empty() {
2251 let written = file.write_at(bytes, offset).map_err(io)?;
2252 if written == 0 {
2253 return Err(invalid("a write to the native file wrote nothing"));
2254 }
2255 offset += written as u64;
2256 bytes = &bytes[written..];
2257 }
2258 Ok(())
2259}
2260
2261#[cfg(windows)]
2263fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2264 use std::os::windows::fs::FileExt;
2265 while !bytes.is_empty() {
2266 let written = file.seek_write(bytes, offset).map_err(io)?;
2267 if written == 0 {
2268 return Err(invalid("a write to the native file wrote nothing"));
2269 }
2270 offset += written as u64;
2271 bytes = &bytes[written..];
2272 }
2273 Ok(())
2274}
2275
2276#[cfg(not(any(unix, windows)))]
2278fn write_at(file: &File, offset: u64, bytes: &[u8]) -> Result<()> {
2279 use std::io::Write;
2280 let mut file = file.try_clone().map_err(io)?;
2281 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2282 file.write_all(bytes).map_err(io)
2283}
2284
2285#[cfg(unix)]
2295fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2296 use std::os::unix::fs::FileExt;
2297 while !bytes.is_empty() {
2298 let read = file.read_at(bytes, offset).map_err(io)?;
2299 if read == 0 {
2300 return Err(invalid("column page ends before its declared length"));
2301 }
2302 offset += read as u64;
2303 bytes = &mut bytes[read..];
2304 }
2305 Ok(())
2306}
2307
2308#[cfg(windows)]
2314fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2315 use std::os::windows::fs::FileExt;
2316 while !bytes.is_empty() {
2317 let read = file.seek_read(bytes, offset).map_err(io)?;
2318 if read == 0 {
2319 return Err(invalid("column page ends before its declared length"));
2320 }
2321 offset += read as u64;
2322 bytes = &mut bytes[read..];
2323 }
2324 Ok(())
2325}
2326
2327#[cfg(not(any(unix, windows)))]
2332fn read_at(file: &File, offset: u64, bytes: &mut [u8]) -> Result<()> {
2333 let mut file = file.try_clone().map_err(io)?;
2334 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2335 file.read_exact(bytes).map_err(io)
2336}
2337
2338fn type_tag(ty: &LogicalType) -> Result<u8> {
2339 match ty {
2340 LogicalType::SmallInt => Ok(1),
2341 LogicalType::Integer => Ok(2),
2342 LogicalType::BigInt => Ok(3),
2343 LogicalType::Varchar => Ok(4),
2344 LogicalType::Date => Ok(5),
2345 LogicalType::Timestamp => Ok(6),
2346 LogicalType::Boolean => Ok(7),
2347 LogicalType::TinyInt => Ok(8),
2348 LogicalType::UTinyInt => Ok(9),
2349 LogicalType::USmallInt => Ok(10),
2350 LogicalType::UInteger => Ok(11),
2351 LogicalType::UBigInt => Ok(12),
2352 _ => Err(Error::not_implemented(format!("native storage for {ty}"))),
2353 }
2354}
2355
2356fn tag_type(tag: u8) -> Result<LogicalType> {
2357 match tag {
2358 1 => Ok(LogicalType::SmallInt),
2359 2 => Ok(LogicalType::Integer),
2360 3 => Ok(LogicalType::BigInt),
2361 4 => Ok(LogicalType::Varchar),
2362 5 => Ok(LogicalType::Date),
2363 6 => Ok(LogicalType::Timestamp),
2364 7 => Ok(LogicalType::Boolean),
2365 8 => Ok(LogicalType::TinyInt),
2366 9 => Ok(LogicalType::UTinyInt),
2367 10 => Ok(LogicalType::USmallInt),
2368 11 => Ok(LogicalType::UInteger),
2369 12 => Ok(LogicalType::UBigInt),
2370 _ => Err(invalid("column type tag is unknown")),
2371 }
2372}
2373
2374fn put_u16(out: &mut Vec<u8>, value: u16) {
2375 out.extend_from_slice(&value.to_le_bytes());
2376}
2377fn put_u32(out: &mut Vec<u8>, value: u32) {
2378 out.extend_from_slice(&value.to_le_bytes());
2379}
2380fn put_u64(out: &mut Vec<u8>, value: u64) {
2381 out.extend_from_slice(&value.to_le_bytes());
2382}
2383fn put_var_u64(out: &mut Vec<u8>, mut value: u64) {
2384 while value >= 0x80 {
2385 out.push((value as u8 & 0x7f) | 0x80);
2386 value >>= 7;
2387 }
2388 out.push(value as u8);
2389}
2390
2391fn frequency_order(left: FrequencyValue, right: FrequencyValue) -> Ordering {
2392 match (left, right) {
2393 (FrequencyValue::Null, FrequencyValue::Null) => Ordering::Equal,
2394 (FrequencyValue::Null, _) => Ordering::Less,
2395 (_, FrequencyValue::Null) => Ordering::Greater,
2396 (FrequencyValue::Integer(left), FrequencyValue::Integer(right)) => left.cmp(&right),
2397 (FrequencyValue::Code(left), FrequencyValue::Code(right)) => left.cmp(&right),
2398 (FrequencyValue::Integer(_), FrequencyValue::Code(_)) => Ordering::Less,
2399 (FrequencyValue::Code(_), FrequencyValue::Integer(_)) => Ordering::Greater,
2400 }
2401}
2402
2403fn code_frequency(dictionary: &GlobalDictionary) -> FrequencySummary {
2404 let mut entries = dictionary
2405 .counts
2406 .iter()
2407 .enumerate()
2408 .filter(|(_, count)| **count != 0)
2409 .map(|(code, &count)| FrequencyEntry { value: FrequencyValue::Code(code as u32), count })
2410 .collect::<Vec<_>>();
2411 if dictionary.nulls != 0 {
2412 entries.push(FrequencyEntry { value: FrequencyValue::Null, count: dictionary.nulls });
2413 }
2414 entries.sort_unstable_by(|left, right| {
2415 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
2416 });
2417 let omitted_max = entries.get(FREQUENCY_ENTRIES).map_or(0, |entry| entry.count);
2418 entries.truncate(FREQUENCY_ENTRIES);
2419 FrequencySummary { entries, omitted_max, ordinals: Vec::new() }
2420}
2421
2422fn encode_directory(table: &Table) -> Result<Vec<u8>> {
2423 let mut out = DIRECTORY.to_vec();
2424 let name = table.name.as_bytes();
2425 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
2426 out.extend_from_slice(name);
2427 put_u16(&mut out, u16::try_from(table.fields.len()).map_err(|_| invalid("too many columns"))?);
2428 for field in &table.fields {
2429 let name = field.name.as_bytes();
2430 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?);
2431 out.extend_from_slice(name);
2432 out.push(type_tag(&field.ty)?);
2433 out.push(u8::from(field.not_null));
2434 }
2435 for dictionary in &table.dictionaries {
2436 match dictionary {
2437 None => out.push(0),
2438 Some(page) => {
2439 out.push(1);
2440 put_u64(&mut out, page.offset);
2441 put_u32(&mut out, page.length);
2442 put_u64(&mut out, page.hash);
2443 }
2444 }
2445 }
2446 put_u64(&mut out, u64::try_from(table.rows).map_err(|_| invalid("row count overflow"))?);
2447 put_u32(&mut out, u32::try_from(table.stripes.len()).map_err(|_| invalid("too many stripes"))?);
2448 for stripe in &table.stripes {
2449 put_u32(
2450 &mut out,
2451 u32::try_from(stripe.parts.len()).map_err(|_| invalid("too many parts in a stripe"))?,
2452 );
2453 for &rows in &stripe.parts {
2454 put_u32(&mut out, rows);
2455 }
2456 put_u64(&mut out, stripe.index.offset);
2457 put_u32(&mut out, stripe.index.length);
2458 for page in &stripe.pages {
2459 put_u64(&mut out, page.offset);
2460 put_u32(&mut out, page.length);
2461 }
2462 for (field, membership) in table.fields.iter().zip(&stripe.memberships) {
2463 if field.ty != LogicalType::Varchar {
2464 continue;
2465 }
2466 let page =
2467 membership.ok_or_else(|| invalid("string page has no code membership index"))?;
2468 put_u64(&mut out, page.offset);
2469 put_u32(&mut out, page.length);
2470 put_u64(&mut out, page.hash);
2471 }
2472 for sieve in &stripe.sieves {
2473 match sieve {
2474 None => out.push(0),
2475 Some(page) => {
2476 out.push(1);
2477 put_u64(&mut out, page.offset);
2478 put_u32(&mut out, page.length);
2479 put_u64(&mut out, page.hash);
2480 }
2481 }
2482 }
2483 for range in stripe.zone.columns() {
2484 put_bound(&mut out, range.low.as_ref())?;
2485 put_bound(&mut out, range.high.as_ref())?;
2486 put_u32(
2487 &mut out,
2488 u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?,
2489 );
2490 out.push(u8::from(range.exact));
2491 match range.sum {
2492 None => out.push(0),
2493 Some(total) => {
2494 out.push(1);
2495 out.extend_from_slice(&total.to_le_bytes());
2496 }
2497 }
2498 }
2499 }
2500 out.extend_from_slice(FREQUENCIES);
2501 put_u16(
2502 &mut out,
2503 u16::try_from(table.frequencies.len())
2504 .map_err(|_| invalid("too many frequency columns"))?,
2505 );
2506 for summary in &table.frequencies {
2507 let Some(summary) = summary else {
2508 out.push(0);
2509 continue;
2510 };
2511 out.push(1);
2512 put_u64(&mut out, summary.omitted_max);
2513 put_u32(
2514 &mut out,
2515 u32::try_from(summary.entries.len())
2516 .map_err(|_| invalid("too many frequency entries"))?,
2517 );
2518 for entry in &summary.entries {
2519 match entry.value {
2520 FrequencyValue::Null => out.push(0),
2521 FrequencyValue::Integer(value) => {
2522 out.push(1);
2523 out.extend_from_slice(&value.to_le_bytes());
2524 }
2525 FrequencyValue::Code(value) => {
2526 out.push(2);
2527 put_u32(&mut out, value);
2528 }
2529 }
2530 put_u64(&mut out, entry.count);
2531 }
2532 put_u32(
2533 &mut out,
2534 u32::try_from(summary.ordinals.len())
2535 .map_err(|_| invalid("too many frequency ordinals"))?,
2536 );
2537 let mut previous = 0_u64;
2538 for (at, &ordinal) in summary.ordinals.iter().enumerate() {
2539 let delta = if at == 0 {
2540 ordinal
2541 } else {
2542 ordinal
2543 .checked_sub(previous)
2544 .ok_or_else(|| invalid("frequency ordinals are not ordered"))?
2545 };
2546 if at != 0 && delta == 0 {
2547 return Err(invalid("frequency ordinals are not unique"));
2548 }
2549 put_var_u64(&mut out, delta);
2550 previous = ordinal;
2551 }
2552 }
2553 Ok(out)
2554}
2555
2556struct Cursor<'a> {
2557 bytes: &'a [u8],
2558 at: usize,
2559}
2560impl<'a> Cursor<'a> {
2561 fn take(&mut self, len: usize) -> Result<&'a [u8]> {
2562 let end = self.at.checked_add(len).ok_or_else(|| invalid("directory offset overflow"))?;
2563 let bytes =
2564 self.bytes.get(self.at..end).ok_or_else(|| invalid("directory is truncated"))?;
2565 self.at = end;
2566 Ok(bytes)
2567 }
2568 fn u8(&mut self) -> Result<u8> {
2569 Ok(self.take(1)?[0])
2570 }
2571 fn u16(&mut self) -> Result<u16> {
2572 Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("two bytes")))
2573 }
2574 fn u32(&mut self) -> Result<u32> {
2575 Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("four bytes")))
2576 }
2577 fn u64(&mut self) -> Result<u64> {
2578 Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("eight bytes")))
2579 }
2580 fn var_u64(&mut self) -> Result<u64> {
2581 let mut value = 0_u64;
2582 for shift in (0..=63).step_by(7) {
2583 let byte = self.u8()?;
2584 let part = u64::from(byte & 0x7f);
2585 if shift == 63 && part > 1 {
2586 return Err(invalid("frequency ordinal varint overflows"));
2587 }
2588 value |= part << shift;
2589 if byte & 0x80 == 0 {
2590 return Ok(value);
2591 }
2592 }
2593 Err(invalid("frequency ordinal varint is too long"))
2594 }
2595 fn bound(&mut self) -> Result<Option<Bound>> {
2596 Ok(match self.u8()? {
2597 0 => None,
2598 1 => Some(Bound::Int(i128::from_le_bytes(
2599 self.take(16)?.try_into().expect("sixteen bytes"),
2600 ))),
2601 2 => Some(Bound::Real(f64::from_le_bytes(
2602 self.take(8)?.try_into().expect("eight bytes"),
2603 ))),
2604 3 => {
2605 let length = self.u32()? as usize;
2606 Some(Bound::Bytes(self.take(length)?.to_vec()))
2607 }
2608 _ => return Err(invalid("bound tag differs")),
2609 })
2610 }
2611 fn text(&mut self) -> Result<String> {
2612 let len = self.u16()? as usize;
2613 String::from_utf8(self.take(len)?.to_vec()).map_err(|_| invalid("name is not UTF-8"))
2614 }
2615}
2616
2617fn decode_directory(bytes: &[u8], size: u64) -> Result<Table> {
2618 let mut cur = Cursor { bytes, at: 0 };
2619 if cur.take(8)? != DIRECTORY {
2620 return Err(invalid("directory magic differs"));
2621 }
2622 let name = cur.text()?;
2623 let width = cur.u16()? as usize;
2624 let mut fields = Vec::with_capacity(width);
2625 for _ in 0..width {
2626 let name = cur.text()?;
2627 let ty = tag_type(cur.u8()?)?;
2628 let not_null = match cur.u8()? {
2629 0 => false,
2630 1 => true,
2631 _ => return Err(invalid("nullability flag differs")),
2632 };
2633 fields.push(Field { name, ty, not_null });
2634 }
2635 let mut dictionaries = Vec::with_capacity(width);
2636 for _ in 0..width {
2637 dictionaries.push(match cur.u8()? {
2638 0 => None,
2639 1 => {
2640 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2641 let end = page
2642 .offset
2643 .checked_add(u64::from(page.length))
2644 .ok_or_else(|| invalid("dictionary page offset overflow"))?;
2645 if page.offset < HEADER || end > size {
2650 return Err(invalid("dictionary page range is outside the file"));
2651 }
2652 Some(page)
2653 }
2654 _ => return Err(invalid("dictionary page tag differs")),
2655 });
2656 }
2657 let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
2658 let count = cur.u32()? as usize;
2659 let mut stripes = Vec::with_capacity(count);
2660 let mut total = 0_usize;
2661 for _ in 0..count {
2662 let count = cur.u32()? as usize;
2663 if count == 0 || count > STRIPE_PARTS {
2664 return Err(invalid("stripe part count is outside its bound"));
2665 }
2666 let mut parts = Vec::with_capacity(count);
2667 let mut stripe_rows = 0_usize;
2668 for _ in 0..count {
2669 let rows = cur.u32()?;
2670 if rows == 0 {
2671 return Err(invalid("empty part"));
2672 }
2673 parts.push(rows);
2674 stripe_rows = stripe_rows
2675 .checked_add(rows as usize)
2676 .ok_or_else(|| invalid("stripe row count overflow"))?;
2677 }
2678 total =
2679 total.checked_add(stripe_rows).ok_or_else(|| invalid("stripe row count overflow"))?;
2680 let index = Span { offset: cur.u64()?, length: cur.u32()? };
2681 let section = index_section(count)?;
2682 let wanted = section
2683 .checked_mul(width)
2684 .and_then(|bytes| u32::try_from(bytes).ok())
2685 .ok_or_else(|| invalid("index page length overflow"))?;
2686 let end = index
2687 .offset
2688 .checked_add(u64::from(index.length))
2689 .ok_or_else(|| invalid("index page offset overflow"))?;
2690 if index.offset < HEADER || end > size || index.length != wanted {
2691 return Err(invalid("index page range is outside the file"));
2692 }
2693 let mut pages = Vec::with_capacity(width);
2694 for _ in 0..width {
2695 let offset = cur.u64()?;
2696 let length = cur.u32()?;
2697 let end = offset
2698 .checked_add(u64::from(length))
2699 .ok_or_else(|| invalid("page offset overflow"))?;
2700 if offset < HEADER || end > size || length as usize > MAX_PAGE {
2701 return Err(invalid("page range is outside the file"));
2702 }
2703 pages.push(Span { offset, length });
2704 }
2705 let mut memberships = vec![None; width];
2706 for (column, field) in fields.iter().enumerate() {
2707 if field.ty != LogicalType::Varchar {
2708 continue;
2709 }
2710 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2711 let end = page
2712 .offset
2713 .checked_add(u64::from(page.length))
2714 .ok_or_else(|| invalid("membership page offset overflow"))?;
2715 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
2716 return Err(invalid("membership page range is outside the file"));
2717 }
2718 memberships[column] = Some(page);
2719 }
2720 let mut sieves = vec![None; width];
2721 for sieve in sieves.iter_mut().take(width) {
2722 match cur.u8()? {
2723 0 => continue,
2724 1 => {}
2725 _ => return Err(invalid("a sieve page has an unknown tag")),
2726 }
2727 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2728 let end = page
2729 .offset
2730 .checked_add(u64::from(page.length))
2731 .ok_or_else(|| invalid("sieve page offset overflow"))?;
2732 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
2733 return Err(invalid("sieve page range is outside the file"));
2734 }
2735 *sieve = Some(page);
2736 }
2737 let mut ranges = Vec::with_capacity(width);
2738 for _ in 0..width {
2739 let low = cur.bound()?;
2740 let high = cur.bound()?;
2741 let nulls = cur.u32()? as usize;
2742 if nulls > stripe_rows {
2743 return Err(invalid("null count exceeds stripe rows"));
2744 }
2745 let exact = cur.u8()? != 0;
2746 let sum = match cur.u8()? {
2747 0 => None,
2748 1 => Some(i128::from_le_bytes(
2749 cur.take(16)?.try_into().map_err(|_| invalid("a stripe sum is truncated"))?,
2750 )),
2751 _ => return Err(invalid("a stripe sum has an unknown tag")),
2752 };
2753 ranges.push(Range { low, high, nulls, exact, sum });
2754 }
2755 stripes.push(Stripe {
2756 rows: stripe_rows,
2757 parts,
2758 index,
2759 pages,
2760 memberships,
2761 sieves,
2762 zone: Zone::from_ranges(ranges),
2763 });
2764 }
2765 if total != rows {
2766 return Err(invalid("table row count differs from stripes"));
2767 }
2768 let frequencies = if cur.at == bytes.len() {
2769 vec![None; width]
2770 } else {
2771 if cur.take(8)? != FREQUENCIES {
2772 return Err(invalid("directory extension magic differs"));
2773 }
2774 if cur.u16()? as usize != width {
2775 return Err(invalid("frequency column count differs"));
2776 }
2777 let mut frequencies = Vec::with_capacity(width);
2778 for field in &fields {
2779 let summary = match cur.u8()? {
2780 0 => None,
2781 1 => {
2782 let omitted_max = cur.u64()?;
2783 let count = cur.u32()? as usize;
2784 if count > FREQUENCY_ENTRIES {
2785 return Err(invalid("frequency entry count exceeds its bound"));
2786 }
2787 let mut entries = Vec::with_capacity(count);
2788 for _ in 0..count {
2790 let value = match cur.u8()? {
2791 0 => FrequencyValue::Null,
2792 1 => FrequencyValue::Integer(i128::from_le_bytes(
2793 cur.take(16)?.try_into().expect("sixteen bytes"),
2794 )),
2795 2 => FrequencyValue::Code(cur.u32()?),
2796 _ => return Err(invalid("frequency value tag differs")),
2797 };
2798 let valid = matches!(
2799 (&field.ty, value),
2800 (_, FrequencyValue::Null)
2801 | (LogicalType::Varchar, FrequencyValue::Code(_))
2802 | (
2803 LogicalType::TinyInt
2804 | LogicalType::SmallInt
2805 | LogicalType::Integer
2806 | LogicalType::BigInt
2807 | LogicalType::UTinyInt
2808 | LogicalType::USmallInt
2809 | LogicalType::UInteger
2810 | LogicalType::UBigInt
2811 | LogicalType::Date
2812 | LogicalType::Timestamp,
2813 FrequencyValue::Integer(_),
2814 )
2815 );
2816 if !valid {
2817 return Err(invalid("frequency value does not match its column"));
2818 }
2819 let count = cur.u64()?;
2820 if count == 0 || count > rows as u64 {
2821 return Err(invalid("frequency count is outside the table"));
2822 }
2823 entries.push(FrequencyEntry { value, count });
2824 }
2825 if entries.windows(2).any(|pair| pair[0].count < pair[1].count) {
2826 return Err(invalid("frequency entries are not descending"));
2827 }
2828 let ordinals = {
2829 let ordinal_count = cur.u32()? as usize;
2830 if ordinal_count > FREQUENCY_ORDINALS || ordinal_count > rows {
2831 return Err(invalid("frequency ordinal count exceeds its bound"));
2832 }
2833 let mut ordinals = Vec::with_capacity(ordinal_count);
2834 let mut previous = 0_u64;
2835 for at in 0..ordinal_count {
2836 let delta = cur.var_u64()?;
2837 if at != 0 && delta == 0 {
2838 return Err(invalid("frequency ordinals are not increasing"));
2839 }
2840 let ordinal = if at == 0 {
2841 delta
2842 } else {
2843 previous
2844 .checked_add(delta)
2845 .ok_or_else(|| invalid("frequency ordinal overflows"))?
2846 };
2847 if ordinal >= rows as u64 {
2848 return Err(invalid("frequency ordinal is outside the table"));
2849 }
2850 ordinals.push(ordinal);
2851 previous = ordinal;
2852 }
2853 ordinals
2854 };
2855 Some(FrequencySummary { entries, omitted_max, ordinals })
2856 }
2857 _ => return Err(invalid("frequency summary tag differs")),
2858 };
2859 frequencies.push(summary);
2860 }
2861 frequencies
2862 };
2863 if cur.at != bytes.len() {
2864 return Err(invalid("directory has trailing bytes"));
2865 }
2866 Ok(Table { name, fields, stripes, rows, dictionaries, frequencies })
2867}
2868
2869fn put_bound(out: &mut Vec<u8>, bound: Option<&Bound>) -> Result<()> {
2870 match bound {
2871 None => out.push(0),
2872 Some(Bound::Int(value)) => {
2873 out.push(1);
2874 out.extend_from_slice(&value.to_le_bytes());
2875 }
2876 Some(Bound::Real(value)) => {
2877 out.push(2);
2878 out.extend_from_slice(&value.to_le_bytes());
2879 }
2880 Some(Bound::Bytes(value)) => {
2881 out.push(3);
2882 put_u32(out, u32::try_from(value.len()).map_err(|_| invalid("bound length overflow"))?);
2883 out.extend_from_slice(value);
2884 }
2885 }
2886 Ok(())
2887}
2888
2889#[derive(Debug)]
2906struct Codes;
2907
2908impl chooser::Chooser for Codes {
2909 fn name(&self) -> &'static str {
2910 "codes"
2911 }
2912
2913 fn narrow_strings(
2914 &self,
2915 _values: &[&[u8]],
2916 offered: &[string::Kind],
2917 _depth: u8,
2918 ) -> Vec<string::Kind> {
2919 offered.to_vec()
2922 }
2923
2924 fn narrow_integers(
2925 &self,
2926 _values: &[i64],
2927 offered: &[integer::Kind],
2928 depth: u8,
2929 ) -> Vec<integer::Kind> {
2930 let keep: &[integer::Kind] = if depth == 0 {
2931 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Rle]
2932 } else {
2933 &[integer::Kind::Constant, integer::Kind::Packed]
2934 };
2935 let narrowed: Vec<integer::Kind> =
2936 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
2937 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
2940 }
2941}
2942
2943fn encoded_codes(codes: &[u32]) -> Result<Option<Vec<u8>>> {
2955 let wide: Vec<i64> = codes.iter().map(|code| i64::from(*code)).collect();
2956 let coded = integer::encode_with(&wide, &Codes)?;
2957 let plain = codes.len().saturating_mul(size_of::<u32>());
2958 Ok((coded.len() < plain).then_some(coded))
2959}
2960
2961fn encode(
2962 vector: &Vector,
2963 global: Option<&mut GlobalDictionary>,
2964) -> Result<(Vec<u8>, Option<Vec<u32>>)> {
2965 let ty = vector.logical_type();
2966 let flat = vector.flatten()?;
2968 let mut out = Vec::new();
2969 let mut global_codes = None;
2970 if let Some(global) = global {
2971 let mut codes = Vec::with_capacity(flat.len());
2972 for row in 0..flat.len() {
2973 let text = flat.text_at(row).unwrap_or("");
2974 let code = global.code(text)?;
2975 global.observe(code, flat.is_null_at(row))?;
2976 codes.push(code);
2977 }
2978 global_codes = Some(codes);
2979 }
2980 let membership = global_codes.as_deref().map(unique_codes);
2981 let dictionary = if global_codes.is_none() && ty == &LogicalType::Varchar {
2982 string_dictionary(&flat)?
2983 } else {
2984 None
2985 };
2986 let packed_vector = if dictionary.is_none() && global_codes.is_none() {
2987 Some(flat.bit_packed()?)
2988 } else {
2989 None
2990 };
2991 let packed = packed_vector.as_ref().and_then(Vector::packed_parts);
2992 let coded = match global_codes.as_deref() {
2993 Some(codes) => encoded_codes(codes)?,
2994 None => None,
2995 };
2996 out.push(if coded.is_some() {
2997 4
2998 } else if global_codes.is_some() {
2999 3
3000 } else if dictionary.is_some() {
3001 1
3002 } else if packed.is_some() {
3003 2
3004 } else {
3005 0
3006 });
3007 let nulls = flat.validity();
3008 let flag = match nulls {
3009 Validity::AllValid => 0,
3010 Validity::AllInvalid => 1,
3011 Validity::Mask(_) => 2,
3012 };
3013 out.push(flag);
3014 if flag == 2 {
3015 for group in (0..vector.len()).step_by(8) {
3016 let mut bits = 0_u8;
3017 for bit in 0..8 {
3018 if group + bit < vector.len() && !flat.is_null_at(group + bit) {
3019 bits |= 1 << bit;
3020 }
3021 }
3022 out.push(bits);
3023 }
3024 }
3025 if let Some(coded) = coded {
3026 out.extend_from_slice(&coded);
3027 return Ok((out, membership));
3028 }
3029 if let Some(codes) = global_codes {
3030 for code in codes {
3031 put_u32(&mut out, code);
3032 }
3033 return Ok((out, membership));
3034 }
3035 if let Some(dictionary) = dictionary {
3036 out.extend_from_slice(&dictionary);
3037 return Ok((out, membership));
3038 }
3039 if let Some(packed) = packed {
3040 if packed.offset() != 0 {
3041 return Err(invalid("writer received a sliced packed vector"));
3042 }
3043 out.push(u8::try_from(packed.width()).map_err(|_| invalid("packed width overflow"))?);
3044 out.extend_from_slice(&packed.base().to_le_bytes());
3045 put_u32(
3046 &mut out,
3047 u32::try_from(packed.words().len()).map_err(|_| invalid("too many packed words"))?,
3048 );
3049 for word in packed.words() {
3050 put_u64(&mut out, *word);
3051 }
3052 return Ok((out, membership));
3053 }
3054 let data = flat.data().ok_or_else(|| invalid("scalar column did not flatten"))?;
3055 match (ty, data) {
3056 (LogicalType::TinyInt, Data::Int8(values)) => {
3057 for value in &**values {
3058 out.extend_from_slice(&value.to_le_bytes());
3059 }
3060 }
3061 (LogicalType::UTinyInt, Data::UInt8(values)) => {
3062 for value in &**values {
3063 out.extend_from_slice(&value.to_le_bytes());
3064 }
3065 }
3066 (LogicalType::SmallInt, Data::Int16(values)) => {
3067 for value in &**values {
3068 out.extend_from_slice(&value.to_le_bytes());
3069 }
3070 }
3071 (LogicalType::USmallInt, Data::UInt16(values)) => {
3072 for value in &**values {
3073 out.extend_from_slice(&value.to_le_bytes());
3074 }
3075 }
3076 (LogicalType::UInteger, Data::UInt32(values)) => {
3077 for value in &**values {
3078 out.extend_from_slice(&value.to_le_bytes());
3079 }
3080 }
3081 (LogicalType::UBigInt, Data::UInt64(values)) => {
3082 for value in &**values {
3083 out.extend_from_slice(&value.to_le_bytes());
3084 }
3085 }
3086 (LogicalType::Integer | LogicalType::Date, Data::Int32(values)) => {
3087 for value in &**values {
3088 out.extend_from_slice(&value.to_le_bytes());
3089 }
3090 }
3091 (LogicalType::BigInt | LogicalType::Timestamp, Data::Int64(values)) => {
3092 for value in &**values {
3093 out.extend_from_slice(&value.to_le_bytes());
3094 }
3095 }
3096 (LogicalType::Boolean, Data::Bool(values)) => {
3097 for value in &**values {
3098 out.push(u8::from(*value));
3099 }
3100 }
3101 (LogicalType::Varchar, Data::Varlen(values)) => {
3102 let mut bytes = Vec::new();
3103 put_u32(&mut out, 0);
3104 for row in 0..vector.len() {
3105 let value = values.bytes(row).ok_or_else(|| invalid("string view is invalid"))?;
3106 bytes.extend_from_slice(value);
3107 put_u32(
3108 &mut out,
3109 u32::try_from(bytes.len())
3110 .map_err(|_| invalid("string payload exceeds 4GiB"))?,
3111 );
3112 }
3113 out.extend_from_slice(&bytes);
3114 }
3115 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
3116 }
3117 Ok((out, membership))
3118}
3119
3120fn put_varint(out: &mut Vec<u8>, mut value: u32) {
3121 while value >= 0x80 {
3122 out.push((value as u8 & 0x7f) | 0x80);
3123 value >>= 7;
3124 }
3125 out.push(value as u8);
3126}
3127
3128fn unique_codes(codes: &[u32]) -> Vec<u32> {
3130 let mut unique = codes.to_vec();
3131 unique.sort_unstable();
3132 unique.dedup();
3133 unique
3134}
3135
3136fn merged_codes(lists: Vec<Vec<u32>>) -> Vec<u32> {
3142 let mut lists = lists;
3143 while lists.len() > 1 {
3144 let mut next = Vec::with_capacity(lists.len().div_ceil(2));
3145 for pair in lists.chunks(2) {
3146 match pair {
3147 [left, right] => next.push(merged_pair(left, right)),
3148 [only] => next.push(only.clone()),
3149 _ => {}
3150 }
3151 }
3152 lists = next;
3153 }
3154 lists.pop().unwrap_or_default()
3155}
3156
3157fn merged_pair(left: &[u32], right: &[u32]) -> Vec<u32> {
3158 let mut out = Vec::with_capacity(left.len().saturating_add(right.len()));
3159 let mut at = 0;
3160 let mut to = 0;
3161 while at < left.len() && to < right.len() {
3162 match left[at].cmp(&right[to]) {
3163 Ordering::Less => {
3164 out.push(left[at]);
3165 at += 1;
3166 }
3167 Ordering::Greater => {
3168 out.push(right[to]);
3169 to += 1;
3170 }
3171 Ordering::Equal => {
3172 out.push(left[at]);
3173 at += 1;
3174 to += 1;
3175 }
3176 }
3177 }
3178 out.extend_from_slice(&left[at..]);
3179 out.extend_from_slice(&right[to..]);
3180 out
3181}
3182
3183fn merged_range(ranges: impl Iterator<Item = Range>) -> Range {
3188 let mut merged = Range::default();
3189 let mut first = true;
3190 for range in ranges {
3191 merged.nulls = merged.nulls.saturating_add(range.nulls);
3192 merged.sum = match (merged.sum.take(), range.sum) {
3196 (Some(held), Some(next)) if !first => held.checked_add(next),
3197 (_, next) if first => next,
3198 _ => None,
3199 };
3200 merged.exact = if first { range.exact } else { merged.exact && range.exact };
3201 if first {
3202 merged.low = range.low;
3203 merged.high = range.high;
3204 first = false;
3205 continue;
3206 }
3207 merged.low = match (merged.low.take(), range.low) {
3208 (Some(held), Some(next)) => Some(held.smaller(next)),
3209 _ => None,
3210 };
3211 merged.high = match (merged.high.take(), range.high) {
3212 (Some(held), Some(next)) => Some(held.larger(next)),
3213 _ => None,
3214 };
3215 }
3216 merged
3217}
3218
3219fn encode_sieves<'a>(sieves: impl Iterator<Item = &'a Option<Sieve>>) -> Result<Vec<u8>> {
3225 let held: Vec<&Option<Sieve>> = sieves.collect();
3226 let mut out = Vec::new();
3227 put_u32(
3228 &mut out,
3229 u32::try_from(held.len()).map_err(|_| invalid("too many parts in a stripe"))?,
3230 );
3231 for sieve in &held {
3232 let length = sieve.as_ref().map_or(0, Sieve::len);
3233 put_u32(&mut out, u32::try_from(length).map_err(|_| invalid("sieve length overflow"))?);
3234 }
3235 for sieve in held.into_iter().flatten() {
3237 out.extend_from_slice(&sieve.to_bytes());
3238 }
3239 Ok(out)
3240}
3241
3242fn decode_sieves(bytes: &[u8]) -> Result<Vec<Option<Sieve>>> {
3248 let parts = u32::from_le_bytes(
3249 bytes
3250 .get(..4)
3251 .ok_or_else(|| invalid("sieve page is truncated"))?
3252 .try_into()
3253 .map_err(|_| invalid("sieve page is truncated"))?,
3254 ) as usize;
3255 let mut lengths = Vec::with_capacity(parts);
3256 for part in 0..parts {
3257 let at = 4 + part * 4;
3258 let field = bytes.get(at..at + 4).ok_or_else(|| invalid("sieve page is truncated"))?;
3259 lengths.push(u32::from_le_bytes(
3260 field.try_into().map_err(|_| invalid("sieve page is truncated"))?,
3261 ) as usize);
3262 }
3263 let mut at = 4 + parts * 4;
3264 let mut out = Vec::with_capacity(parts);
3265 for length in lengths {
3266 if length == 0 {
3267 out.push(None);
3268 continue;
3269 }
3270 let end = at.checked_add(length).ok_or_else(|| invalid("sieve page is truncated"))?;
3271 let field = bytes.get(at..end).ok_or_else(|| invalid("sieve page is truncated"))?;
3272 out.push(Sieve::from_bytes(field));
3273 at = end;
3274 }
3275 if at != bytes.len() {
3276 return Err(invalid("sieve page has trailing bytes"));
3277 }
3278 Ok(out)
3279}
3280
3281fn encode_membership(unique: &[u32]) -> Vec<u8> {
3287 let mut out = Vec::with_capacity(unique.len().saturating_mul(2).saturating_add(5));
3288 put_varint(&mut out, u32::try_from(unique.len()).unwrap_or(u32::MAX));
3289 let mut previous = 0;
3290 for (at, &code) in unique.iter().enumerate() {
3291 put_varint(&mut out, if at == 0 { code } else { code - previous });
3292 previous = code;
3293 }
3294 out
3295}
3296
3297fn take_varint(bytes: &[u8], at: &mut usize) -> Result<u32> {
3298 let mut value = 0_u32;
3299 for shift in (0..35).step_by(7) {
3300 let byte = *bytes.get(*at).ok_or_else(|| invalid("membership varint is truncated"))?;
3301 *at += 1;
3302 let part = u32::from(byte & 0x7f);
3303 if shift == 28 && part > 0x0f {
3304 return Err(invalid("membership varint overflow"));
3305 }
3306 value = value
3307 .checked_add(
3308 part.checked_shl(shift).ok_or_else(|| invalid("membership varint overflow"))?,
3309 )
3310 .ok_or_else(|| invalid("membership varint overflow"))?;
3311 if byte & 0x80 == 0 {
3312 return Ok(value);
3313 }
3314 }
3315 Err(invalid("membership varint is too long"))
3316}
3317
3318fn decode_membership(bytes: &[u8]) -> Result<Vec<u32>> {
3319 let mut at = 0;
3320 let count = take_varint(bytes, &mut at)? as usize;
3321 let mut codes = Vec::with_capacity(count);
3322 let mut previous = 0_u32;
3323 for index in 0..count {
3324 let delta = take_varint(bytes, &mut at)?;
3325 let code = if index == 0 {
3326 delta
3327 } else {
3328 previous.checked_add(delta).ok_or_else(|| invalid("membership code overflow"))?
3329 };
3330 if index > 0 && code <= previous {
3331 return Err(invalid("membership codes are not increasing"));
3332 }
3333 codes.push(code);
3334 previous = code;
3335 }
3336 if at != bytes.len() {
3337 return Err(invalid("membership page has trailing bytes"));
3338 }
3339 Ok(codes)
3340}
3341
3342fn string_dictionary(vector: &Vector) -> Result<Option<Vec<u8>>> {
3343 let mut by_text = HashMap::new();
3344 let mut values = Vec::new();
3345 let mut codes = Vec::with_capacity(vector.len());
3346 let mut plain_bytes = 0_usize;
3347 for row in 0..vector.len() {
3348 let text = vector.text_at(row).unwrap_or("");
3349 plain_bytes = plain_bytes.saturating_add(text.len());
3350 let code = match by_text.get(text) {
3351 Some(&code) => code,
3352 None => {
3353 let code = u32::try_from(values.len())
3354 .map_err(|_| invalid("too many dictionary values"))?;
3355 by_text.insert(text, code);
3356 values.push(text);
3357 code
3358 }
3359 };
3360 codes.push(code);
3361 }
3362 let dictionary_bytes = values.iter().map(|value| value.len()).sum::<usize>();
3363 let encoded = 8_usize
3364 .saturating_add((values.len() + 1).saturating_mul(4))
3365 .saturating_add(dictionary_bytes)
3366 .saturating_add(codes.len().saturating_mul(4));
3367 let plain = (vector.len() + 1).saturating_mul(4).saturating_add(plain_bytes);
3368 if encoded >= plain {
3369 return Ok(None);
3370 }
3371 let mut out = Vec::with_capacity(encoded);
3372 put_u32(
3373 &mut out,
3374 u32::try_from(values.len()).map_err(|_| invalid("too many dictionary values"))?,
3375 );
3376 put_u32(
3377 &mut out,
3378 u32::try_from(dictionary_bytes).map_err(|_| invalid("dictionary payload exceeds 4GiB"))?,
3379 );
3380 let mut offset = 0_u32;
3381 put_u32(&mut out, offset);
3382 for value in &values {
3383 offset = offset
3384 .checked_add(
3385 u32::try_from(value.len()).map_err(|_| invalid("dictionary value is too long"))?,
3386 )
3387 .ok_or_else(|| invalid("dictionary payload exceeds 4GiB"))?;
3388 put_u32(&mut out, offset);
3389 }
3390 for value in values {
3391 out.extend_from_slice(value.as_bytes());
3392 }
3393 for code in codes {
3394 put_u32(&mut out, code);
3395 }
3396 Ok(Some(out))
3397}
3398
3399struct EncodedDictionary {
3400 index: Vec<u8>,
3401 ranks: Vec<u8>,
3402 payload: Vec<u8>,
3403}
3404
3405fn head(bytes: &[u8]) -> u64 {
3407 let mut word = [0; 8];
3408 let take = bytes.len().min(8);
3409 word[..take].copy_from_slice(&bytes[..take]);
3410 u64::from_be_bytes(word)
3411}
3412
3413fn rankings(dictionaries: &[Option<GlobalDictionary>]) -> Result<Vec<Vec<(u64, u32)>>> {
3421 let present =
3422 dictionaries.iter().enumerate().filter(|(_, held)| held.is_some()).map(|(at, _)| at);
3423 let present = present.collect::<Vec<_>>();
3424 let mut orders = vec![Vec::new(); dictionaries.len()];
3425 let workers = std::thread::available_parallelism()
3426 .map_or(1, usize::from)
3427 .min(MAX_FREQUENCY_WORKERS)
3428 .min(present.len());
3429 if workers <= 1 {
3430 for at in present {
3431 if let Some(dictionary) = &dictionaries[at] {
3432 orders[at] = dictionary.ranked();
3433 }
3434 }
3435 return Ok(orders);
3436 }
3437 let width = present.len().div_ceil(workers);
3438 let pieces = std::thread::scope(|scope| {
3439 present
3440 .chunks(width)
3441 .map(|columns| {
3442 scope.spawn(|| {
3443 columns
3444 .iter()
3445 .filter_map(|&at| dictionaries[at].as_ref().map(|held| (at, held.ranked())))
3446 .collect::<Vec<_>>()
3447 })
3448 })
3449 .collect::<Vec<_>>()
3450 .into_iter()
3451 .map(|handle| {
3452 handle.join().map_err(|_| Error::internal("a dictionary sort worker panicked"))
3453 })
3454 .collect::<Result<Vec<_>>>()
3455 })?;
3456 for piece in pieces {
3457 for (at, order) in piece {
3458 orders[at] = order;
3459 }
3460 }
3461 Ok(orders)
3462}
3463
3464fn encode_global_dictionary(
3465 dictionary: GlobalDictionary,
3466 order: &[(u64, u32)],
3467) -> Result<EncodedDictionary> {
3468 let values = dictionary.offsets.len() - 1;
3469 if order.len() != values {
3470 return Err(invalid("global dictionary order does not cover its values"));
3471 }
3472 let payload_len = dictionary.payload.len();
3473 let blocks = payload_len.div_ceil(TEXT_PAYLOAD_BLOCK);
3474 let ranks = encode_ranks(order);
3475 let rank_blocks = values.div_ceil(TEXT_RANK_BLOCK);
3476 let mut index = Vec::with_capacity(12 + (values + 1) * 4 + (blocks + rank_blocks) * 8);
3477 put_u32(
3478 &mut index,
3479 u32::try_from(values).map_err(|_| invalid("global dictionary has too many values"))?,
3480 );
3481 put_u32(&mut index, TEXT_PAYLOAD_BLOCK as u32);
3482 put_u32(
3483 &mut index,
3484 u32::try_from(blocks).map_err(|_| invalid("global dictionary has too many blocks"))?,
3485 );
3486 for offset in dictionary.offsets {
3487 put_u32(&mut index, offset);
3488 }
3489 for block in dictionary.payload.chunks(TEXT_PAYLOAD_BLOCK) {
3490 put_u64(&mut index, checksum(block));
3491 }
3492 for block in ranks.chunks(TEXT_RANK_BLOCK * RANK_ENTRY) {
3493 put_u64(&mut index, checksum(block));
3494 }
3495 Ok(EncodedDictionary { index, ranks, payload: dictionary.payload })
3496}
3497
3498fn encode_ranks(order: &[(u64, u32)]) -> Vec<u8> {
3505 let mut out = Vec::with_capacity(order.len() * RANK_ENTRY);
3506 for block in order.chunks(TEXT_RANK_BLOCK) {
3507 for &(head, _) in block {
3508 put_u64(&mut out, head);
3509 }
3510 for &(_, code) in block {
3511 put_u32(&mut out, code);
3512 }
3513 }
3514 out
3515}
3516
3517fn open_global_dictionary(file: Arc<File>, page: Page, ty: &LogicalType) -> Result<Vector> {
3518 if ty != &LogicalType::Varchar {
3519 return Err(invalid("global dictionary belongs to a non-string column"));
3520 }
3521 let mut header = [0; 12];
3522 read_at(&file, page.offset, &mut header)?;
3523 let count = u32::from_le_bytes(header[0..4].try_into().expect("four bytes")) as usize;
3524 let block_size = u32::from_le_bytes(header[4..8].try_into().expect("four bytes")) as usize;
3525 let blocks = u32::from_le_bytes(header[8..12].try_into().expect("four bytes")) as usize;
3526 if block_size != TEXT_PAYLOAD_BLOCK {
3527 return Err(invalid("global dictionary block width differs"));
3528 }
3529 let offset_len = (count + 1)
3530 .checked_mul(4)
3531 .ok_or_else(|| invalid("global dictionary offset count overflow"))?;
3532 let ranks = count;
3537 let rank_blocks = ranks.div_ceil(TEXT_RANK_BLOCK);
3538 let rank_len =
3539 ranks.checked_mul(RANK_ENTRY).ok_or_else(|| invalid("global dictionary rank overflow"))?;
3540 let hash_len = blocks
3541 .checked_add(rank_blocks)
3542 .and_then(|count| count.checked_mul(8))
3543 .ok_or_else(|| invalid("global dictionary block count overflow"))?;
3544 let index_len = 12usize
3545 .checked_add(offset_len)
3546 .and_then(|len| len.checked_add(hash_len))
3547 .ok_or_else(|| invalid("global dictionary header overflow"))?;
3548 let body_len = index_len
3549 .checked_add(rank_len)
3550 .ok_or_else(|| invalid("global dictionary header overflow"))?;
3551 if body_len > page.length as usize {
3552 return Err(invalid("global dictionary offset index exceeds its page"));
3553 }
3554 let mut index = vec![0; index_len];
3555 index[..12].copy_from_slice(&header);
3556 read_at(&file, page.offset + 12, &mut index[12..])?;
3557 if checksum(&index) != page.hash {
3558 return Err(invalid("global dictionary index checksum differs"));
3559 }
3560 let offsets = index[12..12 + offset_len]
3561 .chunks_exact(4)
3562 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
3563 .collect::<Vec<_>>();
3564 let mut hashes = index[12 + offset_len..]
3565 .chunks_exact(8)
3566 .map(|part| u64::from_le_bytes(part.try_into().expect("eight bytes")))
3567 .collect::<Vec<_>>();
3568 let rank_hashes = hashes.split_off(blocks);
3569 let payload_len = page.length as usize - body_len;
3570 if blocks != payload_len.div_ceil(TEXT_PAYLOAD_BLOCK) {
3571 return Err(invalid("global dictionary block count differs from its payload"));
3572 }
3573 if offsets.first() != Some(&0)
3574 || offsets.last().copied().map(|last| last as usize) != Some(payload_len)
3575 || offsets.windows(2).any(|pair| pair[0] > pair[1])
3576 {
3577 return Err(invalid("global dictionary offsets do not bound the payload"));
3578 }
3579 let payload_extents = (0..payload_len.div_ceil(TEXT_PAYLOAD_BLOCK * TEXT_PAYLOAD_EXTENT))
3580 .map(|_| OnceLock::new())
3581 .collect();
3582 let crossing = (0..count.div_ceil(TEXT_CROSSING_BLOCK)).map(|_| OnceLock::new()).collect();
3583 Vector::external_text(
3584 LogicalType::Varchar,
3585 Arc::new(NativeText {
3586 file,
3587 offsets,
3588 ranks,
3589 rank_at: page.offset + index_len as u64,
3590 rank_hashes,
3591 rank_blocks: (0..rank_blocks).map(|_| OnceLock::new()).collect(),
3592 payload: page.offset + body_len as u64,
3593 payload_len,
3594 hashes,
3595 payload_extents,
3596 crossing,
3597 }),
3598 )
3599}
3600
3601fn decode(
3602 ty: &LogicalType,
3603 rows: usize,
3604 bytes: &[u8],
3605 global: Option<Arc<Vector>>,
3606) -> Result<Vector> {
3607 let mut cur = Cursor { bytes, at: 0 };
3608 let codec = cur.u8()?;
3609 let flag = cur.u8()?;
3610 let validity = match flag {
3611 0 => Validity::AllValid,
3612 1 => Validity::AllInvalid,
3613 2 => {
3614 let mask = cur.take(rows.div_ceil(8))?;
3615 Validity::from_iter(rows, |row| mask[row / 8] >> (row % 8) & 1 == 1)
3616 }
3617 _ => return Err(invalid("page validity tag differs")),
3618 };
3619 if codec == 1 {
3620 if ty != &LogicalType::Varchar {
3621 return Err(invalid("dictionary codec belongs to a non-string page"));
3622 }
3623 let count = cur.u32()? as usize;
3624 let payload_len = cur.u32()? as usize;
3625 let offset_bytes = cur.take(
3626 (count + 1)
3627 .checked_mul(4)
3628 .ok_or_else(|| invalid("dictionary offset count overflow"))?,
3629 )?;
3630 let offsets = offset_bytes
3631 .chunks_exact(4)
3632 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
3633 .collect::<Vec<_>>();
3634 let payload = cur.take(payload_len)?.to_vec();
3635 if offsets.first() != Some(&0)
3636 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
3637 || offsets.windows(2).any(|pair| pair[0] > pair[1])
3638 {
3639 return Err(invalid("dictionary offsets do not bound the payload"));
3640 }
3641 let mut strings = StringColumn::over(Buffer::from_vec(payload));
3642 for pair in offsets.windows(2) {
3643 strings.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
3644 }
3645 let mut codes = Vec::with_capacity(rows);
3646 for _ in 0..rows {
3647 codes.push(cur.u32()?);
3648 }
3649 if codes.iter().any(|code| *code as usize >= count) {
3650 return Err(invalid("dictionary code is out of range"));
3651 }
3652 if cur.at != bytes.len() {
3653 return Err(invalid("dictionary page has trailing bytes"));
3654 }
3655 let dictionary = Vector::flat(LogicalType::Varchar, Data::Varlen(strings))?;
3656 return Ok(Vector::dictionary(codes, dictionary)?.with_validity(validity));
3657 }
3658 if codec == 3 || codec == 4 {
3659 let dictionary = global.ok_or_else(|| invalid("global code page has no dictionary"))?;
3660 let codes = if codec == 4 {
3661 let wide = integer::decode(&bytes[cur.at..])?;
3664 if wide.len() != rows {
3665 return Err(invalid("encoded code page holds the wrong number of rows"));
3666 }
3667 wide.into_iter()
3668 .map(|code| u32::try_from(code).map_err(|_| invalid("code is not a code")))
3669 .collect::<Result<Vec<u32>>>()?
3670 } else {
3671 let mut codes = Vec::with_capacity(rows);
3672 for _ in 0..rows {
3673 codes.push(cur.u32()?);
3674 }
3675 if cur.at != bytes.len() {
3676 return Err(invalid("global code page has trailing bytes"));
3677 }
3678 codes
3679 };
3680 let highest = codes.iter().copied().max();
3681 return Ok(Vector::stable_dictionary_validated(codes, dictionary, highest)?
3682 .with_validity(validity));
3683 }
3684 if codec == 2 {
3685 let width = u32::from(cur.u8()?);
3686 let base = i128::from_le_bytes(cur.take(16)?.try_into().expect("sixteen bytes"));
3687 let count = cur.u32()? as usize;
3688 let mut words = Vec::with_capacity(count);
3689 for _ in 0..count {
3690 words.push(cur.u64()?);
3691 }
3692 if cur.at != bytes.len() {
3693 return Err(invalid("packed page has trailing bytes"));
3694 }
3695 return Ok(Vector::packed(ty.clone(), words, width, base, rows)?.with_validity(validity));
3696 }
3697 if codec != 0 {
3698 return Err(invalid("page codec is unknown"));
3699 }
3700 let data = match ty {
3701 LogicalType::TinyInt => {
3702 let values = cur.take(rows)?;
3703 Data::Int8(values.iter().map(|item| *item as i8).collect::<Vec<_>>().into())
3704 }
3705 LogicalType::UTinyInt => Data::UInt8(cur.take(rows)?.to_vec().into()),
3706 LogicalType::SmallInt => {
3707 let values =
3708 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
3709 Data::Int16(
3710 values
3711 .chunks_exact(2)
3712 .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
3713 .collect::<Vec<_>>()
3714 .into(),
3715 )
3716 }
3717 LogicalType::USmallInt => {
3718 let values =
3719 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
3720 Data::UInt16(
3721 values
3722 .chunks_exact(2)
3723 .map(|item| u16::from_le_bytes(item.try_into().expect("two bytes")))
3724 .collect::<Vec<_>>()
3725 .into(),
3726 )
3727 }
3728 LogicalType::UInteger => {
3729 let values =
3730 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
3731 Data::UInt32(
3732 values
3733 .chunks_exact(4)
3734 .map(|item| u32::from_le_bytes(item.try_into().expect("four bytes")))
3735 .collect::<Vec<_>>()
3736 .into(),
3737 )
3738 }
3739 LogicalType::UBigInt => {
3740 let values =
3741 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
3742 Data::UInt64(
3743 values
3744 .chunks_exact(8)
3745 .map(|item| u64::from_le_bytes(item.try_into().expect("eight bytes")))
3746 .collect::<Vec<_>>()
3747 .into(),
3748 )
3749 }
3750 LogicalType::Integer | LogicalType::Date => {
3751 let values =
3752 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
3753 Data::Int32(
3754 values
3755 .chunks_exact(4)
3756 .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
3757 .collect::<Vec<_>>()
3758 .into(),
3759 )
3760 }
3761 LogicalType::BigInt | LogicalType::Timestamp => {
3762 let values =
3763 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
3764 Data::Int64(
3765 values
3766 .chunks_exact(8)
3767 .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
3768 .collect::<Vec<_>>()
3769 .into(),
3770 )
3771 }
3772 LogicalType::Boolean => {
3773 let values = cur.take(rows)?;
3774 if values.iter().any(|value| *value > 1) {
3775 return Err(invalid("boolean page has another value"));
3776 }
3777 Data::Bool(values.iter().map(|value| *value == 1).collect::<Vec<_>>().into())
3778 }
3779 LogicalType::Varchar => {
3780 let offset_bytes = cur
3781 .take((rows + 1).checked_mul(4).ok_or_else(|| invalid("offset count overflow"))?)?;
3782 let offsets = offset_bytes
3783 .chunks_exact(4)
3784 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
3785 .collect::<Vec<_>>();
3786 let payload = cur.take(bytes.len() - cur.at)?.to_vec();
3787 if offsets.first() != Some(&0)
3788 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
3789 || offsets.windows(2).any(|pair| pair[0] > pair[1])
3790 {
3791 return Err(invalid("string offsets do not bound the payload"));
3792 }
3793 let mut values = StringColumn::over(Buffer::from_vec(payload));
3794 for pair in offsets.windows(2) {
3795 values.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
3796 }
3797 Data::Varlen(values)
3798 }
3799 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
3800 };
3801 if cur.at != bytes.len() {
3802 return Err(invalid("page has trailing bytes"));
3803 }
3804 Ok(Vector::flat(ty.clone(), data)?.with_validity(validity))
3805}
3806
3807#[cfg(test)]
3808mod tests {
3809 use std::fs;
3810 use std::io::{Seek, SeekFrom, Write};
3811 use std::path::PathBuf;
3812 use std::time::{SystemTime, UNIX_EPOCH};
3813
3814 use rudb_common::Value;
3815 use rudb_common::bounds::Op;
3816
3817 use super::*;
3818
3819 #[test]
3820 fn checksum_matches_fixed_vectors() {
3821 assert_eq!(checksum(b""), 0xef46_db37_51d8_e999);
3822 assert_eq!(checksum(b"a"), 0xd24e_c4f1_a98c_6e5b);
3823 assert_eq!(checksum(b"abc"), 0x44bc_2cf5_ad77_0999);
3824 }
3825
3826 fn path(label: &str) -> PathBuf {
3827 let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
3828 std::env::temp_dir().join(format!("rudb-native-{label}-{}-{stamp}.rdb", std::process::id()))
3829 }
3830
3831 #[test]
3833 fn a_read_at_an_offset_ignores_where_another_thread_left_the_cursor() {
3834 const SPANS: usize = 64;
3835 const SPAN: usize = 512;
3836 let path = path("positional");
3837 let content: Vec<u8> =
3838 (0..SPANS).flat_map(|span| std::iter::repeat_n(span as u8, SPAN)).collect();
3839 fs::write(&path, &content).expect("the file is written");
3840 let file = Arc::new(File::open(&path).expect("the file opens"));
3841 std::thread::scope(|scope| {
3842 for _ in 0..8 {
3843 let file = Arc::clone(&file);
3844 scope.spawn(move || {
3845 for _ in 0..64 {
3846 for span in 0..SPANS {
3847 let mut bytes = [0_u8; SPAN];
3848 read_at(&file, (span * SPAN) as u64, &mut bytes)
3849 .expect("the span reads");
3850 assert!(
3851 bytes.iter().all(|byte| *byte == span as u8),
3852 "span {span} came back as {}",
3853 bytes[0],
3854 );
3855 }
3856 }
3857 });
3858 }
3859 });
3860 let mut past = [0_u8; SPAN];
3861 let end = (SPANS * SPAN) as u64;
3862 let error = read_at(&file, end, &mut past).expect_err("a read past the end is refused");
3863 assert!(error.message().contains("ends before its declared length"), "{error}");
3864 drop(file);
3865 let _ = fs::remove_file(&path);
3866 }
3867
3868 #[test]
3874 fn a_writer_puts_a_page_where_it_said_it_did_wherever_the_cursor_has_got_to() {
3875 let path = path("cursor");
3876 let mut writer = Writer::create(
3877 &path,
3878 "items",
3879 vec![
3880 Field::required("id", LogicalType::Integer),
3881 Field::new("text", LogicalType::Varchar),
3882 ],
3883 )
3884 .expect("new file");
3885 writer.append(&sample()).expect("first part");
3886 writer.file.seek(SeekFrom::Start(0)).expect("the cursor goes back to the header");
3887 writer.append(&sample()).expect("second part");
3888 writer.file.seek(SeekFrom::Start(1)).expect("and somewhere useless again");
3889 writer.finish().expect("commit");
3890 let reader = Reader::open(&path).expect("reopen from disk");
3891 assert_eq!(reader.table().rows(), 6);
3892 let ids = reader.read(0, &[0]).expect("the integer page reads back");
3893 assert_eq!(ids.value_at(0, 0), Value::Integer(4));
3894 assert_eq!(ids.value_at(2, 0), Value::Integer(-2));
3895 let text = reader.read(1, &[1]).expect("the text page reads back");
3896 assert_eq!(text.value_at(1, 0), Value::Null);
3897 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
3898 let end = reader.table().stripes().iter().flat_map(|stripe| {
3901 stripe
3902 .pages
3903 .iter()
3904 .map(|page| page.offset + u64::from(page.length))
3905 .chain(std::iter::once(stripe.index.offset + u64::from(stripe.index.length)))
3906 });
3907 let last = end.fold(HEADER, u64::max);
3908 let directory = fs::metadata(&path).expect("the file is there").len();
3909 assert!(last <= directory, "a page runs to {last} in a file of {directory} bytes");
3910 fs::remove_file(path).expect("remove scratch file");
3911 }
3912
3913 fn sample() -> Chunk {
3914 Chunk::new(vec![
3915 Vector::from_values(
3916 LogicalType::Integer,
3917 &[Value::Integer(4), Value::Integer(9), Value::Integer(-2)],
3918 )
3919 .expect("integers"),
3920 Vector::from_values(
3921 LogicalType::Varchar,
3922 &[
3923 Value::Varchar("alpha".into()),
3924 Value::Null,
3925 Value::Varchar("long text after a slash".into()),
3926 ],
3927 )
3928 .expect("strings"),
3929 ])
3930 .expect("matching rows")
3931 }
3932
3933 fn sample_ids() -> Chunk {
3934 Chunk::new(vec![
3935 Vector::flat(LogicalType::Integer, Data::Int32(vec![7, 8, 9].into()))
3936 .expect("integers"),
3937 ])
3938 .expect("one column")
3939 }
3940
3941 #[test]
3942 fn committed_file_reopens_and_reads_only_requested_columns() {
3943 let path = path("reopen");
3944 let mut writer = Writer::create(
3945 &path,
3946 "items",
3947 vec![
3948 Field::required("id", LogicalType::Integer),
3949 Field::new("text", LogicalType::Varchar),
3950 ],
3951 )
3952 .expect("new file");
3953 writer.append(&sample()).expect("first part");
3954 writer.append(&sample()).expect("second part");
3955 writer.finish().expect("commit");
3956 let reader = Reader::open(&path).expect("reopen from disk");
3957 assert_eq!(reader.table().rows(), 6);
3958 assert_eq!(reader.table().stripes().len(), 1);
3961 assert_eq!(reader.parts(), 2);
3962 assert_eq!(reader.part_rows(0), 3);
3963 assert_eq!(reader.part_rows(1), 3);
3964 let text = reader.read(1, &[1]).expect("only text page");
3965 assert_eq!(text.width(), 1);
3966 assert_eq!(text.value_at(1, 0), Value::Null);
3967 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
3968 let sparse = reader.read_sparse(1, &[1]).expect("one part without its whole page");
3969 assert_eq!(sparse.width(), 1);
3970 assert_eq!(sparse.value_at(1, 0), Value::Null);
3971 assert_eq!(sparse.value_at(2, 0), Value::Varchar("long text after a slash".into()));
3972 assert!(!reader.skips_codes(0, 1, &[0]).expect("alpha is in the stripe"));
3973 assert!(!reader.skips_codes(0, 1, &[2]).expect("long text is in the stripe"));
3974 assert!(reader.skips_codes(0, 1, &[3]).expect("unknown code is absent"));
3975 let count = reader.read(0, &[]).expect("no page is needed for count");
3976 assert_eq!(count.len(), 3);
3977 assert!(reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }]));
3978 assert!(!reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(0) }]));
3979 let integers = reader.top_frequencies(0, 1).expect("valid integer synopsis").expect("kept");
3980 assert_eq!(
3981 integers,
3982 vec![(Value::Integer(-2), 2), (Value::Integer(4), 2), (Value::Integer(9), 2),]
3983 );
3984 let strings = reader.top_frequencies(1, 1).expect("valid string synopsis").expect("kept");
3985 assert_eq!(strings.len(), 3);
3986 assert!(strings.contains(&(Value::Null, 2)));
3987 assert!(strings.contains(&(Value::Varchar("alpha".into()), 2)));
3988 assert!(strings.contains(&(Value::Varchar("long text after a slash".into()), 2)));
3989 fs::remove_file(path).expect("remove scratch file");
3990 }
3991
3992 #[test]
3998 fn parts_past_the_stripe_bound_start_a_new_stripe() {
3999 let path = path("stripe-bound");
4000 let mut writer = Writer::create(
4001 &path,
4002 "items",
4003 vec![
4004 Field::required("id", LogicalType::Integer),
4005 Field::new("text", LogicalType::Varchar),
4006 ],
4007 )
4008 .expect("new file");
4009 let parts = STRIPE_PARTS * 2 + 3;
4010 for part in 0..parts {
4011 let id = part as i32;
4012 let chunk = Chunk::new(vec![
4013 Vector::from_values(
4014 LogicalType::Integer,
4015 &[Value::Integer(id), Value::Integer(-id)],
4016 )
4017 .expect("integers"),
4018 Vector::from_values(
4019 LogicalType::Varchar,
4020 &[Value::Varchar(format!("value {part}")), Value::Null],
4021 )
4022 .expect("strings"),
4023 ])
4024 .expect("matching rows");
4025 writer.append(&chunk).expect("one part");
4026 }
4027 writer.finish().expect("commit");
4028
4029 let reader = Reader::open(&path).expect("reopen from disk");
4030 assert_eq!(reader.parts(), parts);
4031 assert_eq!(reader.table().rows(), parts * 2);
4032 assert_eq!(reader.table().stripes().len(), parts.div_ceil(STRIPE_PARTS));
4033 assert_eq!(reader.table().stripes()[0].parts(), STRIPE_PARTS);
4034 assert_eq!(reader.table().stripes()[0].rows(), STRIPE_PARTS * 2);
4035 assert_eq!(reader.table().stripes()[2].parts(), 3);
4036 for part in (0..parts).rev() {
4039 let dense = reader.read(part, &[0, 1]).expect("a whole page read");
4040 let sparse = reader.read_sparse(part, &[0, 1]).expect("one part read");
4041 for chunk in [&dense, &sparse] {
4042 assert_eq!(chunk.len(), 2, "part {part} has its own row count");
4043 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4044 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
4045 assert_eq!(chunk.value_at(0, 1), Value::Varchar(format!("value {part}")));
4046 assert_eq!(chunk.value_at(1, 1), Value::Null);
4047 }
4048 }
4049 let above = [Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }];
4052 assert!(reader.skips(0, &above), "the first stripe stops at 63");
4053 assert!(!reader.skips(STRIPE_PARTS * 2, &above), "the third stripe reaches 130");
4054 fs::remove_file(path).expect("remove scratch file");
4055 }
4056
4057 fn scattered(n: i64) -> i64 {
4059 n.wrapping_mul(-7_046_029_254_386_353_131)
4060 }
4061
4062 #[test]
4068 fn a_part_is_skipped_when_its_sieve_does_not_hold_the_constant() {
4069 let path = path("sieve-skip");
4070 let mut writer =
4071 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
4072 .expect("new file");
4073 let parts = STRIPE_PARTS + 3;
4074 let per_part = 8;
4075 for part in 0..parts {
4076 let held: Vec<Value> = (0..per_part)
4077 .map(|row| Value::BigInt(scattered((part * per_part + row) as i64)))
4078 .collect();
4079 let chunk =
4080 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
4081 .expect("one column");
4082 writer.append(&chunk).expect("one part");
4083 }
4084 writer.finish().expect("commit");
4085
4086 let reader = Reader::open(&path).expect("reopen from disk");
4087 let probe = |value: i64| Probe {
4088 column: 0,
4089 op: Op::Equal,
4090 value: Bound::Int(i128::from(scattered(value))),
4091 };
4092 for wanted in [0_i64, 9, (parts * per_part - 1) as i64] {
4093 let tests = [probe(wanted)];
4094 let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &tests)).collect();
4095 let home = wanted as usize / per_part;
4096 assert_eq!(kept, vec![home], "only the part holding {wanted} is read");
4097 }
4098 let absent = [probe((parts * per_part) as i64 + 1)];
4099 assert!((0..parts).all(|part| reader.skips(part, &absent)), "no part holds it");
4100 let tests = [probe(0)];
4103 assert!(
4104 reader.table().stripes().iter().all(|stripe| !stripe.zone.skips(&tests)),
4105 "the bounds rule out no stripe at all"
4106 );
4107 fs::remove_file(path).expect("remove scratch file");
4108 }
4109
4110 #[test]
4116 fn a_damaged_sieve_page_is_read_through_rather_than_refused() {
4117 let path = path("sieve-damaged");
4118 let mut writer =
4119 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
4120 .expect("new file");
4121 let held: Vec<Value> = (0..8).map(|row| Value::BigInt(scattered(row))).collect();
4122 let chunk =
4123 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
4124 .expect("one column");
4125 writer.append(&chunk).expect("one part");
4126 writer.finish().expect("commit");
4127
4128 let page =
4129 Reader::open(&path).expect("reopen").table.stripes[0].sieves[0].expect("a sieve page");
4130 let mut file = OpenOptions::new().write(true).open(&path).expect("open the sieve page");
4131 file.seek(SeekFrom::Start(page.offset + u64::from(page.length) - 1)).expect("seek");
4132 file.write_all(&[0xff]).expect("damage one byte");
4133 drop(file);
4134
4135 let reader = Reader::open(&path).expect("reopen the damaged file");
4136 let absent =
4137 [Probe { column: 0, op: Op::Equal, value: Bound::Int(i128::from(scattered(99))) }];
4138 assert!(!reader.skips(0, &absent), "a sieve that cannot be read skips nothing");
4139 assert_eq!(reader.read(0, &[0]).expect("the rows are untouched").len(), 8);
4140 fs::remove_file(path).expect("remove scratch file");
4141 }
4142
4143 #[test]
4154 fn workers_that_want_the_same_stripe_read_it_once() {
4155 let path = path("single-flight");
4156 let mut writer =
4157 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4158 .expect("new file");
4159 for part in 0..STRIPE_PARTS {
4160 let id = part as i32;
4161 let chunk = Chunk::new(vec![
4162 Vector::from_values(
4163 LogicalType::Integer,
4164 &[Value::Integer(id), Value::Integer(-id)],
4165 )
4166 .expect("integers"),
4167 ])
4168 .expect("matching rows");
4169 writer.append(&chunk).expect("one part");
4170 }
4171 writer.finish().expect("commit");
4172
4173 let reader = Reader::open(&path).expect("reopen from disk");
4174 assert_eq!(reader.table().stripes().len(), 1, "one stripe is the point of the test");
4175 let barrier = std::sync::Barrier::new(8);
4176 std::thread::scope(|scope| {
4177 for worker in 0..8 {
4178 let reader = &reader;
4179 let barrier = &barrier;
4180 scope.spawn(move || {
4181 barrier.wait();
4182 for part in (worker..STRIPE_PARTS).step_by(8) {
4183 let chunk = reader.read(part, &[0]).expect("a whole page read");
4184 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4185 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
4186 }
4187 });
4188 }
4189 });
4190 assert_eq!(reader.pages.load(Atomic::Relaxed), 1, "one stripe, one page read, whoever won");
4191 fs::remove_file(path).expect("remove scratch file");
4192 }
4193
4194 #[test]
4202 fn an_index_is_read_once_per_stripe_however_often_the_page_is_evicted() {
4203 let path = path("index-cache");
4204 let mut writer =
4205 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4206 .expect("new file");
4207 let parts = STRIPE_PARTS * (CACHED_STRIPES_PER_COLUMN + 2);
4208 for part in 0..parts {
4209 let id = part as i32;
4210 let chunk = Chunk::new(vec![
4211 Vector::from_values(LogicalType::Integer, &[Value::Integer(id)]).expect("integers"),
4212 ])
4213 .expect("matching rows");
4214 writer.append(&chunk).expect("one part");
4215 }
4216 writer.finish().expect("commit");
4217
4218 let reader = Reader::open(&path).expect("reopen from disk");
4219 let stripes = reader.table().stripes().len();
4220 assert!(stripes > CACHED_STRIPES_PER_COLUMN, "the page cache has to be too small for this");
4221 for _ in 0..2 {
4223 for part in 0..parts {
4224 let chunk = reader.read(part, &[0]).expect("a part");
4225 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4226 }
4227 }
4228 assert_eq!(reader.indexes.load(Atomic::Relaxed), stripes, "one index read per stripe");
4229 assert!(
4230 reader.pages.load(Atomic::Relaxed) > stripes,
4231 "the pages are the ones that get read again, which is what makes the index count mean \
4232 something"
4233 );
4234 fs::remove_file(path).expect("remove scratch file");
4235 }
4236
4237 #[test]
4246 fn a_worker_per_stripe_reads_its_page_once_when_the_cache_was_told_to_expect_it() {
4247 let workers = CACHED_STRIPES_PER_COLUMN + 4;
4248 let path = path("stripe-per-worker");
4249 let mut writer =
4250 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4251 .expect("new file");
4252 for part in 0..STRIPE_PARTS * workers {
4253 let chunk = Chunk::new(vec![
4254 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
4255 .expect("integers"),
4256 ])
4257 .expect("matching rows");
4258 writer.append(&chunk).expect("one part");
4259 }
4260 writer.finish().expect("commit");
4261
4262 let read = |told: bool| {
4263 let reader = Reader::open(&path).expect("reopen from disk");
4264 assert_eq!(reader.table().stripes().len(), workers, "a stripe per worker");
4265 if told {
4266 reader.keep_stripes(workers);
4267 }
4268 let barrier = std::sync::Barrier::new(workers);
4269 std::thread::scope(|scope| {
4270 for (worker, run) in reader.stripe_parts().into_iter().enumerate() {
4271 let reader = &reader;
4272 let barrier = &barrier;
4273 scope.spawn(move || {
4274 for part in run {
4275 barrier.wait();
4276 let chunk = reader.read(part, &[0]).expect("a part of my own stripe");
4277 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4278 }
4279 assert!(worker < workers);
4280 });
4281 }
4282 });
4283 reader.pages.load(Atomic::Relaxed)
4284 };
4285
4286 assert_eq!(read(true), workers, "one page read per stripe and no more");
4287 assert!(read(false) > workers, "a cache that small is read again on every part");
4288 fs::remove_file(path).expect("remove scratch file");
4289 }
4290
4291 #[test]
4296 fn a_damaged_index_page_is_an_error() {
4297 let path = path("damaged-index");
4298 let mut writer =
4299 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4300 .expect("new file");
4301 writer.append(&sample_ids()).expect("first part");
4302 writer.append(&sample_ids()).expect("second part");
4303 writer.finish().expect("commit");
4304
4305 let reader = Reader::open(&path).expect("valid directory");
4306 let index = reader.table.stripes[0].index;
4307 let mut byte = [0; 1];
4308 read_at(&reader.file, index.offset, &mut byte).expect("the first part length");
4309 let mut file = OpenOptions::new().write(true).open(&path).expect("open index page");
4310 file.seek(SeekFrom::Start(index.offset)).expect("index start");
4311 file.write_all(&[!byte[0]]).expect("damage the first part length");
4312 let error = reader.read(1, &[0]).expect_err("a damaged index must not be used");
4313 assert!(error.message().contains("index page section checksum differs"), "{error}");
4314 fs::remove_file(path).expect("remove scratch file");
4315 }
4316
4317 #[test]
4324 fn every_integer_width_round_trips_through_a_page() {
4325 let path = path("integer-widths");
4326 let columns = [
4327 (LogicalType::TinyInt, vec![Value::TinyInt(i8::MIN), Value::TinyInt(i8::MAX)]),
4328 (LogicalType::UTinyInt, vec![Value::UTinyInt(0), Value::UTinyInt(u8::MAX)]),
4329 (LogicalType::SmallInt, vec![Value::SmallInt(i16::MIN), Value::SmallInt(i16::MAX)]),
4330 (LogicalType::USmallInt, vec![Value::USmallInt(0), Value::USmallInt(u16::MAX)]),
4331 (LogicalType::Integer, vec![Value::Integer(i32::MIN), Value::Integer(i32::MAX)]),
4332 (LogicalType::UInteger, vec![Value::UInteger(0), Value::UInteger(u32::MAX)]),
4333 (LogicalType::BigInt, vec![Value::BigInt(i64::MIN), Value::BigInt(i64::MAX)]),
4334 (LogicalType::UBigInt, vec![Value::UBigInt(0), Value::UBigInt(u64::MAX)]),
4335 ];
4336 let fields = columns
4337 .iter()
4338 .enumerate()
4339 .map(|(at, (ty, _))| Field::required(format!("c{at}"), ty.clone()))
4340 .collect::<Vec<_>>();
4341 let vectors = columns
4342 .iter()
4343 .map(|(ty, values)| Vector::from_values(ty.clone(), values).expect("a vector"))
4344 .collect::<Vec<_>>();
4345 let mut writer = Writer::create(&path, "widths", fields).expect("new file");
4346 writer.append(&Chunk::new(vectors).expect("matching rows")).expect("one stripe");
4347 writer.finish().expect("commit");
4348
4349 let reader = Reader::open(&path).expect("reopen from disk");
4350 let wanted = (0..columns.len()).collect::<Vec<_>>();
4351 let read = reader.read(0, &wanted).expect("every column");
4352 assert_eq!(read.len(), 2);
4353 for (at, (ty, values)) in columns.iter().enumerate() {
4355 assert_eq!(read.value_at(0, at), values[0], "the low end of {ty}");
4356 assert_eq!(read.value_at(1, at), values[1], "the high end of {ty}");
4357 }
4358 fs::remove_file(path).expect("remove scratch file");
4359 }
4360
4361 #[test]
4362 fn numeric_frequency_candidates_keep_bounded_row_ordinals() {
4363 let path = path("frequency-ordinals");
4364 let mut writer =
4365 Writer::create(&path, "items", vec![Field::required("id", LogicalType::BigInt)])
4366 .expect("new file");
4367 let mut values = Vec::new();
4368 for leader in 0..10_i64 {
4369 values.extend(std::iter::repeat_n(leader, 100));
4370 }
4371 values.extend(1_000_i64..41_000);
4372 for part in values.chunks(1_024) {
4373 let vector = Vector::flat(LogicalType::BigInt, Data::Int64(part.to_vec().into()))
4374 .expect("big integers");
4375 writer.append(&Chunk::new(vec![vector]).expect("one column")).expect("one stripe");
4376 }
4377 writer.finish().expect("commit");
4378
4379 let reader = Reader::open(&path).expect("reopen from disk");
4380 let occurrences =
4381 reader.frequency_occurrences(0).expect("valid metadata").expect("bounded ordinals");
4382 assert!(occurrences.omitted_max < 100);
4383 assert!(occurrences.ordinals.len() <= FREQUENCY_ORDINALS);
4384 assert!(occurrences.ordinals.windows(2).all(|pair| pair[0] < pair[1]));
4385 assert_eq!(&occurrences.ordinals[..1_000], &(0_u64..1_000).collect::<Vec<_>>());
4386 fs::remove_file(path).expect("remove scratch file");
4387 }
4388
4389 #[test]
4395 fn a_file_from_another_format_says_which_format_it_is() {
4396 let older = path("older-format");
4397 let mut writer =
4398 Writer::create(&older, "items", vec![Field::new("id", LogicalType::Integer)])
4399 .expect("new file");
4400 let chunk = Chunk::new(vec![
4401 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
4402 .expect("integers"),
4403 ])
4404 .expect("chunk");
4405 writer.append(&chunk).expect("page written");
4406 writer.finish().expect("commit");
4407
4408 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
4409 file.seek(SeekFrom::Start(8)).expect("the version follows the magic");
4410 file.write_all(&(FORMAT - 1).to_le_bytes()).expect("write an older version");
4411 drop(file);
4412 let complaint = Reader::open(&older).expect_err("an older format is refused").to_string();
4413 assert!(complaint.contains(&format!("format {}", FORMAT - 1)), "{complaint}");
4414 assert!(complaint.contains(&format!("format {FORMAT}")), "{complaint}");
4415
4416 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
4417 file.seek(SeekFrom::Start(0)).expect("the magic is first");
4418 file.write_all(b"NOTRUDB!").expect("write another engine's magic");
4419 drop(file);
4420 let complaint = Reader::open(&older).expect_err("a foreign file is refused").to_string();
4421 assert!(complaint.contains("magic"), "{complaint}");
4422 assert!(!complaint.contains("format"), "a version has nothing to do with it: {complaint}");
4423 fs::remove_file(older).expect("remove scratch file");
4424 }
4425
4426 #[test]
4427 fn an_unfinished_or_damaged_file_does_not_answer_with_partial_rows() {
4428 let unfinished = path("unfinished");
4429 let mut writer =
4430 Writer::create(&unfinished, "items", vec![Field::new("id", LogicalType::Integer)])
4431 .expect("new file");
4432 let chunk = Chunk::new(vec![
4433 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
4434 .expect("integers"),
4435 ])
4436 .expect("chunk");
4437 writer.append(&chunk).expect("page written");
4438 drop(writer);
4439 assert!(Reader::open(&unfinished).is_err(), "no directory was committed");
4440 fs::remove_file(unfinished).expect("remove scratch file");
4441
4442 let damaged = path("damaged");
4443 let mut writer =
4444 Writer::create(&damaged, "items", vec![Field::new("id", LogicalType::Integer)])
4445 .expect("new file");
4446 writer.append(&chunk).expect("page written");
4447 writer.finish().expect("commit");
4448 let reader = Reader::open(&damaged).expect("valid directory");
4449 let mut file =
4450 OpenOptions::new().write(true).open(&damaged).expect("open for a damaged page");
4451 file.seek(SeekFrom::Start(HEADER + 1)).expect("inside first page");
4452 file.write_all(&[255]).expect("damage one byte");
4453 assert!(reader.read(0, &[0]).is_err(), "page checksum rejects corruption");
4454 fs::remove_file(damaged).expect("remove scratch file");
4455 }
4456
4457 #[test]
4458 fn damaged_lazy_dictionary_payload_is_an_error() {
4459 let path = path("damaged-dictionary");
4460 let mut writer = Writer::create(
4461 &path,
4462 "items",
4463 vec![
4464 Field::required("id", LogicalType::Integer),
4465 Field::new("text", LogicalType::Varchar),
4466 ],
4467 )
4468 .expect("new file");
4469 writer.append(&sample()).expect("stripe written");
4470 writer.finish().expect("commit");
4471
4472 let reader = Reader::open(&path).expect("valid directory");
4473 let dictionary = reader.table.dictionaries[1].expect("string dictionary page");
4474 let mut header = [0; 12];
4477 read_at(&reader.file, dictionary.offset, &mut header).expect("dictionary header");
4478 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
4479 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
4480 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
4481 let index_len =
4482 12 + (count + 1) * 4 + (blocks + rank_blocks) * 8 + count * RANK_ENTRY as u64;
4483 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
4484 file.seek(SeekFrom::Start(dictionary.offset + index_len))
4485 .expect("inside dictionary payload");
4486 file.write_all(&[255]).expect("damage dictionary payload");
4487
4488 let chunk = reader.read(0, &[1]).expect("code page and dictionary index remain valid");
4489 let error =
4490 chunk.validate_external().expect_err("payload corruption must reach the caller");
4491 assert!(error.message().contains("payload checksum differs"), "{error}");
4492 fs::remove_file(path).expect("remove scratch file");
4493 }
4494
4495 #[test]
4502 fn a_dictionary_over_one_extent_checks_every_block_of_it() {
4503 let path = path("dictionary-extents");
4504 let value =
4505 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
4506 let parts = 30;
4507 let per_part = 1000;
4508 let mut writer =
4509 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
4510 .expect("new file");
4511 for part in 0..parts {
4512 let values = (0..per_part)
4513 .map(|row| Value::Varchar(value(part * per_part + row)))
4514 .collect::<Vec<_>>();
4515 let chunk = Chunk::new(vec![
4516 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
4517 ])
4518 .expect("matching rows");
4519 writer.append(&chunk).expect("a part");
4520 }
4521 writer.finish().expect("commit");
4522
4523 let reader = Reader::open(&path).expect("reopen from disk");
4524 let dictionary = reader.table.dictionaries[0].expect("string dictionary page");
4525 assert!(
4526 dictionary.length as usize > TEXT_PAYLOAD_BLOCK * TEXT_PAYLOAD_EXTENT,
4527 "the dictionary has to be over one extent for this to be testing anything"
4528 );
4529 for part in [0, parts - 1] {
4530 let chunk = reader.read(part, &[0]).expect("a part");
4531 chunk.validate_external().expect("every payload block checks out");
4532 assert_eq!(chunk.value_at(0, 0), Value::Varchar(value(part * per_part)));
4533 }
4534
4535 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
4536 file.seek(SeekFrom::Start(dictionary.offset + u64::from(dictionary.length) - 4))
4537 .expect("the last bytes of the page are payload");
4538 file.write_all(&[255]).expect("damage the last payload block");
4539 let reader = Reader::open(&path).expect("the directory and the index are untouched");
4540 let chunk = reader.read(parts - 1, &[0]).expect("the code page remains valid");
4541 let error = chunk.validate_external().expect_err("the damage must reach the caller");
4542 assert!(error.message().contains("payload checksum differs"), "{error}");
4543 fs::remove_file(path).expect("remove scratch file");
4544 }
4545
4546 #[test]
4551 fn a_damaged_sorted_order_is_an_error() {
4552 let path = path("damaged-order");
4553 let mut writer = Writer::create(
4554 &path,
4555 "items",
4556 vec![
4557 Field::required("id", LogicalType::Integer),
4558 Field::new("text", LogicalType::Varchar),
4559 ],
4560 )
4561 .expect("new file");
4562 writer.append(&sample()).expect("stripe written");
4563 writer.finish().expect("commit");
4564
4565 let reader = Reader::open(&path).expect("valid directory");
4566 let page = reader.table.dictionaries[1].expect("string dictionary page");
4567 let mut header = [0; 12];
4568 read_at(&reader.file, page.offset, &mut header).expect("dictionary header");
4569 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
4570 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
4571 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
4572 let index_len = 12 + (count + 1) * 4 + (blocks + rank_blocks) * 8;
4573 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
4574 file.seek(SeekFrom::Start(page.offset + index_len)).expect("the first head");
4575 file.write_all(&[255]).expect("damage the order");
4576
4577 let dictionary = reader.dictionary(1).expect("read").expect("a string column has one");
4578 let error = dictionary.compare_rank(0, b"anything").expect_err("a damaged order is caught");
4579 assert!(error.message().contains("rank checksum differs"), "{error}");
4580 fs::remove_file(path).expect("remove scratch file");
4581 }
4582
4583 #[test]
4587 fn a_global_dictionary_carries_the_sorted_order_of_its_values() {
4588 let spellings = ["overlong1z", "b", "", "overlong1a", "overlong", "ab", "a", "overlong1"];
4591 let path = path("dictionary-order");
4592 let mut writer =
4593 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
4594 .expect("new file");
4595 writer
4596 .append(
4597 &Chunk::new(vec![
4598 Vector::from_values(
4599 LogicalType::Varchar,
4600 &spellings.map(|text| Value::Varchar(text.into())),
4601 )
4602 .expect("strings"),
4603 ])
4604 .expect("one column"),
4605 )
4606 .expect("stripe written");
4607 writer.finish().expect("commit");
4608
4609 let reader = Reader::open(&path).expect("valid directory");
4610 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
4611 let count = dictionary.ranks().expect("a v10 file stores one");
4612 assert_eq!(count, spellings.len(), "every distinct value has a rank");
4613 let order = (0..count)
4614 .map(|rank| dictionary.code_at_rank(rank).expect("a code"))
4615 .collect::<Vec<_>>();
4616 let mut seen = order.clone();
4617 seen.sort_unstable();
4618 assert_eq!(seen, (0..spellings.len() as u32).collect::<Vec<_>>(), "a permutation of codes");
4619
4620 let ranked = order
4621 .iter()
4622 .map(|&code| {
4623 dictionary.try_bytes_at(code as usize).expect("read").expect("a value").to_vec()
4624 })
4625 .collect::<Vec<_>>();
4626 let mut expected = spellings.map(|text| text.as_bytes().to_vec()).to_vec();
4627 expected.sort();
4628 assert_eq!(ranked, expected, "rank order is value order");
4629
4630 for (rank, value) in expected.iter().enumerate() {
4633 assert_eq!(
4634 dictionary.compare_rank(rank, value).expect("compare"),
4635 Ordering::Equal,
4636 "rank {rank} is its own value"
4637 );
4638 if rank > 0 {
4639 assert_eq!(
4640 dictionary.compare_rank(rank - 1, value).expect("compare"),
4641 Ordering::Less,
4642 "rank {rank} follows the one before it"
4643 );
4644 }
4645 }
4646 fs::remove_file(path).expect("remove scratch file");
4647 }
4648
4649 #[test]
4650 fn damaged_membership_cannot_skip_a_string_page() {
4651 let path = path("damaged-membership");
4652 let mut writer = Writer::create(
4653 &path,
4654 "items",
4655 vec![
4656 Field::required("id", LogicalType::Integer),
4657 Field::new("text", LogicalType::Varchar),
4658 ],
4659 )
4660 .expect("new file");
4661 writer.append(&sample()).expect("stripe written");
4662 writer.finish().expect("commit");
4663
4664 let reader = Reader::open(&path).expect("valid directory");
4665 let membership = reader.table.stripes[0].memberships[1].expect("string membership");
4666 let mut file = OpenOptions::new().write(true).open(&path).expect("open membership page");
4667 file.seek(SeekFrom::Start(membership.offset)).expect("membership start");
4668 file.write_all(&[255]).expect("damage membership");
4669 let error = reader.skips_codes(0, 1, &[3]).expect_err("corruption must not skip rows");
4670 assert!(error.message().contains("membership page checksum differs"), "{error}");
4671 fs::remove_file(path).expect("remove scratch file");
4672 }
4673
4674 #[test]
4675 fn membership_delta_stream_is_sorted_exact_and_bounded() {
4676 let unique = unique_codes(&[900, 4, 4, 72, 9, u32::MAX]);
4677 assert_eq!(unique, [4, 9, 72, 900, u32::MAX]);
4678 let encoded = encode_membership(&unique);
4679 assert_eq!(
4680 decode_membership(&encoded).expect("valid membership"),
4681 [4, 9, 72, 900, u32::MAX]
4682 );
4683 let merged = merged_codes(vec![vec![4, 900], vec![9, 900, u32::MAX], vec![72]]);
4686 assert_eq!(merged, [4, 9, 72, 900, u32::MAX]);
4687 assert_eq!(
4688 decode_membership(&encode_membership(&merged)).expect("valid membership"),
4689 unique
4690 );
4691 assert!(decode_membership(&[1, 0x80]).is_err(), "a truncated varint is invalid");
4692 assert!(
4693 decode_membership(&[1, 0xff, 0xff, 0xff, 0xff, 0x10]).is_err(),
4694 "a value past u32 is invalid"
4695 );
4696 }
4697
4698 #[test]
4699 fn a_global_dictionary_may_be_larger_than_one_column_page() {
4700 let dictionary = Page {
4701 offset: HEADER,
4702 length: u32::try_from(MAX_PAGE + 1).expect("the page bound fits on disk"),
4703 hash: 0,
4704 };
4705 let table = Table {
4706 name: "items".to_owned(),
4707 fields: vec![Field::new("text", LogicalType::Varchar)],
4708 stripes: Vec::new(),
4709 rows: 0,
4710 dictionaries: vec![Some(dictionary)],
4711 frequencies: vec![None],
4712 };
4713 let directory = encode_directory(&table).expect("directory");
4714 let file_size = dictionary.offset + u64::from(dictionary.length) + 1;
4715
4716 let decoded = decode_directory(&directory, file_size).expect("large lazy dictionary");
4717 assert_eq!(decoded.dictionaries[0].expect("dictionary").length, dictionary.length);
4718 }
4719
4720 #[test]
4721 fn a_column_with_one_value_everywhere_costs_almost_nothing_a_row() {
4722 let path = path("constant-codes");
4723 let mut writer =
4724 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
4725 .expect("new file");
4726 let empty = vec![Value::Varchar(String::new()); 1024];
4727 for _ in 0..4 {
4728 let column = Vector::from_values(LogicalType::Varchar, &empty).expect("strings");
4729 writer.append(&Chunk::new(vec![column]).expect("one column")).expect("a part");
4730 }
4731 writer.finish().expect("commit");
4732
4733 let reader = Reader::open(&path).expect("valid directory");
4734 let pages = reader.layout().columns.first().expect("one column").pages;
4735 assert!(pages < 256, "{pages} bytes of pages for 4,096 rows of one value");
4739 let read = reader.read(3, &[0]).expect("the last part back");
4740 assert_eq!(read.value_at(0, 0), Value::Varchar(String::new()));
4741 assert_eq!(read.value_at(1023, 0), Value::Varchar(String::new()));
4742 fs::remove_file(path).expect("remove scratch file");
4743 }
4744
4745 #[test]
4746 fn a_code_stream_the_cascade_cannot_shrink_is_left_alone() {
4747 let mut state: u32 = 0x9e37_79b9;
4751 let spread: Vec<u32> = (0..1024)
4752 .map(|_| {
4753 state ^= state << 13;
4754 state ^= state >> 17;
4755 state ^= state << 5;
4756 state
4757 })
4758 .collect();
4759 assert_eq!(encoded_codes(&spread).expect("no failure"), None);
4760 let near: Vec<u32> = (0..1024).collect();
4761 let coded = encoded_codes(&near).expect("no failure").expect("counting up is packable");
4762 assert!(coded.len() < near.len() * 4, "{} bytes for a run of 1,024", coded.len());
4763 }
4764}