1#![forbid(unsafe_code)]
34
35use std::cmp::Ordering;
36use std::collections::{HashMap, VecDeque};
37use std::fs::{File, OpenOptions};
38use std::io::{Read, Seek, SeekFrom};
39use std::mem::{size_of, size_of_val};
40use std::path::Path;
41use std::slice;
42use std::sync::atomic::{AtomicUsize, Ordering as Atomic};
43use std::sync::{Arc, Mutex, OnceLock};
44
45use rudb_common::bounds::{Bound, Op, scaled_as};
46use rudb_common::{Clustering, Error, Field, LogicalType, PhysicalType, Result, Value, Width};
47use rudb_encoding::{bitpack, chooser, integer, string};
48use rudb_storage::sieve::Sieve;
49use rudb_storage::{Probe, Range, Zone};
50use rudb_vector::string::StringColumn;
51use rudb_vector::validity::Validity;
52use rudb_vector::{Buffer, Chunk, Data, Packed, TextSource, Vector, search_below};
53
54mod zones;
55
56pub use zones::{Common, Stripes, distincts};
57
58const MAGIC: &[u8; 8] = b"RUDBNV10";
59const DIRECTORY: &[u8; 8] = b"RUDBDI10";
60const CATALOG: &[u8; 8] = b"RUDBCA10";
61const FORMAT: u32 = 22;
62const HEADER: u64 = 80;
63const SLOT_BYTES: usize = 28;
64const MAX_PAGE: usize = 256 * 1024 * 1024;
65const MAX_DIRECTORY: usize = 128 * 1024 * 1024;
66const FREQUENCIES: &[u8; 8] = b"RUDBFQ2\0";
67const CLUSTERING: &[u8; 8] = b"RUDBCL1\0";
74const FREQUENCY_CANDIDATES: usize = 32_768;
75const FREQUENCY_ENTRIES: usize = 512;
76const FREQUENCY_BUILD_RANK: usize = 10;
77const FREQUENCY_ORDINALS: usize = 65_536;
78const MAX_FREQUENCY_WORKERS: usize = 32;
85
86const MAX_ENCODE_WORKERS: usize = 32;
93
94const SIEVE_BUDGET: usize = 8 * 1024;
102
103const PART_BOUND_BYTES: usize = 24;
112
113fn io(error: std::io::Error) -> Error {
114 Error::io(error.to_string())
115}
116
117fn invalid(message: &str) -> Error {
118 Error::invalid_input(format!("invalid rudb native file: {message}"))
119}
120
121fn sum(counts: impl Iterator<Item = u64>) -> u64 {
123 counts.fold(0, u64::saturating_add)
124}
125
126fn span_bytes(spans: &[Span], at: usize) -> u64 {
128 spans.get(at).map_or(0, |span| u64::from(span.length))
129}
130
131fn page_bytes(pages: &[Option<Page>], at: usize) -> u64 {
133 pages.get(at).and_then(Option::as_ref).map_or(0, Page::bytes)
134}
135
136fn checksum(bytes: &[u8]) -> u64 {
146 const P1: u64 = 11_400_714_785_074_694_791;
147 const P2: u64 = 14_029_467_366_897_019_727;
148 const P3: u64 = 1_609_587_929_392_839_161;
149 const P4: u64 = 9_650_029_242_287_828_579;
150 const P5: u64 = 2_870_177_450_012_600_261;
151 let round = |state: u64, word: u64| {
152 state.wrapping_add(word.wrapping_mul(P2)).rotate_left(31).wrapping_mul(P1)
153 };
154 let merge = |state: u64, lane: u64| (state ^ round(0, lane)).wrapping_mul(P1).wrapping_add(P4);
155 let word = |chunk: &[u8]| u64::from_le_bytes(chunk.try_into().expect("eight checksum bytes"));
156
157 let mut blocks = bytes.chunks_exact(32);
160 let mut rest = blocks.remainder();
161 let mut hash = if bytes.len() >= 32 {
162 let mut one = P1.wrapping_add(P2);
163 let mut two = P2;
164 let mut three = 0;
165 let mut four = 0_u64.wrapping_sub(P1);
166 for block in blocks.by_ref() {
167 one = round(one, word(&block[..8]));
168 two = round(two, word(&block[8..16]));
169 three = round(three, word(&block[16..24]));
170 four = round(four, word(&block[24..]));
171 }
172 let combined = one
173 .rotate_left(1)
174 .wrapping_add(two.rotate_left(7))
175 .wrapping_add(three.rotate_left(12))
176 .wrapping_add(four.rotate_left(18));
177 merge(merge(merge(merge(combined, one), two), three), four)
178 } else {
179 P5
180 };
181 hash = hash.wrapping_add(bytes.len() as u64);
182 let mut words = rest.chunks_exact(8);
183 for chunk in words.by_ref() {
184 hash ^= round(0, word(chunk));
185 hash = hash.rotate_left(27).wrapping_mul(P1).wrapping_add(P4);
186 }
187 rest = words.remainder();
188 if rest.len() >= 4 {
189 let (head, tail) = rest.split_at(4);
190 let quarter = u32::from_le_bytes(head.try_into().expect("four checksum bytes"));
191 hash ^= u64::from(quarter).wrapping_mul(P1);
192 hash = hash.rotate_left(23).wrapping_mul(P2).wrapping_add(P3);
193 rest = tail;
194 }
195 for &byte in rest {
196 hash ^= u64::from(byte).wrapping_mul(P5);
197 hash = hash.rotate_left(11).wrapping_mul(P1);
198 }
199 hash ^= hash >> 33;
200 hash = hash.wrapping_mul(P2);
201 hash ^= hash >> 29;
202 hash = hash.wrapping_mul(P3);
203 hash ^ (hash >> 32)
204}
205
206#[derive(Debug, Clone, Copy)]
207struct Slot {
208 offset: u64,
209 length: u32,
210 generation: u64,
211 hash: u64,
212}
213
214impl Slot {
215 fn bytes(self) -> [u8; SLOT_BYTES] {
216 let mut result = [0; SLOT_BYTES];
217 result[..8].copy_from_slice(&self.offset.to_le_bytes());
218 result[8..12].copy_from_slice(&self.length.to_le_bytes());
219 result[12..20].copy_from_slice(&self.generation.to_le_bytes());
220 result[20..28].copy_from_slice(&self.hash.to_le_bytes());
221 result
222 }
223
224 fn read(bytes: &[u8]) -> Self {
225 Self {
226 offset: u64::from_le_bytes(bytes[..8].try_into().expect("eight bytes")),
227 length: u32::from_le_bytes(bytes[8..12].try_into().expect("four bytes")),
228 generation: u64::from_le_bytes(bytes[12..20].try_into().expect("eight bytes")),
229 hash: u64::from_le_bytes(bytes[20..28].try_into().expect("eight bytes")),
230 }
231 }
232}
233
234#[derive(Debug, Clone, Copy)]
235struct Page {
236 offset: u64,
237 length: u32,
238 hash: u64,
239}
240
241impl Page {
242 fn bytes(&self) -> u64 {
244 u64::from(self.length)
245 }
246}
247
248#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
249enum FrequencyValue {
250 Null,
251 Integer(i128),
252 Code(u32),
253}
254
255#[derive(Debug, Clone)]
256struct FrequencyEntry {
257 value: FrequencyValue,
258 count: u64,
259}
260
261#[derive(Debug, Clone)]
266struct FrequencySummary {
267 entries: Vec<FrequencyEntry>,
268 omitted_max: u64,
269 ordinals: Vec<u64>,
270}
271
272#[derive(Debug, Clone)]
277pub struct FrequencyPrefix {
278 pub entries: Vec<(Value, u64)>,
280 pub omitted_max: u64,
282}
283
284#[derive(Debug, Clone, PartialEq, Eq)]
286pub struct FrequencyOccurrences {
287 pub omitted_max: u64,
289 pub ordinals: Vec<u64>,
291}
292
293#[derive(Debug, Clone, Copy, Default)]
300struct Span {
301 offset: u64,
302 length: u32,
303}
304
305#[derive(Debug, Clone)]
307pub struct Stripe {
308 rows: usize,
309 parts: Vec<u32>,
312 index: Span,
316 pages: Vec<Span>,
317 memberships: Vec<Option<Page>>,
318 sieves: Vec<Option<Page>>,
321 part_ranges: Vec<Option<Page>>,
332 zone: Zone,
333}
334
335impl Stripe {
336 #[must_use]
338 pub fn rows(&self) -> usize {
339 self.rows
340 }
341
342 #[must_use]
344 pub fn parts(&self) -> usize {
345 self.parts.len()
346 }
347
348 #[must_use]
354 pub fn zone(&self) -> &Zone {
355 &self.zone
356 }
357}
358
359#[derive(Debug, Clone)]
361pub struct Table {
362 name: String,
363 fields: Vec<Field>,
364 stripes: Vec<Stripe>,
365 rows: usize,
366 dictionaries: Vec<Option<Page>>,
367 frequencies: Vec<Option<FrequencySummary>>,
368 distincts: Vec<Option<u64>>,
378 clustering: Option<Clustering>,
386}
387
388impl Table {
389 #[must_use]
391 pub fn name(&self) -> &str {
392 &self.name
393 }
394
395 #[must_use]
397 pub fn fields(&self) -> &[Field] {
398 &self.fields
399 }
400
401 #[must_use]
403 pub fn rows(&self) -> usize {
404 self.rows
405 }
406
407 #[must_use]
409 pub fn stripes(&self) -> &[Stripe] {
410 &self.stripes
411 }
412
413 #[must_use]
415 pub fn clustering(&self) -> Option<&Clustering> {
416 self.clustering.as_ref()
417 }
418}
419
420#[derive(Debug, Clone)]
432struct Entry {
433 name: String,
434 fields: Vec<Field>,
435 rows: usize,
436 directory: Page,
438}
439
440#[derive(Debug, Clone)]
442pub struct ColumnLayout {
443 pub name: String,
445 pub kind: String,
447 pub pages: u64,
449 pub memberships: u64,
451 pub sieves: u64,
453 pub part_ranges: u64,
455 pub dictionary: u64,
457}
458
459impl ColumnLayout {
460 #[must_use]
462 pub fn total(&self) -> u64 {
463 self.pages
464 .saturating_add(self.memberships)
465 .saturating_add(self.sieves)
466 .saturating_add(self.part_ranges)
467 .saturating_add(self.dictionary)
468 }
469}
470
471#[derive(Debug, Clone)]
482pub struct Layout {
483 pub file: u64,
485 pub rows: usize,
487 pub stripes: usize,
489 pub parts: usize,
491 pub columns: Vec<ColumnLayout>,
493 pub indexes: u64,
496 pub directory: u64,
498 pub header: u64,
500}
501
502impl Layout {
503 #[must_use]
505 pub fn columns_total(&self) -> u64 {
506 self.columns.iter().map(ColumnLayout::total).fold(0, u64::saturating_add)
507 }
508
509 #[must_use]
515 pub fn unaccounted(&self) -> u64 {
516 self.file
517 .saturating_sub(self.columns_total())
518 .saturating_sub(self.indexes)
519 .saturating_sub(self.directory)
520 .saturating_sub(self.header)
521 }
522}
523
524#[derive(Debug)]
526struct GlobalDictionary {
527 primary: HashMap<u64, u32>,
528 collisions: HashMap<u64, Vec<u32>>,
529 offsets: Vec<u32>,
530 payload: Vec<u8>,
531 counts: Vec<u64>,
532 nulls: u64,
533}
534
535impl GlobalDictionary {
536 fn new() -> Self {
537 Self {
538 primary: HashMap::new(),
539 collisions: HashMap::new(),
540 offsets: vec![0],
541 payload: Vec::new(),
542 counts: Vec::new(),
543 nulls: 0,
544 }
545 }
546
547 fn bytes(&self, code: u32) -> Option<&[u8]> {
548 let start = *self.offsets.get(code as usize)? as usize;
549 let end = *self.offsets.get(code as usize + 1)? as usize;
550 self.payload.get(start..end)
551 }
552
553 fn code(&mut self, text: &str) -> Result<u32> {
554 let hash = checksum(text.as_bytes());
555 if let Some(&code) = self.primary.get(&hash) {
556 if self.bytes(code) == Some(text.as_bytes()) {
557 return Ok(code);
558 }
559 if let Some(codes) = self.collisions.get(&hash) {
560 if let Some(code) =
561 codes.iter().copied().find(|&code| self.bytes(code) == Some(text.as_bytes()))
562 {
563 return Ok(code);
564 }
565 }
566 let code = self.insert(text)?;
567 self.collisions.entry(hash).or_default().push(code);
568 return Ok(code);
569 }
570 let code = self.insert(text)?;
571 self.primary.insert(hash, code);
572 Ok(code)
573 }
574
575 fn insert(&mut self, text: &str) -> Result<u32> {
576 let code = u32::try_from(self.offsets.len() - 1)
577 .map_err(|_| invalid("global dictionary has too many values"))?;
578 self.payload.extend_from_slice(text.as_bytes());
579 self.offsets.push(
580 u32::try_from(self.payload.len())
581 .map_err(|_| invalid("global dictionary payload exceeds 4 GiB"))?,
582 );
583 self.counts.push(0);
584 Ok(code)
585 }
586
587 fn ranked(&self) -> Vec<(u64, u32)> {
607 let count = self.offsets.len() - 1;
608 let mut ranked = (0..count)
609 .map(|code| {
610 let code = code as u32;
611 (head(self.bytes(code).unwrap_or_default()), code)
612 })
613 .collect::<Vec<_>>();
614 ranked.sort_unstable_by(|left, right| {
615 left.0.cmp(&right.0).then_with(|| self.bytes(left.1).cmp(&self.bytes(right.1)))
616 });
617 ranked
618 }
619
620 fn observe(&mut self, code: u32, null: bool) -> Result<()> {
621 if null {
622 self.nulls = self.nulls.saturating_add(1);
623 return Ok(());
624 }
625 let count = self
626 .counts
627 .get_mut(code as usize)
628 .ok_or_else(|| invalid("global dictionary count code is out of range"))?;
629 *count = count.saturating_add(1);
630 Ok(())
631 }
632}
633
634#[derive(Debug)]
642pub struct Writer {
643 file: File,
644 at: u64,
652 table: Table,
653 generation: u64,
654 order: Vec<((u64, u64), (u64, u64))>,
657 next_order: u64,
658 dictionaries: Vec<Option<GlobalDictionary>>,
659 pending: Vec<PendingChunk>,
660 closed: Vec<Entry>,
662}
663
664#[derive(Debug)]
672struct PendingChunk {
673 order: (u64, u64),
674 chunk: Chunk,
675}
676
677#[derive(Debug)]
683struct ColumnStripe {
684 pages: Vec<Vec<u8>>,
685 codes: Vec<Option<Vec<u32>>>,
686 sieves: Vec<Option<Sieve>>,
687 ranges: Vec<Range>,
688}
689
690fn weight(ty: &LogicalType) -> usize {
698 match ty {
699 LogicalType::Varchar | LogicalType::Blob => 64,
700 LogicalType::BigInt
701 | LogicalType::UBigInt
702 | LogicalType::Timestamp
703 | LogicalType::Double
704 | LogicalType::Decimal { .. } => 8,
705 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date | LogicalType::Float => 4,
706 LogicalType::SmallInt | LogicalType::USmallInt => 2,
707 _ => 1,
708 }
709}
710
711pub const STRIPE_PARTS: usize = 64;
718
719const DICTIONARY_DECIDE_ROWS: usize = 4_096;
727
728const DICTIONARY_DISTINCT_IN_TEN: usize = 9;
744
745const INDEX_ENTRY: usize = size_of::<u32>() + size_of::<u64>();
747
748fn index_section(parts: usize) -> Result<usize> {
750 parts
751 .checked_mul(INDEX_ENTRY)
752 .and_then(|bytes| bytes.checked_add(size_of::<u64>()))
753 .ok_or_else(|| invalid("index page length overflow"))
754}
755
756impl Writer {
757 pub fn open(
775 path: impl AsRef<Path>,
776 name: impl Into<String>,
777 fields: Vec<Field>,
778 ) -> Result<Self> {
779 for field in &fields {
780 type_tag(&field.ty)?;
781 }
782 let name = name.into();
783 let path = path.as_ref();
784 let (_, size, slot, bytes, _) = slot_bytes(path)?;
785 let closed = decode_catalog(&bytes, size)?;
786 if closed.iter().any(|held| held.name == name) {
787 return Err(invalid("two tables in one native file have the same name"));
788 }
789 let generation = slot
794 .generation
795 .checked_add(1)
796 .ok_or_else(|| invalid("native file generation overflow"))?;
797 let file = OpenOptions::new().write(true).read(true).open(path).map_err(io)?;
798 Ok(Self {
799 file,
800 at: size,
803 dictionaries: fields
804 .iter()
805 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
806 .collect(),
807 table: Table {
808 name,
809 dictionaries: vec![None; fields.len()],
810 distincts: vec![None; fields.len()],
811 fields,
812 stripes: Vec::new(),
813 rows: 0,
814 frequencies: Vec::new(),
815 clustering: None,
816 },
817 generation,
818 order: Vec::new(),
819 next_order: 0,
820 pending: Vec::with_capacity(STRIPE_PARTS),
821 closed,
822 })
823 }
824
825 pub fn create(
831 path: impl AsRef<Path>,
832 name: impl Into<String>,
833 fields: Vec<Field>,
834 ) -> Result<Self> {
835 for field in &fields {
836 type_tag(&field.ty)?;
837 }
838 let file =
839 OpenOptions::new().write(true).read(true).create_new(true).open(path).map_err(io)?;
840 let mut header = [0; HEADER as usize];
841 header[..8].copy_from_slice(MAGIC);
842 header[8..12].copy_from_slice(&FORMAT.to_le_bytes());
843 write_at(&file, 0, &header)?;
844 Ok(Self {
845 file,
846 at: HEADER,
847 dictionaries: fields
848 .iter()
849 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
850 .collect(),
851 table: Table {
852 name: name.into(),
853 dictionaries: vec![None; fields.len()],
854 distincts: vec![None; fields.len()],
855 fields,
856 stripes: Vec::new(),
857 rows: 0,
858 frequencies: Vec::new(),
859 clustering: None,
860 },
861 generation: 1,
862 order: Vec::new(),
863 next_order: 0,
864 pending: Vec::with_capacity(STRIPE_PARTS),
865 closed: Vec::new(),
866 })
867 }
868
869 pub fn next(mut self, name: impl Into<String>, fields: Vec<Field>) -> Result<Self> {
880 for field in &fields {
881 type_tag(&field.ty)?;
882 }
883 let name = name.into();
884 let entry = self.close()?;
885 if self.closed.iter().chain(std::iter::once(&entry)).any(|held| held.name == name) {
886 return Err(invalid("two tables in one native file have the same name"));
887 }
888 let Self { file, at, generation, mut closed, .. } = self;
889 closed.push(entry);
890 Ok(Self {
891 file,
892 at,
893 generation,
894 closed,
895 dictionaries: fields
896 .iter()
897 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
898 .collect(),
899 table: Table {
900 name,
901 dictionaries: vec![None; fields.len()],
902 distincts: vec![None; fields.len()],
903 fields,
904 stripes: Vec::new(),
905 rows: 0,
906 frequencies: Vec::new(),
907 clustering: None,
908 },
909 order: Vec::new(),
910 next_order: 0,
911 pending: Vec::with_capacity(STRIPE_PARTS),
912 })
913 }
914
915 pub fn declare(mut self, clustering: Clustering) -> Result<Self> {
930 self.table.clustering = Some(Clustering::new(
933 clustering.columns().to_vec(),
934 clustering.width(),
935 self.table.fields.len(),
936 )?);
937 Ok(self)
938 }
939
940 fn put(&mut self, bytes: &[u8]) -> Result<()> {
945 write_at(&self.file, self.at, bytes)?;
946 self.at = self
947 .at
948 .checked_add(bytes.len() as u64)
949 .ok_or_else(|| invalid("native file length overflow"))?;
950 Ok(())
951 }
952
953 pub fn append(&mut self, chunk: &Chunk) -> Result<()> {
959 let order = (self.next_order, 0);
960 self.next_order = self.next_order.saturating_add(1);
961 self.append_at(order, chunk)
962 }
963
964 pub fn append_at(&mut self, order: (u64, u64), chunk: &Chunk) -> Result<()> {
975 if chunk.is_empty() {
976 return Ok(());
977 }
978 self.admit(chunk)?;
979 if self.pending.last().is_some_and(|last| last.order > order) {
980 self.flush_pending()?;
981 }
982 self.pending.push(PendingChunk { order, chunk: chunk.clone() });
987 if self.pending.len() == STRIPE_PARTS {
988 self.flush_pending()?;
989 }
990 Ok(())
991 }
992
993 pub fn append_stripe(&mut self, parts: Vec<((u64, u64), Chunk)>) -> Result<()> {
1009 if parts.len() > STRIPE_PARTS {
1010 return Err(invalid("a stripe was handed more parts than it holds"));
1011 }
1012 self.flush_pending()?;
1015 for (order, chunk) in parts {
1016 if chunk.is_empty() {
1017 continue;
1018 }
1019 self.admit(&chunk)?;
1020 self.pending.push(PendingChunk { order, chunk });
1021 }
1022 self.flush_pending()
1023 }
1024
1025 fn admit(&mut self, chunk: &Chunk) -> Result<()> {
1027 if chunk.width() != self.table.fields.len() {
1028 return Err(invalid("chunk width differs from table schema"));
1029 }
1030 for (index, field) in self.table.fields.iter().enumerate() {
1031 if chunk.column(index)?.logical_type() != &field.ty {
1032 return Err(invalid("chunk type differs from table schema"));
1033 }
1034 }
1035 self.table.rows = self
1036 .table
1037 .rows
1038 .checked_add(chunk.len())
1039 .ok_or_else(|| invalid("row count overflow"))?;
1040 Ok(())
1041 }
1042
1043 fn encode_column(
1074 index: usize,
1075 held: &[PendingChunk],
1076 dictionary: &mut Option<GlobalDictionary>,
1077 ) -> Result<ColumnStripe> {
1078 let deciding = dictionary.as_ref().is_some_and(|held| held.offsets.len() == 1);
1081 let stripe = Self::encode_pages(index, held, dictionary.as_mut())?;
1082 if !deciding {
1083 return Ok(stripe);
1084 }
1085 let rows: usize = held.iter().map(|pending| pending.chunk.len()).sum();
1086 let distinct = dictionary.as_ref().map_or(0, |held| held.offsets.len() - 1);
1087 if rows < DICTIONARY_DECIDE_ROWS
1088 || distinct.saturating_mul(10) <= rows.saturating_mul(DICTIONARY_DISTINCT_IN_TEN)
1089 {
1090 return Ok(stripe);
1091 }
1092 *dictionary = None;
1093 Self::encode_pages(index, held, None)
1094 }
1095
1096 fn encode_pages(
1098 index: usize,
1099 held: &[PendingChunk],
1100 mut dictionary: Option<&mut GlobalDictionary>,
1101 ) -> Result<ColumnStripe> {
1102 let mut stripe = ColumnStripe {
1103 pages: Vec::with_capacity(held.len()),
1104 codes: Vec::with_capacity(held.len()),
1105 sieves: Vec::with_capacity(held.len()),
1106 ranges: Vec::with_capacity(held.len()),
1107 };
1108 for pending in held {
1109 let column = pending.chunk.column(index)?;
1110 let (bytes, unique) = encode(column, dictionary.as_deref_mut())?;
1111 if bytes.len() > MAX_PAGE {
1112 return Err(invalid("column page exceeds the configured bound"));
1113 }
1114 let range = Range::of(column);
1117 let sieve = match dictionary {
1130 Some(_) => None,
1131 None => Sieve::of(column, &range, SIEVE_BUDGET)
1132 .filter(|sieve| sieve.len() < bytes.len()),
1133 };
1134 stripe.pages.push(bytes);
1135 stripe.codes.push(unique);
1136 stripe.sieves.push(sieve);
1137 stripe.ranges.push(range);
1138 }
1139 Ok(stripe)
1140 }
1141
1142 fn encode_columns(&mut self, held: &[PendingChunk]) -> Result<Vec<ColumnStripe>> {
1151 let width = self.table.fields.len();
1152 let workers = std::thread::available_parallelism()
1153 .map_or(1, usize::from)
1154 .min(MAX_ENCODE_WORKERS)
1155 .min(width);
1156 if workers <= 1 || held.len() <= 1 {
1157 return self
1158 .dictionaries
1159 .iter_mut()
1160 .enumerate()
1161 .map(|(index, dictionary)| Self::encode_column(index, held, dictionary))
1162 .collect();
1163 }
1164 let mut jobs: Vec<(usize, Option<GlobalDictionary>)> =
1167 std::mem::take(&mut self.dictionaries).into_iter().enumerate().collect();
1168 jobs.sort_by_key(|(index, _)| weight(&self.table.fields[*index].ty));
1170 let queue = Mutex::new(jobs);
1171 let pieces = std::thread::scope(|scope| {
1172 (0..workers)
1173 .map(|_| {
1174 scope.spawn(|| {
1175 let mut mine = Vec::new();
1176 loop {
1177 let taken = queue
1178 .lock()
1179 .map_err(|_| Error::internal("a native encode worker panicked"))?
1180 .pop();
1181 let Some((index, mut dictionary)) = taken else { break };
1182 let encoded = Self::encode_column(index, held, &mut dictionary)?;
1183 mine.push((index, dictionary, encoded));
1184 }
1185 Ok(mine)
1186 })
1187 })
1188 .collect::<Vec<_>>()
1189 .into_iter()
1190 .map(|handle| {
1191 handle.join().map_err(|_| Error::internal("a native encode worker panicked"))?
1192 })
1193 .collect::<Result<Vec<_>>>()
1194 })?;
1195 let mut dictionaries: Vec<Option<GlobalDictionary>> = (0..width).map(|_| None).collect();
1196 let mut encoded: Vec<Option<ColumnStripe>> = (0..width).map(|_| None).collect();
1197 for piece in pieces {
1198 for (index, dictionary, stripe) in piece {
1199 dictionaries[index] = dictionary;
1200 encoded[index] = Some(stripe);
1201 }
1202 }
1203 self.dictionaries = dictionaries;
1204 encoded
1205 .into_iter()
1206 .map(|stripe| stripe.ok_or_else(|| Error::internal("a column was never encoded")))
1207 .collect()
1208 }
1209
1210 fn flush_pending(&mut self) -> Result<()> {
1212 if self.pending.is_empty() {
1213 return Ok(());
1214 }
1215 let width = self.table.fields.len();
1216 let mut held = std::mem::take(&mut self.pending);
1219 let parts = held.len();
1220 let encoded = self.encode_columns(&held)?;
1221 let mut pages = Vec::with_capacity(width);
1222 let mut memberships = vec![None; width];
1223 let mut ranges = Vec::with_capacity(width);
1224 let mut index = Vec::with_capacity(width.saturating_mul(index_section(parts)?));
1225 for stripe in &encoded {
1226 let offset = self.at;
1227 let section = index.len();
1228 let mut length = 0_usize;
1229 for bytes in &stripe.pages {
1230 write_at(&self.file, self.at + length as u64, bytes)?;
1231 put_u32(
1232 &mut index,
1233 u32::try_from(bytes.len()).map_err(|_| invalid("part length overflow"))?,
1234 );
1235 put_u64(&mut index, checksum(bytes));
1236 length = length
1237 .checked_add(bytes.len())
1238 .ok_or_else(|| invalid("column page length overflow"))?;
1239 }
1240 let hash = checksum(&index[section..]);
1241 put_u64(&mut index, hash);
1242 if length > MAX_PAGE {
1243 return Err(invalid("column page exceeds the configured bound"));
1244 }
1245 self.at = self
1246 .at
1247 .checked_add(length as u64)
1248 .ok_or_else(|| invalid("native file length overflow"))?;
1249 pages.push(Span {
1250 offset,
1251 length: u32::try_from(length).map_err(|_| invalid("page length overflow"))?,
1252 });
1253 ranges.push(merged_range(stripe.ranges.iter().cloned()));
1254 }
1255 for (membership, stripe) in memberships.iter_mut().zip(&encoded) {
1256 if stripe.codes.iter().all(Option::is_none) {
1257 continue;
1258 }
1259 let lists = stripe
1260 .codes
1261 .iter()
1262 .map(|codes| codes.clone().unwrap_or_default())
1263 .collect::<Vec<_>>();
1264 let bytes = encode_membership(&merged_codes(lists));
1265 let offset = self.at;
1266 self.put(&bytes)?;
1267 *membership = Some(Page {
1268 offset,
1269 length: u32::try_from(bytes.len())
1270 .map_err(|_| invalid("membership page length overflow"))?,
1271 hash: checksum(&bytes),
1272 });
1273 }
1274 let mut sieves = vec![None; width];
1275 for (page, stripe) in sieves.iter_mut().zip(&encoded) {
1276 if stripe.sieves.iter().all(Option::is_none) {
1277 continue;
1278 }
1279 let bytes = encode_sieves(stripe.sieves.iter())?;
1280 let offset = self.at;
1281 self.put(&bytes)?;
1282 *page = Some(Page {
1283 offset,
1284 length: u32::try_from(bytes.len())
1285 .map_err(|_| invalid("sieve page length overflow"))?,
1286 hash: checksum(&bytes),
1287 });
1288 }
1289 let mut part_ranges = vec![None; width];
1295 if parts > 1 {
1296 for ((page, stripe), span) in part_ranges.iter_mut().zip(&encoded).zip(&pages) {
1297 let bytes = encode_part_ranges(&stripe.ranges)?;
1298 if bytes.len() >= span.length as usize {
1299 continue;
1300 }
1301 let offset = self.at;
1302 self.put(&bytes)?;
1303 *page = Some(Page {
1304 offset,
1305 length: u32::try_from(bytes.len())
1306 .map_err(|_| invalid("part range page length overflow"))?,
1307 hash: checksum(&bytes),
1308 });
1309 }
1310 }
1311 let offset = self.at;
1312 self.put(&index)?;
1313 let index = Span {
1314 offset,
1315 length: u32::try_from(index.len())
1316 .map_err(|_| invalid("index page length overflow"))?,
1317 };
1318 let mut rows = 0_usize;
1319 let mut lengths = Vec::with_capacity(parts);
1320 let mut span = None;
1321 for pending in held.drain(..) {
1322 let part = pending.chunk.len();
1323 rows = rows.checked_add(part).ok_or_else(|| invalid("row count overflow"))?;
1324 lengths.push(u32::try_from(part).map_err(|_| invalid("part row count overflow"))?);
1325 span = Some(
1326 span.map_or((pending.order, pending.order), |(first, _)| (first, pending.order)),
1327 );
1328 }
1329 self.order.push(span.ok_or_else(|| invalid("a stripe was flushed with no parts"))?);
1330 self.table.stripes.push(Stripe {
1331 rows,
1332 parts: lengths,
1333 index,
1334 pages,
1335 memberships,
1336 sieves,
1337 part_ranges,
1338 zone: Zone::from_ranges(ranges),
1339 });
1340 self.pending = held;
1342 Ok(())
1343 }
1344
1345 fn numeric_frequency(&self, column: usize) -> Result<Option<FrequencySummary>> {
1349 let ty = &self.table.fields[column].ty;
1350 if !matches!(
1351 ty,
1352 LogicalType::TinyInt
1353 | LogicalType::SmallInt
1354 | LogicalType::Integer
1355 | LogicalType::BigInt
1356 | LogicalType::UTinyInt
1357 | LogicalType::USmallInt
1358 | LogicalType::UInteger
1359 | LogicalType::UBigInt
1360 | LogicalType::Date
1361 | LogicalType::Timestamp
1362 ) {
1363 return Ok(None);
1364 }
1365 let mut candidates: HashMap<FrequencyValue, u32> = HashMap::new();
1366 let mut decrements = 0_u64;
1367 self.visit_numeric(column, |_, value| {
1368 if let Some(count) = candidates.get_mut(&value) {
1369 *count = count.saturating_add(1);
1370 } else if candidates.len() < FREQUENCY_CANDIDATES {
1371 candidates.insert(value, 1);
1372 } else {
1373 candidates.retain(|_, count| {
1374 *count -= 1;
1375 *count != 0
1376 });
1377 decrements = decrements.saturating_add(1);
1378 }
1379 })?;
1380 let (exact, ordinals) = if decrements == 0 {
1381 (
1382 candidates
1383 .into_iter()
1384 .map(|(value, count)| (value, u64::from(count)))
1385 .collect::<HashMap<_, _>>(),
1386 Vec::new(),
1387 )
1388 } else {
1389 let mut lower = candidates.values().copied().collect::<Vec<_>>();
1390 lower.sort_unstable_by(|left, right| right.cmp(left));
1391 if lower.len() < FREQUENCY_BUILD_RANK
1392 || u64::from(lower[FREQUENCY_BUILD_RANK - 1]) <= decrements
1393 {
1394 return Ok(None);
1395 }
1396 let mut exact =
1397 candidates.into_keys().map(|value| (value, 0_u64)).collect::<HashMap<_, _>>();
1398 let mut ordinals = Vec::new();
1399 let mut exceeded = false;
1400 self.visit_numeric(column, |ordinal, value| {
1401 if let Some(count) = exact.get_mut(&value) {
1402 *count = count.saturating_add(1);
1403 if !exceeded {
1404 if ordinals.len() < FREQUENCY_ORDINALS {
1405 ordinals.push(ordinal);
1406 } else {
1407 ordinals.clear();
1408 exceeded = true;
1409 }
1410 }
1411 }
1412 })?;
1413 (exact, ordinals)
1414 };
1415 let mut entries = exact
1416 .into_iter()
1417 .map(|(value, count)| FrequencyEntry { value, count })
1418 .collect::<Vec<_>>();
1419 entries.sort_unstable_by(|left, right| {
1420 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
1421 });
1422 let omitted_max =
1423 entries.get(FREQUENCY_ENTRIES).map_or(decrements, |entry| decrements.max(entry.count));
1424 entries.truncate(FREQUENCY_ENTRIES);
1425 Ok(Some(FrequencySummary { entries, omitted_max, ordinals }))
1426 }
1427
1428 fn visit_numeric(
1429 &self,
1430 column: usize,
1431 mut visit: impl FnMut(u64, FrequencyValue),
1432 ) -> Result<()> {
1433 let ty = &self.table.fields[column].ty;
1434 let mut start = 0_u64;
1435 for stripe in &self.table.stripes {
1436 let spans = read_index(&self.file, stripe, column)?;
1437 let page = stripe.pages[column];
1438 let mut bytes = vec![0; page.length as usize];
1439 read_at(&self.file, page.offset, &mut bytes)?;
1440 for (span, &rows) in spans.iter().zip(&stripe.parts) {
1441 let part = part_bytes(&bytes, *span)?;
1442 if checksum(part) != span.hash {
1443 return Err(invalid("column page checksum differs while building frequencies"));
1444 }
1445 let rows = rows as usize;
1446 let vector = decode(ty, rows, part, None)?;
1447 for row in 0..rows {
1449 let value = if vector.is_null_at(row) {
1450 FrequencyValue::Null
1451 } else {
1452 let widened = match vector.signed_at(row) {
1456 Some(value) => Some(value),
1457 None => match vector.value_at(row) {
1458 Value::UTinyInt(value) => Some(i128::from(value)),
1459 Value::USmallInt(value) => Some(i128::from(value)),
1460 Value::UInteger(value) => Some(i128::from(value)),
1461 Value::UBigInt(value) => Some(i128::from(value)),
1462 _ => None,
1463 },
1464 };
1465 FrequencyValue::Integer(widened.ok_or_else(|| {
1466 invalid("numeric frequency page did not contain an integer value")
1467 })?)
1468 };
1469 visit(start.saturating_add(row as u64), value);
1470 }
1471 start = start.saturating_add(rows as u64);
1472 }
1473 }
1474 Ok(())
1475 }
1476
1477 fn numeric_frequencies(&self) -> Result<Vec<Option<FrequencySummary>>> {
1485 let mut columns = self
1486 .table
1487 .fields
1488 .iter()
1489 .enumerate()
1490 .filter_map(|(column, field)| {
1491 matches!(
1492 field.ty,
1493 LogicalType::TinyInt
1494 | LogicalType::SmallInt
1495 | LogicalType::Integer
1496 | LogicalType::BigInt
1497 | LogicalType::UTinyInt
1498 | LogicalType::USmallInt
1499 | LogicalType::UInteger
1500 | LogicalType::UBigInt
1501 | LogicalType::Date
1502 | LogicalType::Timestamp
1503 )
1504 .then_some(column)
1505 })
1506 .collect::<Vec<_>>();
1507 let workers = std::thread::available_parallelism()
1508 .map_or(1, usize::from)
1509 .min(MAX_FREQUENCY_WORKERS)
1510 .min(columns.len());
1511 if workers <= 1 {
1512 let mut frequencies = vec![None; self.table.fields.len()];
1513 for column in columns {
1514 frequencies[column] = self.numeric_frequency(column)?;
1515 }
1516 return Ok(frequencies);
1517 }
1518 columns.sort_by_key(|&column| weight(&self.table.fields[column].ty));
1521 let queue = Mutex::new(columns);
1522 let pieces = std::thread::scope(|scope| {
1523 (0..workers)
1524 .map(|_| {
1525 scope.spawn(|| {
1526 let mut mine = Vec::new();
1527 loop {
1528 let taken = queue
1529 .lock()
1530 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1531 .pop();
1532 let Some(column) = taken else { break };
1533 mine.push((column, self.numeric_frequency(column)?));
1534 }
1535 Ok(mine)
1536 })
1537 })
1538 .collect::<Vec<_>>()
1539 .into_iter()
1540 .map(|handle| {
1541 handle
1542 .join()
1543 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1544 })
1545 .collect::<Result<Vec<_>>>()
1546 })?;
1547 let mut frequencies = vec![None; self.table.fields.len()];
1548 for piece in pieces {
1549 for (column, summary) in piece {
1550 frequencies[column] = summary;
1551 }
1552 }
1553 Ok(frequencies)
1554 }
1555
1556 fn close(&mut self) -> Result<Entry> {
1567 self.flush_pending()?;
1568 let mut stripes = std::mem::take(&mut self.order)
1569 .into_iter()
1570 .zip(std::mem::take(&mut self.table.stripes))
1571 .collect::<Vec<_>>();
1572 stripes.sort_by_key(|(order, _)| order.0);
1573 let mut previous: Option<(u64, u64)> = None;
1574 for ((first, last), _) in &stripes {
1575 if previous.is_some_and(|previous| previous >= *first) {
1576 return Err(invalid("chunks did not arrive in source order"));
1577 }
1578 previous = Some(*last);
1579 }
1580 self.table.stripes = stripes.into_iter().map(|(_, stripe)| stripe).collect();
1581 self.table.frequencies = self.numeric_frequencies()?;
1582 let dictionaries = std::mem::take(&mut self.dictionaries);
1583 let orders = rankings(&dictionaries)?;
1584 for (index, (dictionary, order)) in dictionaries.into_iter().zip(orders).enumerate() {
1585 let Some(dictionary) = dictionary else { continue };
1586 self.table.distincts[index] =
1590 Some(dictionary.counts.iter().filter(|count| **count != 0).count() as u64);
1591 self.table.frequencies[index] = Some(code_frequency(&dictionary));
1592 let encoded = encode_global_dictionary(dictionary, &order)?;
1593 let offset = self.at;
1594 self.put(&encoded.index)?;
1595 self.put(&encoded.ranks)?;
1596 for block in &encoded.payload {
1597 self.put(block)?;
1598 }
1599 let payload_len =
1600 encoded.payload.iter().try_fold(0_usize, |len, block| len.checked_add(block.len()));
1601 let length = payload_len
1602 .and_then(|len| len.checked_add(encoded.index.len()))
1603 .and_then(|len| len.checked_add(encoded.ranks.len()))
1604 .ok_or_else(|| invalid("dictionary page length overflow"))?;
1605 self.table.dictionaries[index] = Some(Page {
1606 offset,
1607 length: u32::try_from(length)
1608 .map_err(|_| invalid("dictionary page length overflow"))?,
1609 hash: checksum(&encoded.index),
1610 });
1611 }
1612 let directory = encode_directory(&self.table)?;
1613 if directory.len() > MAX_DIRECTORY {
1614 return Err(invalid("directory exceeds the configured bound"));
1615 }
1616 let offset = self.at;
1617 self.put(&directory)?;
1618 Ok(Entry {
1619 name: self.table.name.clone(),
1620 fields: self.table.fields.clone(),
1621 rows: self.table.rows,
1622 directory: Page {
1623 offset,
1624 length: u32::try_from(directory.len())
1625 .map_err(|_| invalid("directory length overflow"))?,
1626 hash: checksum(&directory),
1627 },
1628 })
1629 }
1630
1631 pub fn finish(mut self) -> Result<Table> {
1641 let entry = self.close()?;
1642 let mut tables = std::mem::take(&mut self.closed);
1643 tables.push(entry);
1644 let catalog = encode_catalog(&tables)?;
1645 if catalog.len() > MAX_DIRECTORY {
1646 return Err(invalid("catalog exceeds the configured bound"));
1647 }
1648 let offset = self.at;
1649 self.put(&catalog)?;
1650 self.file.sync_all().map_err(io)?;
1654 let slot = Slot {
1655 offset,
1656 length: u32::try_from(catalog.len()).map_err(|_| invalid("catalog length overflow"))?,
1657 generation: self.generation,
1658 hash: checksum(&catalog),
1659 };
1660 write_at(&self.file, slot_offset(self.generation), &slot.bytes())?;
1665 self.file.sync_all().map_err(io)?;
1666 Ok(self.table)
1667 }
1668}
1669
1670#[derive(Debug, Clone)]
1672pub struct Reader {
1673 file: Arc<File>,
1674 table: Arc<Table>,
1675 dictionaries: Arc<Vec<OnceLock<Arc<Vector>>>>,
1676 loading: Arc<Vec<Mutex<()>>>,
1685 opened: Arc<AtomicUsize>,
1689 sieves: Arc<Vec<Vec<SieveSlot>>>,
1693 part_ranges: Arc<Vec<Vec<RangeSlot>>>,
1696 places: Arc<Vec<Place>>,
1698 cache: Arc<Vec<Mutex<Cached>>>,
1699 pages: Arc<AtomicUsize>,
1702 indexes: Arc<AtomicUsize>,
1705 kept: Arc<AtomicUsize>,
1708 size: u64,
1710 directory: u64,
1712 opening: Opening,
1714}
1715
1716#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1728pub struct Opening {
1729 pub reads: u32,
1732 pub bytes: u64,
1734}
1735
1736#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1738pub struct Reads {
1739 pub opening: Opening,
1741 pub pages: usize,
1743 pub indexes: usize,
1745 pub dictionaries: usize,
1748}
1749
1750#[derive(Debug, Clone, Copy)]
1752struct Place {
1753 stripe: u32,
1754 part: u32,
1755 rows: u32,
1756}
1757
1758#[derive(Debug, Clone, Copy)]
1760struct PartSpan {
1761 start: usize,
1762 length: usize,
1763 hash: u64,
1764}
1765
1766#[derive(Debug, Clone)]
1772struct CachedColumn {
1773 stripe: usize,
1774 index: Arc<Vec<PartSpan>>,
1775 page: Option<Arc<Vec<u8>>>,
1776}
1777
1778#[derive(Debug, Default)]
1798struct Cached {
1799 pages: Vec<Option<Arc<Vec<u8>>>>,
1800 order: VecDeque<usize>,
1801 loading: Vec<usize>,
1802 index: Vec<Option<Arc<Vec<PartSpan>>>>,
1803}
1804
1805const CACHED_STRIPES_PER_COLUMN: usize = 4;
1817
1818type SieveSlot = OnceLock<Arc<Vec<Option<Sieve>>>>;
1820
1821type RangeSlot = OnceLock<Arc<Vec<Range>>>;
1822
1823#[derive(Debug)]
1824struct NativeText {
1825 file: Arc<File>,
1826 values: usize,
1828 offsets: Vec<u8>,
1837 offset_bits: usize,
1840 ranks: usize,
1842 rank_at: u64,
1846 rank_ends: Vec<u64>,
1850 rank_hashes: Vec<u64>,
1851 rank_blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1852 code_bits: usize,
1855 code_ranks: OnceLock<Option<Vec<u32>>>,
1862 payload: u64,
1863 ends: Vec<u64>,
1866 hashes: Vec<u64>,
1867 blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1869 keep_budget: usize,
1872 payload_kept: AtomicUsize,
1880 searched: Mutex<HashMap<Vec<u8>, (usize, bool)>>,
1897}
1898
1899const TEXT_SEARCH_MEMO: usize = 64;
1904
1905const TEXT_PAYLOAD_VALUES: usize = 1024;
1921
1922const TEXT_KEEP_BUDGET: usize = 256 * 1024 * 1024;
1943
1944const TEXT_OFFSET_RUN: usize = 512;
1951
1952const DICTIONARY_HEADER: usize = 16;
1955
1956const TEXT_RANK_BLOCK: usize = 512;
1967
1968const RANK_BLOCK_HEADER: usize = size_of::<u64>() + 1;
1982
1983impl NativeText {
1984 fn payload_block(&self, block: usize) -> Result<Option<&[u8]>> {
1991 let Some(slot) = self.blocks.get(block) else { return Ok(None) };
1992 let bytes = slot.get_or_init(|| self.decode_block(block)).as_ref().map_err(Clone::clone)?;
1993 Ok(Some(bytes.as_slice()))
1994 }
1995
1996 fn decode_block(&self, block: usize) -> Result<Vec<u8>> {
2001 let start = if block == 0 { 0 } else { self.ends[block - 1] };
2002 let end = self.ends[block];
2003 let len = end
2004 .checked_sub(start)
2005 .ok_or_else(|| invalid("global dictionary block ends before it starts"))?;
2006 let mut stored = vec![
2007 0;
2008 usize::try_from(len).map_err(|_| invalid(
2009 "global dictionary block does not fit in memory"
2010 ))?
2011 ];
2012 read_at(&self.file, self.payload + start, &mut stored)?;
2013 if checksum(&stored) != self.hashes[block] {
2014 return Err(invalid("global dictionary payload checksum differs"));
2015 }
2016 let first = block * TEXT_PAYLOAD_VALUES;
2017 let last = (first + TEXT_PAYLOAD_VALUES).min(self.values);
2018 let want = self.end_within(last - 1)? as usize;
2019 let values = string::decode_flat(&stored)?;
2020 if values.len() != last - first {
2021 return Err(invalid("global dictionary block holds the wrong value count"));
2022 }
2023 let bytes = values.into_bytes();
2024 if bytes.len() != want {
2025 return Err(invalid("global dictionary block decodes to the wrong length"));
2026 }
2027 Ok(bytes)
2028 }
2029
2030 fn end_within(&self, index: usize) -> Result<u32> {
2032 let run = index / TEXT_OFFSET_RUN;
2033 let bytes = self
2034 .offsets
2035 .get(run * TEXT_OFFSET_RUN / 8 * self.offset_bits..)
2036 .ok_or_else(|| invalid("global dictionary offsets are short"))?;
2037 let end = bitpack::tail_at(bytes, self.offset_bits, index % TEXT_OFFSET_RUN)
2038 .map_err(|_| invalid("global dictionary offsets are short"))?;
2039 u32::try_from(end).map_err(|_| invalid("global dictionary offset is past the payload"))
2040 }
2041
2042 fn ends_within(&self, first: usize, last: usize) -> Result<Vec<u64>> {
2055 let mut ends = Vec::with_capacity(last.saturating_sub(first));
2056 let mut at = first;
2057 while at < last {
2058 let run = at / TEXT_OFFSET_RUN;
2059 let stop = ((run + 1) * TEXT_OFFSET_RUN).min(last);
2060 let held = self.values.saturating_sub(run * TEXT_OFFSET_RUN).min(TEXT_OFFSET_RUN);
2061 let bytes = self
2062 .offsets
2063 .get(run * TEXT_OFFSET_RUN / 8 * self.offset_bits..)
2064 .ok_or_else(|| invalid("global dictionary offsets are short"))?;
2065 let run_ends = bitpack::unpack_tail(bytes, self.offset_bits, held)
2066 .map_err(|_| invalid("global dictionary offsets are short"))?;
2067 let within = run_ends
2068 .get(at % TEXT_OFFSET_RUN..stop - run * TEXT_OFFSET_RUN)
2069 .ok_or_else(|| invalid("global dictionary offsets are short"))?;
2070 ends.extend_from_slice(within);
2071 at = stop;
2072 }
2073 Ok(ends)
2074 }
2075
2076 fn start_within(&self, index: usize) -> Result<u32> {
2079 if index % TEXT_PAYLOAD_VALUES == 0 { Ok(0) } else { self.end_within(index - 1) }
2080 }
2081
2082 fn span_within(&self, index: usize) -> Result<(u32, u32)> {
2090 let within = index % TEXT_OFFSET_RUN;
2091 let (start, end) = if within == 0 {
2092 (self.start_within(index)?, self.end_within(index)?)
2093 } else {
2094 let run = index / TEXT_OFFSET_RUN;
2095 let bytes = self
2096 .offsets
2097 .get(run * TEXT_OFFSET_RUN / 8 * self.offset_bits..)
2098 .ok_or_else(|| invalid("global dictionary offsets are short"))?;
2099 let (start, end) = bitpack::tail_pair(bytes, self.offset_bits, within)
2100 .map_err(|_| invalid("global dictionary offsets are short"))?;
2101 let ends = u32::try_from(end)
2102 .map_err(|_| invalid("global dictionary offset is past the payload"))?;
2103 let starts = u32::try_from(start)
2104 .map_err(|_| invalid("global dictionary offset is past the payload"))?;
2105 (starts, ends)
2106 };
2107 if start > end {
2108 return Err(invalid("global dictionary value ends before it starts"));
2109 }
2110 Ok((start, end))
2111 }
2112
2113 fn rank_parts(&self, rank: usize) -> Result<(&[u8], usize)> {
2120 let slot = self
2121 .rank_blocks
2122 .get(rank / TEXT_RANK_BLOCK)
2123 .ok_or_else(|| invalid("global dictionary rank is past the order"))?;
2124 let block = slot
2125 .get_or_init(|| {
2126 let which = rank / TEXT_RANK_BLOCK;
2127 let start = if which == 0 { 0 } else { self.rank_ends[which - 1] };
2128 let end = self.rank_ends[which];
2129 let mut bytes = vec![0; (end - start) as usize];
2130 read_at(&self.file, self.rank_at + start, &mut bytes)?;
2131 if checksum(&bytes)
2132 != *self
2133 .rank_hashes
2134 .get(rank / TEXT_RANK_BLOCK)
2135 .ok_or_else(|| invalid("global dictionary rank block has no checksum"))?
2136 {
2137 return Err(invalid("global dictionary rank checksum differs"));
2138 }
2139 Ok(bytes)
2140 })
2141 .as_ref()
2142 .map_err(Clone::clone)?;
2143 Ok((block.as_slice(), rank % TEXT_RANK_BLOCK))
2144 }
2145
2146 fn head_at(&self, rank: usize) -> Result<u64> {
2148 let (block, within) = self.rank_parts(rank)?;
2149 let (base, width, packed) = rank_heads(block)?;
2150 let above = bitpack::tail_at(packed, width, within)
2151 .map_err(|_| invalid("global dictionary rank block is short of heads"))?;
2152 Ok(base.wrapping_add(above))
2153 }
2154
2155 fn rank_codes<'block>(&self, block: &'block [u8], count: usize) -> Result<&'block [u8]> {
2157 let (_, width, packed) = rank_heads(block)?;
2158 packed
2159 .get(bitpack::tail_len(count, width)..)
2160 .ok_or_else(|| invalid("global dictionary rank block is short of codes"))
2161 }
2162
2163 fn rank_block_len(&self, rank: usize) -> usize {
2165 let first = rank / TEXT_RANK_BLOCK * TEXT_RANK_BLOCK;
2166 TEXT_RANK_BLOCK.min(self.ranks - first)
2167 }
2168}
2169
2170fn rank_heads(block: &[u8]) -> Result<(u64, usize, &[u8])> {
2172 let header = block
2173 .get(..RANK_BLOCK_HEADER)
2174 .ok_or_else(|| invalid("global dictionary rank block is short"))?;
2175 let base = u64::from_le_bytes(header[..8].try_into().expect("eight bytes"));
2176 let width = header[8] as usize;
2177 if width > 64 {
2178 return Err(invalid("global dictionary rank block packs heads past a word"));
2179 }
2180 Ok((base, width, &block[RANK_BLOCK_HEADER..]))
2181}
2182
2183fn offset_width(offsets: &[u32]) -> usize {
2190 let values = offsets.len() - 1;
2191 let mut span = 0;
2192 for first in (0..values).step_by(TEXT_PAYLOAD_VALUES) {
2193 let last = (first + TEXT_PAYLOAD_VALUES).min(values);
2194 span = span.max(offsets[last] - offsets[first]);
2195 }
2196 (u32::BITS - span.leading_zeros()) as usize
2197}
2198
2199fn offset_bytes(values: usize, bits: usize) -> usize {
2202 let full = values / TEXT_OFFSET_RUN;
2203 let rest = values % TEXT_OFFSET_RUN;
2204 full * TEXT_OFFSET_RUN / 8 * bits + bitpack::tail_len(rest, bits)
2205}
2206
2207fn encode_offsets(offsets: &[u32], bits: usize, out: &mut Vec<u8>) -> Result<()> {
2209 let values = offsets.len() - 1;
2210 let mut run = Vec::with_capacity(TEXT_OFFSET_RUN);
2211 for first in (0..values).step_by(TEXT_OFFSET_RUN) {
2212 let last = (first + TEXT_OFFSET_RUN).min(values);
2213 let base = offsets[first / TEXT_PAYLOAD_VALUES * TEXT_PAYLOAD_VALUES];
2214 run.clear();
2215 run.extend((first..last).map(|value| u64::from(offsets[value + 1] - base)));
2216 bitpack::pack_tail(&run, bits, out)
2217 .map_err(|_| invalid("global dictionary offsets do not pack"))?;
2218 }
2219 Ok(())
2220}
2221
2222fn code_width(values: usize) -> usize {
2224 match u64::try_from(values).unwrap_or(u64::MAX) {
2225 0 | 1 => 0,
2226 last => (u64::BITS - (last - 1).leading_zeros()) as usize,
2227 }
2228}
2229
2230impl TextSource for NativeText {
2231 fn len(&self) -> usize {
2232 self.values
2233 }
2234
2235 fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
2236 if index >= self.values {
2237 return Ok(None);
2238 }
2239 let (start, end) = self.span_within(index)?;
2240 if start == end {
2241 return Ok(Some(&[]));
2242 }
2243 let block = index / TEXT_PAYLOAD_VALUES;
2246 let Some(bytes) = self.payload_block(block)? else { return Ok(None) };
2247 Ok(bytes.get(start as usize..end as usize))
2248 }
2249
2250 fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
2251 if index >= self.values {
2252 return Ok(None);
2253 }
2254 let (start, end) = self.span_within(index)?;
2255 Ok(Some((end - start) as usize))
2256 }
2257
2258 fn sweep(
2271 &self,
2272 first: usize,
2273 limit: usize,
2274 body: &mut dyn FnMut(usize, &[u8]) -> Result<()>,
2275 ) -> Result<usize> {
2276 let limit = limit.min(self.values);
2277 if first >= limit {
2278 return Ok(first);
2279 }
2280 let block = first / TEXT_PAYLOAD_VALUES;
2281 let last = ((block + 1) * TEXT_PAYLOAD_VALUES).min(limit);
2282 let decoded;
2283 let bytes: &[u8] = match self.blocks.get(block).and_then(OnceLock::get) {
2284 Some(Ok(kept)) => kept,
2285 _ if self.payload_kept.load(Atomic::Relaxed) < self.keep_budget => {
2286 let kept = self
2287 .payload_block(block)?
2288 .ok_or_else(|| invalid("global dictionary block is past the payload"))?;
2289 self.payload_kept.fetch_add(kept.len(), Atomic::Relaxed);
2290 kept
2291 }
2292 _ => {
2293 decoded = self.decode_block(block)?;
2294 &decoded
2295 }
2296 };
2297 let ends = self.ends_within(first, last)?;
2298 if ends.len() != last - first {
2299 return Err(invalid("global dictionary offsets are short"));
2300 }
2301 let mut start = u64::from(self.start_within(first)?);
2302 for (index, &end) in (first..last).zip(&ends) {
2305 let value = usize::try_from(start)
2306 .ok()
2307 .zip(usize::try_from(end).ok())
2308 .and_then(|(from, to)| bytes.get(from..to))
2309 .ok_or_else(|| invalid("global dictionary value is past its block"))?;
2310 body(index, value)?;
2311 start = end;
2312 }
2313 Ok(last)
2314 }
2315
2316 fn ranks(&self) -> Option<usize> {
2317 (self.ranks > 0).then_some(self.ranks)
2318 }
2319
2320 fn below(&self, ranks: usize, wanted: &[u8]) -> Result<(usize, bool)> {
2328 let mut memo = self.searched.lock().map_err(|_| invalid("a poisoned dictionary search"))?;
2329 if let Some(&answer) = memo.get(wanted) {
2330 return Ok(answer);
2331 }
2332 let answer = search_below(self, ranks, wanted)?;
2333 if memo.len() >= TEXT_SEARCH_MEMO {
2334 memo.clear();
2335 }
2336 memo.insert(wanted.to_vec(), answer);
2337 Ok(answer)
2338 }
2339
2340 fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
2341 let settled = self.head_at(rank)?.cmp(&head(wanted));
2345 if settled != Ordering::Equal {
2346 return Ok(settled);
2347 }
2348 let code = self.code_at_rank(rank)?;
2349 let bytes = self
2350 .bytes_at(code as usize)?
2351 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
2352 Ok(bytes.cmp(wanted))
2353 }
2354
2355 fn code_at_rank(&self, rank: usize) -> Result<u32> {
2356 let (block, within) = self.rank_parts(rank)?;
2357 let codes = self.rank_codes(block, self.rank_block_len(rank))?;
2358 let code = bitpack::tail_at(codes, self.code_bits, within)
2359 .map_err(|_| invalid("global dictionary rank block is short of codes"))?;
2360 let code = u32::try_from(code)
2361 .map_err(|_| invalid("global dictionary order names a code it does not have"))?;
2362 if code as usize >= self.len() {
2363 return Err(invalid("global dictionary order names a code it does not have"));
2364 }
2365 Ok(code)
2366 }
2367
2368 fn code_ranks(&self) -> Option<&[u32]> {
2369 if self.ranks == 0 || self.ranks != self.len() {
2373 return None;
2374 }
2375 self.code_ranks
2376 .get_or_init(|| {
2377 let mut ranks = vec![u32::MAX; self.ranks];
2378 for first in (0..self.ranks).step_by(TEXT_RANK_BLOCK) {
2381 let (block, _) = self.rank_parts(first).ok()?;
2382 let count = self.rank_block_len(first);
2383 let codes = self.rank_codes(block, count).ok()?;
2384 for (within, code) in bitpack::unpack_tail(codes, self.code_bits, count)
2385 .ok()?
2386 .into_iter()
2387 .enumerate()
2388 {
2389 let code = usize::try_from(code).ok()?;
2390 *ranks.get_mut(code)? = u32::try_from(first + within).ok()?;
2391 }
2392 }
2393 if ranks.contains(&u32::MAX) {
2394 return None;
2395 }
2396 Some(ranks)
2397 })
2398 .as_deref()
2399 }
2400
2401 fn footprint(&self) -> usize {
2402 self.offsets.capacity()
2403 + self
2404 .code_ranks
2405 .get()
2406 .and_then(Option::as_ref)
2407 .map_or(0, |ranks| ranks.capacity() * size_of::<u32>())
2408 + self.rank_hashes.capacity() * size_of::<u64>()
2409 + self.rank_ends.capacity() * size_of::<u64>()
2410 + self.rank_blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
2411 + self
2412 .rank_blocks
2413 .iter()
2414 .filter_map(OnceLock::get)
2415 .filter_map(|result| result.as_ref().ok())
2416 .map(Vec::capacity)
2417 .sum::<usize>()
2418 + self.blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
2419 + self.hashes.capacity() * size_of::<u64>()
2420 + self.ends.capacity() * size_of::<u64>()
2421 + self
2422 .blocks
2423 .iter()
2424 .filter_map(OnceLock::get)
2425 .filter_map(|result| result.as_ref().ok())
2426 .map(Vec::capacity)
2427 .sum::<usize>()
2428 }
2429}
2430
2431fn places(table: &Table) -> Result<Vec<Place>> {
2433 let mut places = Vec::with_capacity(table.stripes.len().saturating_mul(STRIPE_PARTS));
2434 for (at, stripe) in table.stripes.iter().enumerate() {
2435 let index = u32::try_from(at).map_err(|_| invalid("too many stripes"))?;
2436 for (part, &rows) in stripe.parts.iter().enumerate() {
2437 places.push(Place {
2438 stripe: index,
2439 part: u32::try_from(part).map_err(|_| invalid("too many parts in a stripe"))?,
2440 rows,
2441 });
2442 }
2443 }
2444 Ok(places)
2445}
2446
2447fn read_index(file: &File, stripe: &Stripe, column: usize) -> Result<Vec<PartSpan>> {
2452 let parts = stripe.parts.len();
2453 let section = index_section(parts)?;
2454 let at = column.checked_mul(section).ok_or_else(|| invalid("index page offset overflow"))?;
2455 let end = at.checked_add(section).ok_or_else(|| invalid("index page offset overflow"))?;
2456 if end > stripe.index.length as usize {
2457 return Err(invalid("index page is shorter than its columns"));
2458 }
2459 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2460 let mut bytes = vec![0; section];
2461 let offset = stripe
2462 .index
2463 .offset
2464 .checked_add(at as u64)
2465 .ok_or_else(|| invalid("index page offset overflow"))?;
2466 read_at(file, offset, &mut bytes)?;
2467 let entries = section - size_of::<u64>();
2468 let stored = u64::from_le_bytes(bytes[entries..].try_into().expect("eight bytes"));
2469 if checksum(&bytes[..entries]) != stored {
2470 return Err(invalid(&format!(
2473 "index page section checksum differs, column {column} of {parts} parts at {offset}, \
2474 wanted {stored:016x} and got {:016x}",
2475 checksum(&bytes[..entries]),
2476 )));
2477 }
2478 let mut spans = Vec::with_capacity(parts);
2479 let mut start = 0_usize;
2480 for part in 0..parts {
2481 let at = part * INDEX_ENTRY;
2482 let length = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four bytes")) as usize;
2483 let hash = u64::from_le_bytes(bytes[at + 4..at + 12].try_into().expect("eight bytes"));
2484 spans.push(PartSpan { start, length, hash });
2485 start = start.checked_add(length).ok_or_else(|| invalid("column page length overflow"))?;
2486 }
2487 if start != page.length as usize {
2488 return Err(invalid("column page length differs from its index"));
2489 }
2490 Ok(spans)
2491}
2492
2493fn part_bytes(page: &[u8], span: PartSpan) -> Result<&[u8]> {
2495 let end = span.start.checked_add(span.length).ok_or_else(|| invalid("part range overflow"))?;
2496 page.get(span.start..end).ok_or_else(|| invalid("part exceeds its column page"))
2497}
2498
2499fn remember(cached: &mut Cached, held: &CachedColumn, kept: usize) {
2504 if let Some(slot) = cached.index.get_mut(held.stripe) {
2505 if slot.is_none() {
2506 *slot = Some(Arc::clone(&held.index));
2507 }
2508 }
2509 let Some(page) = held.page.clone() else { return };
2510 let Some(slot) = cached.pages.get_mut(held.stripe) else { return };
2511 if slot.is_none() {
2512 cached.order.push_back(held.stripe);
2513 }
2514 *slot = Some(page);
2515 while cached.order.len() > kept.max(1) {
2516 let Some(oldest) = cached.order.pop_front() else { break };
2517 if let Some(slot) = cached.pages.get_mut(oldest) {
2518 *slot = None;
2519 }
2520 }
2521}
2522
2523#[derive(Debug, Clone)]
2532pub struct Catalog {
2533 file: Arc<File>,
2534 size: u64,
2535 entries: Arc<Vec<Entry>>,
2536 opening: Opening,
2537}
2538
2539impl Catalog {
2540 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
2546 let (file, size, _, bytes, opening) = slot_bytes(path)?;
2547 let entries = decode_catalog(&bytes, size)?;
2548 Ok(Self { file: Arc::new(file), size, entries: Arc::new(entries), opening })
2549 }
2550
2551 pub fn names(&self) -> impl ExactSizeIterator<Item = &str> {
2553 self.entries.iter().map(|entry| entry.name.as_str())
2554 }
2555
2556 #[must_use]
2558 pub fn len(&self) -> usize {
2559 self.entries.len()
2560 }
2561
2562 #[must_use]
2564 pub fn is_empty(&self) -> bool {
2565 self.entries.is_empty()
2566 }
2567
2568 pub fn table(&self, name: &str) -> Result<Reader> {
2574 let entry = self
2575 .entries
2576 .iter()
2577 .find(|entry| entry.name == name)
2578 .ok_or_else(|| invalid(&format!("the file holds no table called {name}")))?;
2579 let mut bytes = vec![0; entry.directory.length as usize];
2580 read_at(&self.file, entry.directory.offset, &mut bytes)?;
2581 if checksum(&bytes) != entry.directory.hash {
2582 return Err(invalid(&format!("the directory of table {name} does not checksum")));
2583 }
2584 let mut opening = self.opening;
2585 opening.reads += 1;
2586 opening.bytes += u64::from(entry.directory.length);
2587 Reader::build(
2588 Arc::clone(&self.file),
2589 self.size,
2590 decode_directory(&bytes, self.size)?,
2591 u64::from(entry.directory.length),
2592 opening,
2593 )
2594 }
2595}
2596
2597fn slot_offset(generation: u64) -> u64 {
2602 16 + (generation - 1) % 2 * SLOT_BYTES as u64
2603}
2604
2605fn slot_bytes(path: impl AsRef<Path>) -> Result<(File, u64, Slot, Vec<u8>, Opening)> {
2610 let mut file = File::open(path).map_err(io)?;
2611 let size = file.metadata().map_err(io)?.len();
2612 if size < HEADER {
2613 return Err(invalid("file is shorter than its header"));
2614 }
2615 let mut header = [0; HEADER as usize];
2616 file.read_exact(&mut header).map_err(io)?;
2617 let mut opening = Opening { reads: 1, bytes: HEADER };
2618 let version = u32::from_le_bytes([header[8], header[9], header[10], header[11]]);
2619 if &header[..8] != MAGIC {
2624 return Err(invalid("the header does not begin with a rudb native magic"));
2625 }
2626 if version != FORMAT {
2627 return Err(invalid(&format!(
2628 "the file is format {version} and this build reads format {FORMAT}, so it has to \
2629 be written again"
2630 )));
2631 }
2632 let mut selected = None;
2633 for start in [16, 16 + SLOT_BYTES] {
2634 let slot = Slot::read(&header[start..start + SLOT_BYTES]);
2635 if slot.generation == 0 || slot.length == 0 || slot.length as usize > MAX_DIRECTORY {
2636 continue;
2637 }
2638 let Some(end) = slot.offset.checked_add(u64::from(slot.length)) else { continue };
2639 if slot.offset < HEADER || end > size {
2640 continue;
2641 }
2642 let mut bytes = vec![0; slot.length as usize];
2643 file.seek(SeekFrom::Start(slot.offset)).map_err(io)?;
2644 file.read_exact(&mut bytes).map_err(io)?;
2645 opening.reads += 1;
2646 opening.bytes += u64::from(slot.length);
2647 if checksum(&bytes) == slot.hash
2648 && selected
2649 .as_ref()
2650 .is_none_or(|(old, _): &(Slot, Vec<u8>)| old.generation < slot.generation)
2651 {
2652 selected = Some((slot, bytes));
2653 }
2654 }
2655 let (slot, bytes) = selected.ok_or_else(|| invalid("no committed directory slot is valid"))?;
2656 Ok((file, size, slot, bytes, opening))
2657}
2658
2659impl Reader {
2660 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
2667 let catalog = Catalog::open(path)?;
2668 let mut names = catalog.names();
2669 let name = names.next().ok_or_else(|| invalid("the file holds no table"))?.to_string();
2670 if names.next().is_some() {
2671 return Err(invalid(
2672 "the file holds more than one table, so it has to be opened by name",
2673 ));
2674 }
2675 catalog.table(&name)
2676 }
2677
2678 fn build(
2680 file: Arc<File>,
2681 size: u64,
2682 table: Table,
2683 directory: u64,
2684 opening: Opening,
2685 ) -> Result<Self> {
2686 let places = places(&table)?;
2687 let dictionaries = (0..table.fields.len()).map(|_| OnceLock::new()).collect();
2688 let table_fields = table.fields.len();
2689 let stripes = table.stripes.len();
2690 let cache = (0..table.fields.len())
2691 .map(|_| {
2692 Mutex::new(Cached {
2693 pages: (0..stripes).map(|_| None).collect(),
2694 index: (0..stripes).map(|_| None).collect(),
2695 ..Cached::default()
2696 })
2697 })
2698 .collect::<Vec<_>>();
2699 let sieves: Vec<Vec<SieveSlot>> = (0..table.fields.len())
2700 .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
2701 .collect();
2702 let part_ranges: Vec<Vec<RangeSlot>> = (0..table.fields.len())
2703 .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
2704 .collect();
2705 Ok(Self {
2706 file,
2707 table: Arc::new(table),
2708 dictionaries: Arc::new(dictionaries),
2709 loading: Arc::new((0..table_fields).map(|_| Mutex::new(())).collect()),
2710 opened: Arc::new(AtomicUsize::new(0)),
2711 sieves: Arc::new(sieves),
2712 part_ranges: Arc::new(part_ranges),
2713 places: Arc::new(places),
2714 cache: Arc::new(cache),
2715 pages: Arc::new(AtomicUsize::new(0)),
2716 indexes: Arc::new(AtomicUsize::new(0)),
2717 kept: Arc::new(AtomicUsize::new(CACHED_STRIPES_PER_COLUMN)),
2718 size,
2719 directory,
2720 opening,
2721 })
2722 }
2723
2724 #[must_use]
2731 pub fn reads(&self) -> Reads {
2732 Reads {
2733 opening: self.opening,
2734 pages: self.pages.load(Atomic::Relaxed),
2735 indexes: self.indexes.load(Atomic::Relaxed),
2736 dictionaries: self.opened.load(Atomic::Relaxed),
2737 }
2738 }
2739
2740 #[must_use]
2745 pub fn layout(&self) -> Layout {
2746 let table = &self.table;
2747 let stripes = table.stripes.as_slice();
2748 let columns = table
2749 .fields
2750 .iter()
2751 .enumerate()
2752 .map(|(at, field)| ColumnLayout {
2753 name: field.name.clone(),
2754 kind: field.ty.to_string(),
2755 pages: sum(stripes.iter().map(|stripe| span_bytes(&stripe.pages, at))),
2756 memberships: sum(stripes.iter().map(|stripe| page_bytes(&stripe.memberships, at))),
2757 sieves: sum(stripes.iter().map(|stripe| page_bytes(&stripe.sieves, at))),
2758 part_ranges: sum(stripes.iter().map(|stripe| page_bytes(&stripe.part_ranges, at))),
2759 dictionary: page_bytes(&table.dictionaries, at),
2760 })
2761 .collect();
2762 Layout {
2763 file: self.size,
2764 rows: table.rows,
2765 stripes: stripes.len(),
2766 parts: self.places.len(),
2767 columns,
2768 indexes: sum(stripes.iter().map(|stripe| u64::from(stripe.index.length))),
2769 directory: self.directory,
2770 header: HEADER,
2771 }
2772 }
2773
2774 #[must_use]
2776 pub fn parts(&self) -> usize {
2777 self.places.len()
2778 }
2779
2780 #[must_use]
2787 pub fn stripe_parts(&self) -> Vec<std::ops::Range<usize>> {
2788 let mut runs = Vec::with_capacity(self.table.stripes.len());
2789 let mut start = 0;
2790 for stripe in &self.table.stripes {
2791 let end = start + stripe.parts.len();
2792 runs.push(start..end);
2793 start = end;
2794 }
2795 runs
2796 }
2797
2798 #[must_use]
2803 pub fn stripe_rows(&self, stripe: usize) -> usize {
2804 self.table.stripes.get(stripe).map_or(0, |held| held.rows)
2805 }
2806
2807 pub fn keep_stripes(&self, stripes: usize) {
2814 self.kept.fetch_max(stripes, Atomic::Relaxed);
2815 }
2816
2817 #[must_use]
2819 pub fn part_rows(&self, at: usize) -> usize {
2820 self.places.get(at).map_or(0, |place| place.rows as usize)
2821 }
2822
2823 #[must_use]
2825 pub fn table(&self) -> &Table {
2826 &self.table
2827 }
2828
2829 pub fn top_frequencies(&self, column: usize, top: usize) -> Result<Option<Vec<(Value, u64)>>> {
2838 let field = self
2839 .table
2840 .fields
2841 .get(column)
2842 .ok_or_else(|| invalid("frequency column index out of range"))?;
2843 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2844 return Ok(None);
2845 };
2846 if top == 0 || summary.entries.len() < top {
2847 return Ok(None);
2848 }
2849 let boundary = summary.entries[top - 1].count;
2850 if boundary <= summary.omitted_max {
2851 return Ok(None);
2852 }
2853 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
2854 }
2855
2856 pub fn exact_frequencies(&self, column: usize) -> Result<Option<Vec<(Value, u64)>>> {
2876 let Some(prefix) = self.frequency_prefix(column)? else {
2877 return Ok(None);
2878 };
2879 Ok((prefix.omitted_max == 0).then_some(prefix.entries))
2880 }
2881
2882 pub fn frequency_prefix(&self, column: usize) -> Result<Option<FrequencyPrefix>> {
2905 let field = self
2906 .table
2907 .fields
2908 .get(column)
2909 .ok_or_else(|| invalid("frequency column index out of range"))?;
2910 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2911 return Ok(None);
2912 };
2913 let entries = self.decode_frequencies(column, &field.ty, &summary.entries)?;
2914 Ok(Some(FrequencyPrefix { entries, omitted_max: summary.omitted_max }))
2915 }
2916
2917 fn decode_frequencies(
2919 &self,
2920 column: usize,
2921 ty: &LogicalType,
2922 entries: &[FrequencyEntry],
2923 ) -> Result<Vec<(Value, u64)>> {
2924 let dictionary = if *ty == LogicalType::Varchar { self.dictionary(column)? } else { None };
2925 let mut out = Vec::with_capacity(entries.len());
2926 for entry in entries {
2927 let value = match entry.value {
2928 FrequencyValue::Null => Value::Null,
2929 FrequencyValue::Integer(value) => match *ty {
2930 LogicalType::TinyInt => Value::TinyInt(
2931 i8::try_from(value)
2932 .map_err(|_| invalid("frequency TINYINT is out of range"))?,
2933 ),
2934 LogicalType::UTinyInt => Value::UTinyInt(
2935 u8::try_from(value)
2936 .map_err(|_| invalid("frequency UTINYINT is out of range"))?,
2937 ),
2938 LogicalType::USmallInt => Value::USmallInt(
2939 u16::try_from(value)
2940 .map_err(|_| invalid("frequency USMALLINT is out of range"))?,
2941 ),
2942 LogicalType::UInteger => Value::UInteger(
2943 u32::try_from(value)
2944 .map_err(|_| invalid("frequency UINTEGER is out of range"))?,
2945 ),
2946 LogicalType::UBigInt => Value::UBigInt(
2947 u64::try_from(value)
2948 .map_err(|_| invalid("frequency UBIGINT is out of range"))?,
2949 ),
2950 LogicalType::SmallInt => Value::SmallInt(
2951 i16::try_from(value)
2952 .map_err(|_| invalid("frequency SMALLINT is out of range"))?,
2953 ),
2954 LogicalType::Integer => Value::Integer(
2955 i32::try_from(value)
2956 .map_err(|_| invalid("frequency INTEGER is out of range"))?,
2957 ),
2958 LogicalType::BigInt => Value::BigInt(
2959 i64::try_from(value)
2960 .map_err(|_| invalid("frequency BIGINT is out of range"))?,
2961 ),
2962 LogicalType::Date => Value::Date(
2963 i32::try_from(value)
2964 .map_err(|_| invalid("frequency DATE is out of range"))?,
2965 ),
2966 LogicalType::Timestamp => Value::Timestamp(
2967 i64::try_from(value)
2968 .map_err(|_| invalid("frequency TIMESTAMP is out of range"))?,
2969 ),
2970 _ => return Err(invalid("integer frequency belongs to another type")),
2971 },
2972 FrequencyValue::Code(code) => dictionary
2973 .as_ref()
2974 .ok_or_else(|| invalid("frequency code has no dictionary"))?
2975 .try_value_at(code as usize)?,
2976 };
2977 out.push((value, entry.count));
2978 }
2979 Ok(out)
2980 }
2981
2982 pub fn frequency_occurrences(&self, column: usize) -> Result<Option<FrequencyOccurrences>> {
2992 self.table
2993 .fields
2994 .get(column)
2995 .ok_or_else(|| invalid("frequency column index out of range"))?;
2996 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2997 return Ok(None);
2998 };
2999 if summary.ordinals.is_empty() {
3000 return Ok(None);
3001 }
3002 Ok(Some(FrequencyOccurrences {
3003 omitted_max: summary.omitted_max,
3004 ordinals: summary.ordinals.clone(),
3005 }))
3006 }
3007
3008 pub fn distinct_values(&self, column: usize) -> Result<Option<u64>> {
3032 self.table
3033 .distincts
3034 .get(column)
3035 .copied()
3036 .ok_or_else(|| invalid("distinct column index out of range"))
3037 }
3038
3039 pub fn null_count(&self, column: usize) -> Result<u64> {
3050 if column >= self.table.fields.len() {
3051 return Err(invalid("null count column index out of range"));
3052 }
3053 let mut nulls = 0_u64;
3054 for stripe in &self.table.stripes {
3055 let range = stripe
3056 .zone
3057 .column(column)
3058 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
3059 nulls = nulls
3060 .checked_add(range.nulls as u64)
3061 .ok_or_else(|| invalid("null count overflow"))?;
3062 }
3063 Ok(nulls)
3064 }
3065
3066 pub fn text_extremes(&self, column: usize) -> Result<Option<(Value, Value)>> {
3081 if self.null_count(column)? > 0 {
3082 return Ok(None);
3083 }
3084 let Some(dictionary) = self.dictionary(column)? else { return Ok(None) };
3085 let Some(ranks) = dictionary.ranks() else { return Ok(None) };
3086 if ranks == 0 {
3087 return Ok(None);
3088 }
3089 let low = text_at_rank(&dictionary, 0)?;
3090 let high = text_at_rank(&dictionary, ranks - 1)?;
3091 Ok(Some((low, high)))
3092 }
3093
3094 pub fn exact_extremes(&self, column: usize) -> Result<Option<(Bound, Bound)>> {
3117 if column >= self.table.fields.len() {
3118 return Err(invalid("extremes column index out of range"));
3119 }
3120 let mut low: Option<Bound> = None;
3121 let mut high: Option<Bound> = None;
3122 for stripe in &self.table.stripes {
3123 let range = stripe
3124 .zone
3125 .column(column)
3126 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
3127 if !range.exact {
3128 return Ok(None);
3129 }
3130 let (Some(small), Some(large)) = (range.low.as_ref(), range.high.as_ref()) else {
3135 if stripe.rows > range.nulls {
3136 return Ok(None);
3137 }
3138 continue;
3139 };
3140 low = Some(low.map_or_else(|| small.clone(), |held| held.smaller(small.clone())));
3141 high = Some(high.map_or_else(|| large.clone(), |held| held.larger(large.clone())));
3142 }
3143 Ok(low.zip(high))
3144 }
3145
3146 pub fn exact_sum(&self, column: usize) -> Result<Option<(i128, u64)>> {
3159 if column >= self.table.fields.len() {
3160 return Err(invalid("sum column index out of range"));
3161 }
3162 let mut total = 0_i128;
3163 let mut rows = 0_u64;
3164 for stripe in &self.table.stripes {
3165 let range = stripe
3166 .zone
3167 .column(column)
3168 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
3169 let Some(part) = range.sum else { return Ok(None) };
3170 let Some(sum) = total.checked_add(part) else { return Ok(None) };
3171 total = sum;
3172 rows = rows.saturating_add(stripe.rows as u64 - range.nulls as u64);
3173 }
3174 Ok(Some((total, rows)))
3175 }
3176
3177 fn dictionary(&self, column: usize) -> Result<Option<Arc<Vector>>> {
3186 let Some(page) = self.table.dictionaries[column] else { return Ok(None) };
3187 if let Some(dictionary) = self.dictionaries[column].get() {
3188 return Ok(Some(Arc::clone(dictionary)));
3189 }
3190 let _queued = self.loading[column].lock().map_err(|_| invalid("a poisoned dictionary"))?;
3191 if let Some(dictionary) = self.dictionaries[column].get() {
3192 return Ok(Some(Arc::clone(dictionary)));
3193 }
3194 self.opened.fetch_add(1, Atomic::Relaxed);
3195 let dictionary = Arc::new(open_global_dictionary(
3196 Arc::clone(&self.file),
3197 page,
3198 &self.table.fields[column].ty,
3199 TEXT_KEEP_BUDGET,
3200 )?);
3201 let _ = self.dictionaries[column].set(Arc::clone(&dictionary));
3202 Ok(Some(dictionary))
3203 }
3204
3205 pub fn read(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
3214 self.read_impl(part, columns, true)
3215 }
3216
3217 pub fn read_sparse(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
3227 self.read_impl(part, columns, false)
3228 }
3229
3230 pub fn skips_codes(&self, part: usize, column: usize, candidates: &[u32]) -> Result<bool> {
3237 if candidates.is_empty() {
3238 return Ok(true);
3239 }
3240 if candidates.windows(2).any(|pair| pair[0] >= pair[1]) {
3241 return Err(Error::internal("native code candidates are not sorted and unique"));
3242 }
3243 let stripe = self.stripe_of(part)?;
3244 let Some(page) = stripe.memberships.get(column).copied().flatten() else {
3245 return Ok(false);
3246 };
3247 let mut bytes = vec![0; page.length as usize];
3248 read_at(&self.file, page.offset, &mut bytes)?;
3249 if checksum(&bytes) != page.hash {
3250 return Err(invalid("membership page checksum differs"));
3251 }
3252 let codes = decode_membership(&bytes)?;
3253 let mut left = 0;
3254 let mut right = 0;
3255 while left < codes.len() && right < candidates.len() {
3256 match codes[left].cmp(&candidates[right]) {
3257 Ordering::Less => left += 1,
3258 Ordering::Greater => right += 1,
3259 Ordering::Equal => return Ok(false),
3260 }
3261 }
3262 Ok(true)
3263 }
3264
3265 fn stripe_of(&self, part: usize) -> Result<&Stripe> {
3266 let place = self.places.get(part).ok_or_else(|| invalid("part index out of range"))?;
3267 self.table
3268 .stripes
3269 .get(place.stripe as usize)
3270 .ok_or_else(|| invalid("stripe index out of range"))
3271 }
3272
3273 fn held(&self, at: usize, stripe: &Stripe, column: usize, whole: bool) -> Result<CachedColumn> {
3290 let cache = self.cache.get(column).ok_or_else(|| invalid("column index out of range"))?;
3291 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
3292 let known = cached.index.get(at).and_then(Clone::clone);
3293 let page = cached.pages.get(at).and_then(Clone::clone);
3294 if let Some(index) = known.clone() {
3295 if !whole || page.is_some() {
3296 return Ok(CachedColumn { stripe: at, index, page });
3297 }
3298 }
3299 if cached.loading.contains(&at) {
3300 drop(cached);
3301 if let Some(index) = known {
3305 return Ok(CachedColumn { stripe: at, index, page: None });
3306 }
3307 let held = self.page_of(stripe, column, at, false, None)?;
3308 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
3309 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
3310 return Ok(held);
3311 }
3312 cached.loading.push(at);
3313 drop(cached);
3314
3315 let read = self.page_of(stripe, column, at, whole, known);
3316
3317 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
3321 if let Some(position) = cached.loading.iter().position(|loading| *loading == at) {
3322 cached.loading.remove(position);
3323 }
3324 let held = read?;
3325 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
3326 Ok(held)
3327 }
3328
3329 fn page_of(
3335 &self,
3336 stripe: &Stripe,
3337 column: usize,
3338 at: usize,
3339 whole: bool,
3340 known: Option<Arc<Vec<PartSpan>>>,
3341 ) -> Result<CachedColumn> {
3342 let index = match known {
3343 Some(index) => index,
3344 None => {
3345 self.indexes.fetch_add(1, Atomic::Relaxed);
3346 Arc::new(read_index(&self.file, stripe, column)?)
3347 }
3348 };
3349 let page = if whole {
3350 self.pages.fetch_add(1, Atomic::Relaxed);
3351 let span = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
3352 let mut bytes = vec![0; span.length as usize];
3353 read_at(&self.file, span.offset, &mut bytes)?;
3354 Some(Arc::new(bytes))
3355 } else {
3356 None
3357 };
3358 Ok(CachedColumn { stripe: at, index, page })
3359 }
3360
3361 fn read_impl(&self, at: usize, columns: &[usize], whole: bool) -> Result<Chunk> {
3362 let place = *self.places.get(at).ok_or_else(|| invalid("part index out of range"))?;
3363 let index = place.stripe as usize;
3364 let stripe =
3365 self.table.stripes.get(index).ok_or_else(|| invalid("stripe index out of range"))?;
3366 let rows = place.rows as usize;
3367 let mut picked = Vec::with_capacity(columns.len());
3368 for &column in columns {
3369 let field = self
3370 .table
3371 .fields
3372 .get(column)
3373 .ok_or_else(|| invalid("column index out of range"))?;
3374 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
3375 let held = self.held(index, stripe, column, whole)?;
3376 let span = *held
3377 .index
3378 .get(place.part as usize)
3379 .ok_or_else(|| invalid("part index out of range"))?;
3380 let owned;
3381 let bytes = match &held.page {
3382 Some(held) => part_bytes(held, span)?,
3383 None => {
3384 let offset = page
3385 .offset
3386 .checked_add(span.start as u64)
3387 .ok_or_else(|| invalid("part range overflow"))?;
3388 let mut bytes = vec![0; span.length];
3389 read_at(&self.file, offset, &mut bytes)?;
3390 owned = bytes;
3391 &owned
3392 }
3393 };
3394 if checksum(bytes) != span.hash {
3395 return Err(invalid(&format!(
3396 "column page checksum differs, column {column} part {} at {}+{} of {} bytes, \
3397 wanted {:016x} and got {:016x}",
3398 place.part,
3399 page.offset,
3400 span.start,
3401 span.length,
3402 span.hash,
3403 checksum(bytes),
3404 )));
3405 }
3406 let dictionary = self.dictionary(column)?;
3407 picked.push(decode(&field.ty, rows, bytes, dictionary)?.into_pages());
3413 }
3414 Chunk::with_rows(picked, rows)
3415 }
3416
3417 #[must_use]
3433 pub fn skips(&self, part: usize, probes: &[Probe]) -> bool {
3434 let Some(place) = self.places.get(part).copied() else { return false };
3435 let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
3436 if stripe.zone.skips(probes) {
3437 return true;
3438 }
3439 probes.iter().any(|probe| self.outside(place, probe) || self.sifted(place, probe))
3440 }
3441
3442 fn outside(&self, place: Place, probe: &Probe) -> bool {
3448 match self.stripe_part_ranges(place.stripe as usize, probe.column) {
3449 Some(ranges) => ranges
3450 .get(place.part as usize)
3451 .is_some_and(|range| range.excludes(probe.op, &probe.value)),
3452 None => false,
3453 }
3454 }
3455
3456 fn stripe_part_ranges(&self, stripe: usize, column: usize) -> Option<&[Range]> {
3462 let slot = self.part_ranges.get(column)?.get(stripe)?;
3463 if let Some(held) = slot.get() {
3464 return Some(held);
3465 }
3466 let page = self.table.stripes.get(stripe)?.part_ranges.get(column).copied().flatten()?;
3467 let mut bytes = vec![0; page.length as usize];
3468 read_at(&self.file, page.offset, &mut bytes).ok()?;
3469 if checksum(&bytes) != page.hash {
3470 return None;
3471 }
3472 let ranges = Arc::new(decode_part_ranges(&bytes).ok()?);
3473 let _ = slot.set(ranges);
3474 slot.get().map(|held| held.as_slice())
3475 }
3476
3477 #[must_use]
3494 pub fn certain(&self, part: usize, probes: &[Probe]) -> bool {
3495 let Some(place) = self.places.get(part).copied() else { return false };
3496 let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
3497 if stripe.zone.certain(probes) {
3498 return true;
3499 }
3500 probes
3501 .iter()
3502 .all(|probe| stripe.zone.certain(slice::from_ref(probe)) || self.inside(place, probe))
3503 }
3504
3505 fn inside(&self, place: Place, probe: &Probe) -> bool {
3511 match self.stripe_part_ranges(place.stripe as usize, probe.column) {
3512 Some(ranges) => ranges
3513 .get(place.part as usize)
3514 .is_some_and(|range| range.certain(probe.op, &probe.value)),
3515 None => false,
3516 }
3517 }
3518
3519 #[must_use]
3530 pub fn stripe_skips(&self, stripe: usize, probes: &[Probe]) -> bool {
3531 self.table.stripes.get(stripe).is_some_and(|held| held.zone.skips(probes))
3532 }
3533
3534 fn sifted(&self, place: Place, probe: &Probe) -> bool {
3540 if probe.op != Op::Equal {
3541 return false;
3542 }
3543 match self.stripe_sieves(place.stripe as usize, probe.column) {
3544 Some(sieves) => sieves
3545 .get(place.part as usize)
3546 .and_then(Option::as_ref)
3547 .is_some_and(|sieve| sieve.excludes(&probe.value)),
3548 None => false,
3549 }
3550 }
3551
3552 fn stripe_sieves(&self, stripe: usize, column: usize) -> Option<&[Option<Sieve>]> {
3559 let slot = self.sieves.get(column)?.get(stripe)?;
3560 if let Some(held) = slot.get() {
3561 return Some(held);
3562 }
3563 let page = self.table.stripes.get(stripe)?.sieves.get(column).copied().flatten()?;
3564 let mut bytes = vec![0; page.length as usize];
3565 read_at(&self.file, page.offset, &mut bytes).ok()?;
3566 if checksum(&bytes) != page.hash {
3567 return None;
3568 }
3569 let sieves = Arc::new(decode_sieves(&bytes).ok()?);
3570 let _ = slot.set(sieves);
3571 slot.get().map(|held| held.as_slice())
3572 }
3573}
3574
3575fn text_at_rank(dictionary: &Vector, rank: usize) -> Result<Value> {
3577 let code = dictionary.code_at_rank(rank)? as usize;
3578 let text = dictionary
3579 .try_text_at(code)?
3580 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
3581 Ok(Value::Varchar(text.into()))
3582}
3583
3584#[cfg(unix)]
3589fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
3590 use std::os::unix::fs::FileExt;
3591 while !bytes.is_empty() {
3592 let written = file.write_at(bytes, offset).map_err(io)?;
3593 if written == 0 {
3594 return Err(invalid("a write to the native file wrote nothing"));
3595 }
3596 offset += written as u64;
3597 bytes = &bytes[written..];
3598 }
3599 Ok(())
3600}
3601
3602#[cfg(windows)]
3604fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
3605 use std::os::windows::fs::FileExt;
3606 while !bytes.is_empty() {
3607 let written = file.seek_write(bytes, offset).map_err(io)?;
3608 if written == 0 {
3609 return Err(invalid("a write to the native file wrote nothing"));
3610 }
3611 offset += written as u64;
3612 bytes = &bytes[written..];
3613 }
3614 Ok(())
3615}
3616
3617#[cfg(not(any(unix, windows)))]
3619fn write_at(file: &File, offset: u64, bytes: &[u8]) -> Result<()> {
3620 use std::io::Write;
3621 let mut file = file.try_clone().map_err(io)?;
3622 file.seek(SeekFrom::Start(offset)).map_err(io)?;
3623 file.write_all(bytes).map_err(io)
3624}
3625
3626#[cfg(unix)]
3636fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
3637 use std::os::unix::fs::FileExt;
3638 while !bytes.is_empty() {
3639 let read = file.read_at(bytes, offset).map_err(io)?;
3640 if read == 0 {
3641 return Err(invalid("column page ends before its declared length"));
3642 }
3643 offset += read as u64;
3644 bytes = &mut bytes[read..];
3645 }
3646 Ok(())
3647}
3648
3649#[cfg(windows)]
3655fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
3656 use std::os::windows::fs::FileExt;
3657 while !bytes.is_empty() {
3658 let read = file.seek_read(bytes, offset).map_err(io)?;
3659 if read == 0 {
3660 return Err(invalid("column page ends before its declared length"));
3661 }
3662 offset += read as u64;
3663 bytes = &mut bytes[read..];
3664 }
3665 Ok(())
3666}
3667
3668#[cfg(not(any(unix, windows)))]
3673fn read_at(file: &File, offset: u64, bytes: &mut [u8]) -> Result<()> {
3674 let mut file = file.try_clone().map_err(io)?;
3675 file.seek(SeekFrom::Start(offset)).map_err(io)?;
3676 file.read_exact(bytes).map_err(io)
3677}
3678
3679fn type_tag(ty: &LogicalType) -> Result<u8> {
3680 match ty {
3681 LogicalType::SmallInt => Ok(1),
3682 LogicalType::Integer => Ok(2),
3683 LogicalType::BigInt => Ok(3),
3684 LogicalType::Varchar => Ok(4),
3685 LogicalType::Date => Ok(5),
3686 LogicalType::Timestamp => Ok(6),
3687 LogicalType::Boolean => Ok(7),
3688 LogicalType::TinyInt => Ok(8),
3689 LogicalType::UTinyInt => Ok(9),
3690 LogicalType::USmallInt => Ok(10),
3691 LogicalType::UInteger => Ok(11),
3692 LogicalType::UBigInt => Ok(12),
3693 LogicalType::Decimal { .. } => Ok(13),
3694 _ => Err(Error::not_implemented(format!("native storage for {ty}"))),
3695 }
3696}
3697
3698fn put_type(out: &mut Vec<u8>, ty: &LogicalType) -> Result<()> {
3704 out.push(type_tag(ty)?);
3705 if let LogicalType::Decimal { width, scale } = ty {
3706 out.push(*width);
3707 out.push(*scale);
3708 }
3709 Ok(())
3710}
3711
3712fn read_type(cur: &mut Cursor<'_>) -> Result<LogicalType> {
3714 let tag = cur.u8()?;
3715 if tag == 13 {
3716 let width = cur.u8()?;
3717 let scale = cur.u8()?;
3718 return LogicalType::decimal(width, scale)
3719 .map_err(|_| invalid("decimal column width and scale are not a decimal"));
3720 }
3721 tag_type(tag)
3722}
3723
3724fn tag_type(tag: u8) -> Result<LogicalType> {
3725 match tag {
3726 1 => Ok(LogicalType::SmallInt),
3727 2 => Ok(LogicalType::Integer),
3728 3 => Ok(LogicalType::BigInt),
3729 4 => Ok(LogicalType::Varchar),
3730 5 => Ok(LogicalType::Date),
3731 6 => Ok(LogicalType::Timestamp),
3732 7 => Ok(LogicalType::Boolean),
3733 8 => Ok(LogicalType::TinyInt),
3734 9 => Ok(LogicalType::UTinyInt),
3735 10 => Ok(LogicalType::USmallInt),
3736 11 => Ok(LogicalType::UInteger),
3737 12 => Ok(LogicalType::UBigInt),
3738 _ => Err(invalid("column type tag is unknown")),
3739 }
3740}
3741
3742fn put_u16(out: &mut Vec<u8>, value: u16) {
3743 out.extend_from_slice(&value.to_le_bytes());
3744}
3745fn put_u32(out: &mut Vec<u8>, value: u32) {
3746 out.extend_from_slice(&value.to_le_bytes());
3747}
3748fn put_u64(out: &mut Vec<u8>, value: u64) {
3749 out.extend_from_slice(&value.to_le_bytes());
3750}
3751fn put_var_u64(out: &mut Vec<u8>, mut value: u64) {
3752 while value >= 0x80 {
3753 out.push((value as u8 & 0x7f) | 0x80);
3754 value >>= 7;
3755 }
3756 out.push(value as u8);
3757}
3758
3759fn frequency_order(left: FrequencyValue, right: FrequencyValue) -> Ordering {
3760 match (left, right) {
3761 (FrequencyValue::Null, FrequencyValue::Null) => Ordering::Equal,
3762 (FrequencyValue::Null, _) => Ordering::Less,
3763 (_, FrequencyValue::Null) => Ordering::Greater,
3764 (FrequencyValue::Integer(left), FrequencyValue::Integer(right)) => left.cmp(&right),
3765 (FrequencyValue::Code(left), FrequencyValue::Code(right)) => left.cmp(&right),
3766 (FrequencyValue::Integer(_), FrequencyValue::Code(_)) => Ordering::Less,
3767 (FrequencyValue::Code(_), FrequencyValue::Integer(_)) => Ordering::Greater,
3768 }
3769}
3770
3771fn code_frequency(dictionary: &GlobalDictionary) -> FrequencySummary {
3772 let mut entries = dictionary
3773 .counts
3774 .iter()
3775 .enumerate()
3776 .filter(|(_, count)| **count != 0)
3777 .map(|(code, &count)| FrequencyEntry { value: FrequencyValue::Code(code as u32), count })
3778 .collect::<Vec<_>>();
3779 if dictionary.nulls != 0 {
3780 entries.push(FrequencyEntry { value: FrequencyValue::Null, count: dictionary.nulls });
3781 }
3782 entries.sort_unstable_by(|left, right| {
3783 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
3784 });
3785 let omitted_max = entries.get(FREQUENCY_ENTRIES).map_or(0, |entry| entry.count);
3786 entries.truncate(FREQUENCY_ENTRIES);
3787 FrequencySummary { entries, omitted_max, ordinals: Vec::new() }
3788}
3789
3790fn encode_directory(table: &Table) -> Result<Vec<u8>> {
3791 let mut out = DIRECTORY.to_vec();
3792 let name = table.name.as_bytes();
3793 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
3794 out.extend_from_slice(name);
3795 put_u16(&mut out, u16::try_from(table.fields.len()).map_err(|_| invalid("too many columns"))?);
3796 for field in &table.fields {
3797 let name = field.name.as_bytes();
3798 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?);
3799 out.extend_from_slice(name);
3800 put_type(&mut out, &field.ty)?;
3801 out.push(u8::from(field.not_null));
3802 }
3803 for dictionary in &table.dictionaries {
3804 match dictionary {
3805 None => out.push(0),
3806 Some(page) => {
3807 out.push(1);
3808 put_u64(&mut out, page.offset);
3809 put_u32(&mut out, page.length);
3810 put_u64(&mut out, page.hash);
3811 }
3812 }
3813 }
3814 for distinct in &table.distincts {
3815 match distinct {
3816 None => out.push(0),
3817 Some(count) => {
3818 out.push(1);
3819 put_u64(&mut out, *count);
3820 }
3821 }
3822 }
3823 put_u64(&mut out, u64::try_from(table.rows).map_err(|_| invalid("row count overflow"))?);
3824 put_u32(&mut out, u32::try_from(table.stripes.len()).map_err(|_| invalid("too many stripes"))?);
3825 for stripe in &table.stripes {
3826 put_u32(
3827 &mut out,
3828 u32::try_from(stripe.parts.len()).map_err(|_| invalid("too many parts in a stripe"))?,
3829 );
3830 for &rows in &stripe.parts {
3831 put_u32(&mut out, rows);
3832 }
3833 put_u64(&mut out, stripe.index.offset);
3834 put_u32(&mut out, stripe.index.length);
3835 for page in &stripe.pages {
3836 put_u64(&mut out, page.offset);
3837 put_u32(&mut out, page.length);
3838 }
3839 for ((field, dictionary), membership) in
3844 table.fields.iter().zip(&table.dictionaries).zip(&stripe.memberships)
3845 {
3846 if field.ty != LogicalType::Varchar || dictionary.is_none() {
3847 continue;
3848 }
3849 let page =
3850 membership.ok_or_else(|| invalid("string page has no code membership index"))?;
3851 put_u64(&mut out, page.offset);
3852 put_u32(&mut out, page.length);
3853 put_u64(&mut out, page.hash);
3854 }
3855 for sieve in &stripe.sieves {
3856 match sieve {
3857 None => out.push(0),
3858 Some(page) => {
3859 out.push(1);
3860 put_u64(&mut out, page.offset);
3861 put_u32(&mut out, page.length);
3862 put_u64(&mut out, page.hash);
3863 }
3864 }
3865 }
3866 for held in &stripe.part_ranges {
3867 match held {
3868 None => out.push(0),
3869 Some(page) => {
3870 out.push(1);
3871 put_u64(&mut out, page.offset);
3872 put_u32(&mut out, page.length);
3873 put_u64(&mut out, page.hash);
3874 }
3875 }
3876 }
3877 for range in stripe.zone.columns() {
3878 put_bound(&mut out, range.low.as_ref())?;
3879 put_bound(&mut out, range.high.as_ref())?;
3880 put_u32(
3881 &mut out,
3882 u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?,
3883 );
3884 out.push(u8::from(range.exact));
3885 match range.sum {
3886 None => out.push(0),
3887 Some(total) => {
3888 out.push(1);
3889 out.extend_from_slice(&total.to_le_bytes());
3890 }
3891 }
3892 }
3893 }
3894 out.extend_from_slice(FREQUENCIES);
3895 put_u16(
3896 &mut out,
3897 u16::try_from(table.frequencies.len())
3898 .map_err(|_| invalid("too many frequency columns"))?,
3899 );
3900 for summary in &table.frequencies {
3901 let Some(summary) = summary else {
3902 out.push(0);
3903 continue;
3904 };
3905 out.push(1);
3906 put_u64(&mut out, summary.omitted_max);
3907 put_u32(
3908 &mut out,
3909 u32::try_from(summary.entries.len())
3910 .map_err(|_| invalid("too many frequency entries"))?,
3911 );
3912 for entry in &summary.entries {
3913 match entry.value {
3914 FrequencyValue::Null => out.push(0),
3915 FrequencyValue::Integer(value) => {
3916 out.push(1);
3917 out.extend_from_slice(&value.to_le_bytes());
3918 }
3919 FrequencyValue::Code(value) => {
3920 out.push(2);
3921 put_u32(&mut out, value);
3922 }
3923 }
3924 put_u64(&mut out, entry.count);
3925 }
3926 put_u32(
3927 &mut out,
3928 u32::try_from(summary.ordinals.len())
3929 .map_err(|_| invalid("too many frequency ordinals"))?,
3930 );
3931 let mut previous = 0_u64;
3932 for (at, &ordinal) in summary.ordinals.iter().enumerate() {
3933 let delta = if at == 0 {
3934 ordinal
3935 } else {
3936 ordinal
3937 .checked_sub(previous)
3938 .ok_or_else(|| invalid("frequency ordinals are not ordered"))?
3939 };
3940 if at != 0 && delta == 0 {
3941 return Err(invalid("frequency ordinals are not unique"));
3942 }
3943 put_var_u64(&mut out, delta);
3944 previous = ordinal;
3945 }
3946 }
3947 if let Some(clustering) = &table.clustering {
3950 out.extend_from_slice(CLUSTERING);
3951 out.push(clustering.width().tag());
3952 put_u16(
3953 &mut out,
3954 u16::try_from(clustering.columns().len())
3955 .map_err(|_| invalid("too many clustering columns"))?,
3956 );
3957 for &column in clustering.columns() {
3958 put_u16(
3959 &mut out,
3960 u16::try_from(column).map_err(|_| invalid("clustering column index overflow"))?,
3961 );
3962 }
3963 }
3964 Ok(out)
3965}
3966
3967fn encode_catalog(entries: &[Entry]) -> Result<Vec<u8>> {
3973 let mut out = CATALOG.to_vec();
3974 put_u32(&mut out, u32::try_from(entries.len()).map_err(|_| invalid("too many tables"))?);
3975 for entry in entries {
3976 let name = entry.name.as_bytes();
3977 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
3978 out.extend_from_slice(name);
3979 put_u64(&mut out, u64::try_from(entry.rows).map_err(|_| invalid("row count overflow"))?);
3980 put_u16(
3981 &mut out,
3982 u16::try_from(entry.fields.len()).map_err(|_| invalid("too many columns"))?,
3983 );
3984 for field in &entry.fields {
3985 let name = field.name.as_bytes();
3986 put_u16(
3987 &mut out,
3988 u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?,
3989 );
3990 out.extend_from_slice(name);
3991 put_type(&mut out, &field.ty)?;
3992 out.push(u8::from(field.not_null));
3993 }
3994 put_u64(&mut out, entry.directory.offset);
3995 put_u32(&mut out, entry.directory.length);
3996 put_u64(&mut out, entry.directory.hash);
3997 }
3998 Ok(out)
3999}
4000
4001fn decode_catalog(bytes: &[u8], size: u64) -> Result<Vec<Entry>> {
4004 let mut cur = Cursor { bytes, at: 0 };
4005 if cur.take(8)? != CATALOG {
4006 return Err(invalid("catalog magic differs"));
4007 }
4008 let count = cur.u32()? as usize;
4009 let mut entries: Vec<Entry> = Vec::with_capacity(count.min(1024));
4010 for _ in 0..count {
4011 let name = cur.text()?;
4012 let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
4013 let width = cur.u16()? as usize;
4014 let mut fields = Vec::with_capacity(width);
4015 for _ in 0..width {
4016 let name = cur.text()?;
4017 let ty = read_type(&mut cur)?;
4018 let not_null = match cur.u8()? {
4019 0 => false,
4020 1 => true,
4021 _ => return Err(invalid("nullability flag differs")),
4022 };
4023 fields.push(Field { name, ty, not_null });
4024 }
4025 let directory = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
4026 let end = directory
4027 .offset
4028 .checked_add(u64::from(directory.length))
4029 .ok_or_else(|| invalid("table directory offset overflow"))?;
4030 if directory.offset < HEADER
4031 || end > size
4032 || directory.length as usize > MAX_DIRECTORY
4033 || directory.length == 0
4034 {
4035 return Err(invalid("table directory range is outside the file"));
4036 }
4037 if entries.iter().any(|held| held.name == name) {
4038 return Err(invalid("two tables in the catalog have the same name"));
4039 }
4040 entries.push(Entry { name, fields, rows, directory });
4041 }
4042 Ok(entries)
4043}
4044
4045struct Cursor<'a> {
4046 bytes: &'a [u8],
4047 at: usize,
4048}
4049impl<'a> Cursor<'a> {
4050 fn take(&mut self, len: usize) -> Result<&'a [u8]> {
4051 let end = self.at.checked_add(len).ok_or_else(|| invalid("directory offset overflow"))?;
4052 let bytes =
4053 self.bytes.get(self.at..end).ok_or_else(|| invalid("directory is truncated"))?;
4054 self.at = end;
4055 Ok(bytes)
4056 }
4057 fn u8(&mut self) -> Result<u8> {
4058 Ok(self.take(1)?[0])
4059 }
4060 fn u16(&mut self) -> Result<u16> {
4061 Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("two bytes")))
4062 }
4063 fn u32(&mut self) -> Result<u32> {
4064 Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("four bytes")))
4065 }
4066 fn u64(&mut self) -> Result<u64> {
4067 Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("eight bytes")))
4068 }
4069 fn var_u64(&mut self) -> Result<u64> {
4070 let mut value = 0_u64;
4071 for shift in (0..=63).step_by(7) {
4072 let byte = self.u8()?;
4073 let part = u64::from(byte & 0x7f);
4074 if shift == 63 && part > 1 {
4075 return Err(invalid("frequency ordinal varint overflows"));
4076 }
4077 value |= part << shift;
4078 if byte & 0x80 == 0 {
4079 return Ok(value);
4080 }
4081 }
4082 Err(invalid("frequency ordinal varint is too long"))
4083 }
4084 fn bound(&mut self) -> Result<Option<Bound>> {
4085 Ok(match self.u8()? {
4086 0 => None,
4087 1 => Some(Bound::Int(i128::from_le_bytes(
4088 self.take(16)?.try_into().expect("sixteen bytes"),
4089 ))),
4090 2 => Some(Bound::Real(f64::from_le_bytes(
4091 self.take(8)?.try_into().expect("eight bytes"),
4092 ))),
4093 3 => {
4094 let length = self.u32()? as usize;
4095 Some(Bound::Bytes(self.take(length)?.to_vec()))
4096 }
4097 4 => {
4098 let unscaled =
4099 i128::from_le_bytes(self.take(16)?.try_into().expect("sixteen bytes"));
4100 Some(Bound::Scaled { unscaled, scale: self.u8()? })
4101 }
4102 _ => return Err(invalid("bound tag differs")),
4103 })
4104 }
4105 fn text(&mut self) -> Result<String> {
4106 let len = self.u16()? as usize;
4107 String::from_utf8(self.take(len)?.to_vec()).map_err(|_| invalid("name is not UTF-8"))
4108 }
4109}
4110
4111fn decode_directory(bytes: &[u8], size: u64) -> Result<Table> {
4112 let mut cur = Cursor { bytes, at: 0 };
4113 if cur.take(8)? != DIRECTORY {
4114 return Err(invalid("directory magic differs"));
4115 }
4116 let name = cur.text()?;
4117 let width = cur.u16()? as usize;
4118 let mut fields = Vec::with_capacity(width);
4119 for _ in 0..width {
4120 let name = cur.text()?;
4121 let ty = read_type(&mut cur)?;
4122 let not_null = match cur.u8()? {
4123 0 => false,
4124 1 => true,
4125 _ => return Err(invalid("nullability flag differs")),
4126 };
4127 fields.push(Field { name, ty, not_null });
4128 }
4129 let mut dictionaries = Vec::with_capacity(width);
4130 for _ in 0..width {
4131 dictionaries.push(match cur.u8()? {
4132 0 => None,
4133 1 => {
4134 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
4135 let end = page
4136 .offset
4137 .checked_add(u64::from(page.length))
4138 .ok_or_else(|| invalid("dictionary page offset overflow"))?;
4139 if page.offset < HEADER || end > size {
4144 return Err(invalid("dictionary page range is outside the file"));
4145 }
4146 Some(page)
4147 }
4148 _ => return Err(invalid("dictionary page tag differs")),
4149 });
4150 }
4151 let mut distincts = Vec::with_capacity(width);
4152 for _ in 0..width {
4153 distincts.push(match cur.u8()? {
4154 0 => None,
4155 1 => Some(cur.u64()?),
4156 _ => return Err(invalid("distinct count tag differs")),
4157 });
4158 }
4159 let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
4160 let count = cur.u32()? as usize;
4161 let mut stripes = Vec::with_capacity(count);
4162 let mut total = 0_usize;
4163 for _ in 0..count {
4164 let count = cur.u32()? as usize;
4165 if count == 0 || count > STRIPE_PARTS {
4166 return Err(invalid("stripe part count is outside its bound"));
4167 }
4168 let mut parts = Vec::with_capacity(count);
4169 let mut stripe_rows = 0_usize;
4170 for _ in 0..count {
4171 let rows = cur.u32()?;
4172 if rows == 0 {
4173 return Err(invalid("empty part"));
4174 }
4175 parts.push(rows);
4176 stripe_rows = stripe_rows
4177 .checked_add(rows as usize)
4178 .ok_or_else(|| invalid("stripe row count overflow"))?;
4179 }
4180 total =
4181 total.checked_add(stripe_rows).ok_or_else(|| invalid("stripe row count overflow"))?;
4182 let index = Span { offset: cur.u64()?, length: cur.u32()? };
4183 let section = index_section(count)?;
4184 let wanted = section
4185 .checked_mul(width)
4186 .and_then(|bytes| u32::try_from(bytes).ok())
4187 .ok_or_else(|| invalid("index page length overflow"))?;
4188 let end = index
4189 .offset
4190 .checked_add(u64::from(index.length))
4191 .ok_or_else(|| invalid("index page offset overflow"))?;
4192 if index.offset < HEADER || end > size || index.length != wanted {
4193 return Err(invalid("index page range is outside the file"));
4194 }
4195 let mut pages = Vec::with_capacity(width);
4196 for _ in 0..width {
4197 let offset = cur.u64()?;
4198 let length = cur.u32()?;
4199 let end = offset
4200 .checked_add(u64::from(length))
4201 .ok_or_else(|| invalid("page offset overflow"))?;
4202 if offset < HEADER || end > size || length as usize > MAX_PAGE {
4203 return Err(invalid("page range is outside the file"));
4204 }
4205 pages.push(Span { offset, length });
4206 }
4207 let mut memberships = vec![None; width];
4208 for (column, field) in fields.iter().enumerate() {
4209 if field.ty != LogicalType::Varchar || dictionaries[column].is_none() {
4210 continue;
4211 }
4212 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
4213 let end = page
4214 .offset
4215 .checked_add(u64::from(page.length))
4216 .ok_or_else(|| invalid("membership page offset overflow"))?;
4217 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
4218 return Err(invalid("membership page range is outside the file"));
4219 }
4220 memberships[column] = Some(page);
4221 }
4222 let mut sieves = vec![None; width];
4223 for sieve in sieves.iter_mut().take(width) {
4224 match cur.u8()? {
4225 0 => continue,
4226 1 => {}
4227 _ => return Err(invalid("a sieve page has an unknown tag")),
4228 }
4229 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
4230 let end = page
4231 .offset
4232 .checked_add(u64::from(page.length))
4233 .ok_or_else(|| invalid("sieve page offset overflow"))?;
4234 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
4235 return Err(invalid("sieve page range is outside the file"));
4236 }
4237 *sieve = Some(page);
4238 }
4239 let mut part_ranges = vec![None; width];
4240 for held in part_ranges.iter_mut().take(width) {
4241 match cur.u8()? {
4242 0 => continue,
4243 1 => {}
4244 _ => return Err(invalid("a part range page has an unknown tag")),
4245 }
4246 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
4247 let end = page
4248 .offset
4249 .checked_add(u64::from(page.length))
4250 .ok_or_else(|| invalid("part range page offset overflow"))?;
4251 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
4252 return Err(invalid("part range page range is outside the file"));
4253 }
4254 *held = Some(page);
4255 }
4256 let mut ranges = Vec::with_capacity(width);
4257 for column in 0..width {
4258 let low = cur.bound()?;
4259 let high = cur.bound()?;
4260 let nulls = cur.u32()? as usize;
4261 if nulls > stripe_rows {
4262 return Err(invalid("null count exceeds stripe rows"));
4263 }
4264 let exact = cur.u8()? != 0;
4265 let sum = match cur.u8()? {
4266 0 => None,
4267 1 => Some(i128::from_le_bytes(
4268 cur.take(16)?.try_into().map_err(|_| invalid("a stripe sum is truncated"))?,
4269 )),
4270 _ => return Err(invalid("a stripe sum has an unknown tag")),
4271 };
4272 let ty = &fields.get(column).ok_or_else(|| invalid("a stripe range has no column"))?.ty;
4278 let low = low.map(|bound| scaled_as(bound, ty));
4279 let high = high.map(|bound| scaled_as(bound, ty));
4280 ranges.push(Range { low, high, nulls, exact, sum });
4281 }
4282 stripes.push(Stripe {
4283 rows: stripe_rows,
4284 parts,
4285 index,
4286 pages,
4287 memberships,
4288 sieves,
4289 part_ranges,
4290 zone: Zone::from_ranges(ranges),
4291 });
4292 }
4293 if total != rows {
4294 return Err(invalid("table row count differs from stripes"));
4295 }
4296 let frequencies = if cur.at == bytes.len() {
4297 vec![None; width]
4298 } else {
4299 if cur.take(8)? != FREQUENCIES {
4300 return Err(invalid("directory extension magic differs"));
4301 }
4302 if cur.u16()? as usize != width {
4303 return Err(invalid("frequency column count differs"));
4304 }
4305 let mut frequencies = Vec::with_capacity(width);
4306 for field in &fields {
4307 let summary = match cur.u8()? {
4308 0 => None,
4309 1 => {
4310 let omitted_max = cur.u64()?;
4311 let count = cur.u32()? as usize;
4312 if count > FREQUENCY_ENTRIES {
4313 return Err(invalid("frequency entry count exceeds its bound"));
4314 }
4315 let mut entries = Vec::with_capacity(count);
4316 for _ in 0..count {
4318 let value = match cur.u8()? {
4319 0 => FrequencyValue::Null,
4320 1 => FrequencyValue::Integer(i128::from_le_bytes(
4321 cur.take(16)?.try_into().expect("sixteen bytes"),
4322 )),
4323 2 => FrequencyValue::Code(cur.u32()?),
4324 _ => return Err(invalid("frequency value tag differs")),
4325 };
4326 let valid = matches!(
4327 (&field.ty, value),
4328 (_, FrequencyValue::Null)
4329 | (LogicalType::Varchar, FrequencyValue::Code(_))
4330 | (
4331 LogicalType::TinyInt
4332 | LogicalType::SmallInt
4333 | LogicalType::Integer
4334 | LogicalType::BigInt
4335 | LogicalType::UTinyInt
4336 | LogicalType::USmallInt
4337 | LogicalType::UInteger
4338 | LogicalType::UBigInt
4339 | LogicalType::Date
4340 | LogicalType::Timestamp,
4341 FrequencyValue::Integer(_),
4342 )
4343 );
4344 if !valid {
4345 return Err(invalid("frequency value does not match its column"));
4346 }
4347 let count = cur.u64()?;
4348 if count == 0 || count > rows as u64 {
4349 return Err(invalid("frequency count is outside the table"));
4350 }
4351 entries.push(FrequencyEntry { value, count });
4352 }
4353 if entries.windows(2).any(|pair| pair[0].count < pair[1].count) {
4354 return Err(invalid("frequency entries are not descending"));
4355 }
4356 let ordinals = {
4357 let ordinal_count = cur.u32()? as usize;
4358 if ordinal_count > FREQUENCY_ORDINALS || ordinal_count > rows {
4359 return Err(invalid("frequency ordinal count exceeds its bound"));
4360 }
4361 let mut ordinals = Vec::with_capacity(ordinal_count);
4362 let mut previous = 0_u64;
4363 for at in 0..ordinal_count {
4364 let delta = cur.var_u64()?;
4365 if at != 0 && delta == 0 {
4366 return Err(invalid("frequency ordinals are not increasing"));
4367 }
4368 let ordinal = if at == 0 {
4369 delta
4370 } else {
4371 previous
4372 .checked_add(delta)
4373 .ok_or_else(|| invalid("frequency ordinal overflows"))?
4374 };
4375 if ordinal >= rows as u64 {
4376 return Err(invalid("frequency ordinal is outside the table"));
4377 }
4378 ordinals.push(ordinal);
4379 previous = ordinal;
4380 }
4381 ordinals
4382 };
4383 Some(FrequencySummary { entries, omitted_max, ordinals })
4384 }
4385 _ => return Err(invalid("frequency summary tag differs")),
4386 };
4387 frequencies.push(summary);
4388 }
4389 frequencies
4390 };
4391 let clustering = if cur.at == bytes.len() {
4392 None
4393 } else {
4394 if cur.take(8)? != CLUSTERING {
4395 return Err(invalid("directory extension magic differs"));
4396 }
4397 let bucket =
4398 Width::from_tag(cur.u8()?).ok_or_else(|| invalid("clustering width tag differs"))?;
4399 let count = cur.u16()? as usize;
4400 let mut columns = Vec::with_capacity(count.min(width));
4401 for _ in 0..count {
4402 columns.push(u32::from(cur.u16()?));
4403 }
4404 Some(Clustering::new(columns, bucket, width).map_err(|_| {
4407 invalid("stored clustering declaration does not match the table it is on")
4408 })?)
4409 };
4410 if cur.at != bytes.len() {
4411 return Err(invalid("directory has trailing bytes"));
4412 }
4413 Ok(Table { name, fields, stripes, rows, dictionaries, distincts, frequencies, clustering })
4414}
4415
4416fn put_bound(out: &mut Vec<u8>, bound: Option<&Bound>) -> Result<()> {
4417 match bound {
4418 None => out.push(0),
4419 Some(Bound::Int(value)) => {
4420 out.push(1);
4421 out.extend_from_slice(&value.to_le_bytes());
4422 }
4423 Some(Bound::Real(value)) => {
4424 out.push(2);
4425 out.extend_from_slice(&value.to_le_bytes());
4426 }
4427 Some(Bound::Bytes(value)) => {
4428 out.push(3);
4429 put_u32(out, u32::try_from(value.len()).map_err(|_| invalid("bound length overflow"))?);
4430 out.extend_from_slice(value);
4431 }
4432 Some(Bound::Scaled { unscaled, scale }) => {
4433 out.push(4);
4434 out.extend_from_slice(&unscaled.to_le_bytes());
4435 out.push(*scale);
4436 }
4437 }
4438 Ok(())
4439}
4440
4441#[derive(Debug)]
4458struct Codes;
4459
4460impl chooser::Chooser for Codes {
4461 fn name(&self) -> &'static str {
4462 "codes"
4463 }
4464
4465 fn narrow_strings(
4466 &self,
4467 _values: &[&[u8]],
4468 offered: &[string::Kind],
4469 _depth: u8,
4470 ) -> Vec<string::Kind> {
4471 offered.to_vec()
4474 }
4475
4476 fn narrow_integers(
4477 &self,
4478 _values: &[i64],
4479 offered: &[integer::Kind],
4480 depth: u8,
4481 ) -> Vec<integer::Kind> {
4482 let keep: &[integer::Kind] = if depth == 0 {
4483 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Rle]
4484 } else {
4485 &[integer::Kind::Constant, integer::Kind::Packed]
4486 };
4487 let narrowed: Vec<integer::Kind> =
4488 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
4489 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
4492 }
4493}
4494
4495#[derive(Debug)]
4507struct Fixed;
4508
4509impl chooser::Chooser for Fixed {
4510 fn name(&self) -> &'static str {
4511 "fixed"
4512 }
4513
4514 fn narrow_strings(
4515 &self,
4516 _values: &[&[u8]],
4517 offered: &[string::Kind],
4518 _depth: u8,
4519 ) -> Vec<string::Kind> {
4520 offered.to_vec()
4521 }
4522
4523 fn narrow_integers(
4524 &self,
4525 _values: &[i64],
4526 offered: &[integer::Kind],
4527 depth: u8,
4528 ) -> Vec<integer::Kind> {
4529 let keep: &[integer::Kind] = if depth == 0 {
4530 &[
4531 integer::Kind::Constant,
4532 integer::Kind::Packed,
4533 integer::Kind::Delta,
4534 integer::Kind::Rle,
4535 integer::Kind::Sparse,
4536 integer::Kind::Strided,
4537 ]
4538 } else {
4539 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Delta]
4540 };
4541 let narrowed: Vec<integer::Kind> =
4542 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
4543 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
4544 }
4545}
4546
4547fn widened(data: &Data) -> Option<Vec<i64>> {
4554 match data {
4555 Data::Int8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
4556 Data::UInt8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
4557 Data::Int16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
4558 Data::UInt16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
4559 Data::Int32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
4560 Data::UInt32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
4561 Data::Int64(values) => Some(values.to_vec()),
4562 _ => None,
4563 }
4564}
4565
4566trait Narrow: Copy {
4573 const BIASED: (u32, u64);
4578
4579 fn narrow(value: i64) -> Self;
4581}
4582
4583#[allow(clippy::cast_sign_loss, reason = "a residue is a bit pattern and not a number")]
4600fn residue<T: Narrow>(value: i64) -> u64 {
4601 let (bits, bias) = T::BIASED;
4602 (value as u64).wrapping_add(bias) >> bits
4603}
4604
4605macro_rules! narrows {
4610 ($($ty:ty => $bias:expr),* $(,)?) => {$(
4611 impl Narrow for $ty {
4612 const BIASED: (u32, u64) = (<$ty>::BITS, $bias);
4613
4614 #[allow(
4615 clippy::cast_possible_truncation,
4616 clippy::cast_sign_loss,
4617 reason = "the caller has checked the bits this truncates away"
4618 )]
4619 fn narrow(value: i64) -> Self {
4620 value as Self
4621 }
4622 }
4623 )*};
4624}
4625
4626narrows! {
4627 i8 => 1 << 7,
4628 u8 => 0,
4629 i16 => 1 << 15,
4630 u16 => 0,
4631 i32 => 1 << 31,
4632 u32 => 0,
4633}
4634
4635fn fit<T: Narrow>(values: &[i64]) -> Result<Vec<T>> {
4648 let mut spilled = 0u64;
4649 for value in values {
4650 spilled |= residue::<T>(*value);
4651 }
4652 if spilled != 0 {
4653 return Err(invalid("page value is not of its type"));
4654 }
4655 Ok(values.iter().map(|value| T::narrow(*value)).collect())
4656}
4657
4658fn narrowed(ty: &LogicalType, values: Vec<i64>) -> Result<Data> {
4663 Ok(match ty {
4664 LogicalType::TinyInt => Data::Int8(fit::<i8>(&values)?.into()),
4665 LogicalType::UTinyInt => Data::UInt8(fit::<u8>(&values)?.into()),
4666 LogicalType::SmallInt => Data::Int16(fit::<i16>(&values)?.into()),
4667 LogicalType::USmallInt => Data::UInt16(fit::<u16>(&values)?.into()),
4668 LogicalType::Integer | LogicalType::Date => Data::Int32(fit::<i32>(&values)?.into()),
4669 LogicalType::UInteger => Data::UInt32(fit::<u32>(&values)?.into()),
4670 LogicalType::BigInt | LogicalType::Timestamp => Data::Int64(values.into()),
4671 LogicalType::Decimal { .. } => match ty.physical() {
4674 PhysicalType::Int16 => Data::Int16(fit::<i16>(&values)?.into()),
4675 PhysicalType::Int32 => Data::Int32(fit::<i32>(&values)?.into()),
4676 PhysicalType::Int64 => Data::Int64(values.into()),
4677 _ => return Err(invalid("cascade codec belongs to a decimal that is not an integer")),
4678 },
4679 _ => return Err(invalid("cascade codec belongs to a page that is not integers")),
4680 })
4681}
4682
4683fn plain_width(ty: &LogicalType) -> Option<usize> {
4686 Some(match ty {
4687 LogicalType::TinyInt | LogicalType::UTinyInt => 1,
4688 LogicalType::SmallInt | LogicalType::USmallInt => 2,
4689 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date => 4,
4690 LogicalType::BigInt | LogicalType::Timestamp => 8,
4691 LogicalType::Decimal { .. } => match ty.physical() {
4692 PhysicalType::Int16 => 2,
4693 PhysicalType::Int32 => 4,
4694 PhysicalType::Int64 => 8,
4695 _ => return None,
4698 },
4699 _ => return None,
4700 })
4701}
4702
4703fn cascaded(
4709 flat: &Vector,
4710 ty: &LogicalType,
4711 packed: Option<&Packed<'_>>,
4712) -> Result<Option<Vec<u8>>> {
4713 let (Some(width), Some(data)) = (plain_width(ty), flat.data()) else { return Ok(None) };
4714 let Some(values) = widened(data) else { return Ok(None) };
4715 let plain = values.len().saturating_mul(width);
4716 let best = match packed {
4717 Some(packed) => plain.min(21 + size_of_val(packed.words())),
4719 None => plain,
4720 };
4721 let out = integer::encode_with(&values, &Fixed)?;
4722 Ok((out.len() < best).then_some(out))
4723}
4724
4725fn text_compressed(flat: &Vector) -> Result<Option<Vec<u8>>> {
4763 let mut values: Vec<&[u8]> = Vec::with_capacity(flat.len());
4764 let mut payload = 0_usize;
4765 for row in 0..flat.len() {
4766 let text = flat.text_at(row).unwrap_or("").as_bytes();
4767 payload = payload.saturating_add(text.len());
4768 values.push(text);
4769 }
4770 let plain = (flat.len() + 1).saturating_mul(4).saturating_add(payload);
4772 let Some(out) = string::encode_only(string::Kind::Fsst, &values)? else {
4773 return Ok(None);
4774 };
4775 Ok((out.len() < plain).then_some(out))
4776}
4777
4778fn encoded_codes(codes: &[u32]) -> Result<Option<Vec<u8>>> {
4779 let wide: Vec<i64> = codes.iter().map(|code| i64::from(*code)).collect();
4780 let coded = integer::encode_with(&wide, &Codes)?;
4781 let plain = codes.len().saturating_mul(size_of::<u32>());
4782 Ok((coded.len() < plain).then_some(coded))
4783}
4784
4785fn encode(
4786 vector: &Vector,
4787 global: Option<&mut GlobalDictionary>,
4788) -> Result<(Vec<u8>, Option<Vec<u32>>)> {
4789 let ty = vector.logical_type();
4790 let flat = vector.flatten()?;
4792 let mut out = Vec::new();
4793 let mut global_codes = None;
4794 if let Some(global) = global {
4795 let mut codes = Vec::with_capacity(flat.len());
4796 for row in 0..flat.len() {
4797 let text = flat.text_at(row).unwrap_or("");
4798 let code = global.code(text)?;
4799 global.observe(code, flat.is_null_at(row))?;
4800 codes.push(code);
4801 }
4802 global_codes = Some(codes);
4803 }
4804 let membership = global_codes.as_deref().map(unique_codes);
4805 let dictionary = if global_codes.is_none() && ty == &LogicalType::Varchar {
4806 string_dictionary(&flat)?
4807 } else {
4808 None
4809 };
4810 let compressed_text =
4811 if global_codes.is_none() && dictionary.is_none() && ty == &LogicalType::Varchar {
4812 text_compressed(&flat)?
4813 } else {
4814 None
4815 };
4816 let packed_vector = if dictionary.is_none() && global_codes.is_none() {
4817 Some(flat.bit_packed()?)
4818 } else {
4819 None
4820 };
4821 let packed = packed_vector.as_ref().and_then(Vector::packed_parts);
4822 let coded = match global_codes.as_deref() {
4823 Some(codes) => encoded_codes(codes)?,
4824 None => None,
4825 };
4826 let cascade = if dictionary.is_none() && global_codes.is_none() {
4830 cascaded(&flat, ty, packed.as_ref())?
4831 } else {
4832 None
4833 };
4834 out.push(if coded.is_some() {
4835 4
4836 } else if cascade.is_some() {
4837 5
4838 } else if global_codes.is_some() {
4839 3
4840 } else if dictionary.is_some() {
4841 1
4842 } else if compressed_text.is_some() {
4843 6
4844 } else if packed.is_some() {
4845 2
4846 } else {
4847 0
4848 });
4849 let nulls = flat.validity();
4850 let flag = match nulls {
4851 Validity::AllValid => 0,
4852 Validity::AllInvalid => 1,
4853 Validity::Mask(_) => 2,
4854 };
4855 out.push(flag);
4856 if flag == 2 {
4857 for group in (0..vector.len()).step_by(8) {
4858 let mut bits = 0_u8;
4859 for bit in 0..8 {
4860 if group + bit < vector.len() && !flat.is_null_at(group + bit) {
4861 bits |= 1 << bit;
4862 }
4863 }
4864 out.push(bits);
4865 }
4866 }
4867 if let Some(coded) = coded {
4868 out.extend_from_slice(&coded);
4869 return Ok((out, membership));
4870 }
4871 if let Some(cascade) = cascade {
4872 out.extend_from_slice(&cascade);
4873 return Ok((out, membership));
4874 }
4875 if let Some(codes) = global_codes {
4876 for code in codes {
4877 put_u32(&mut out, code);
4878 }
4879 return Ok((out, membership));
4880 }
4881 if let Some(dictionary) = dictionary {
4882 out.extend_from_slice(&dictionary);
4883 return Ok((out, membership));
4884 }
4885 if let Some(compressed_text) = compressed_text {
4886 out.extend_from_slice(&compressed_text);
4887 return Ok((out, membership));
4888 }
4889 if let Some(packed) = packed {
4890 if packed.offset() != 0 {
4891 return Err(invalid("writer received a sliced packed vector"));
4892 }
4893 out.push(u8::try_from(packed.width()).map_err(|_| invalid("packed width overflow"))?);
4894 out.extend_from_slice(&packed.base().to_le_bytes());
4895 put_u32(
4896 &mut out,
4897 u32::try_from(packed.words().len()).map_err(|_| invalid("too many packed words"))?,
4898 );
4899 for word in packed.words() {
4900 put_u64(&mut out, *word);
4901 }
4902 return Ok((out, membership));
4903 }
4904 let data = flat.data().ok_or_else(|| invalid("scalar column did not flatten"))?;
4905 match (ty, data) {
4906 (LogicalType::TinyInt, Data::Int8(values)) => {
4907 for value in &**values {
4908 out.extend_from_slice(&value.to_le_bytes());
4909 }
4910 }
4911 (LogicalType::UTinyInt, Data::UInt8(values)) => {
4912 for value in &**values {
4913 out.extend_from_slice(&value.to_le_bytes());
4914 }
4915 }
4916 (LogicalType::SmallInt, Data::Int16(values)) => {
4917 for value in &**values {
4918 out.extend_from_slice(&value.to_le_bytes());
4919 }
4920 }
4921 (LogicalType::USmallInt, Data::UInt16(values)) => {
4922 for value in &**values {
4923 out.extend_from_slice(&value.to_le_bytes());
4924 }
4925 }
4926 (LogicalType::UInteger, Data::UInt32(values)) => {
4927 for value in &**values {
4928 out.extend_from_slice(&value.to_le_bytes());
4929 }
4930 }
4931 (LogicalType::UBigInt, Data::UInt64(values)) => {
4932 for value in &**values {
4933 out.extend_from_slice(&value.to_le_bytes());
4934 }
4935 }
4936 (LogicalType::Integer | LogicalType::Date, Data::Int32(values)) => {
4937 for value in &**values {
4938 out.extend_from_slice(&value.to_le_bytes());
4939 }
4940 }
4941 (LogicalType::BigInt | LogicalType::Timestamp, Data::Int64(values)) => {
4942 for value in &**values {
4943 out.extend_from_slice(&value.to_le_bytes());
4944 }
4945 }
4946 (LogicalType::Boolean, Data::Bool(values)) => {
4947 for value in &**values {
4948 out.push(u8::from(*value));
4949 }
4950 }
4951 (LogicalType::Decimal { .. }, Data::Int16(values)) => {
4954 for value in &**values {
4955 out.extend_from_slice(&value.to_le_bytes());
4956 }
4957 }
4958 (LogicalType::Decimal { .. }, Data::Int32(values)) => {
4959 for value in &**values {
4960 out.extend_from_slice(&value.to_le_bytes());
4961 }
4962 }
4963 (LogicalType::Decimal { .. }, Data::Int64(values)) => {
4964 for value in &**values {
4965 out.extend_from_slice(&value.to_le_bytes());
4966 }
4967 }
4968 (LogicalType::Decimal { .. }, Data::Int128(values)) => {
4969 for value in &**values {
4970 out.extend_from_slice(&value.to_le_bytes());
4971 }
4972 }
4973 (LogicalType::Varchar, Data::Varlen(values)) => {
4974 let mut bytes = Vec::new();
4975 put_u32(&mut out, 0);
4976 for row in 0..vector.len() {
4977 let value = values.bytes(row).ok_or_else(|| invalid("string view is invalid"))?;
4978 bytes.extend_from_slice(value);
4979 put_u32(
4980 &mut out,
4981 u32::try_from(bytes.len())
4982 .map_err(|_| invalid("string payload exceeds 4GiB"))?,
4983 );
4984 }
4985 out.extend_from_slice(&bytes);
4986 }
4987 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
4988 }
4989 Ok((out, membership))
4990}
4991
4992fn put_varint(out: &mut Vec<u8>, mut value: u32) {
4993 while value >= 0x80 {
4994 out.push((value as u8 & 0x7f) | 0x80);
4995 value >>= 7;
4996 }
4997 out.push(value as u8);
4998}
4999
5000fn unique_codes(codes: &[u32]) -> Vec<u32> {
5002 let mut unique = codes.to_vec();
5003 unique.sort_unstable();
5004 unique.dedup();
5005 unique
5006}
5007
5008fn merged_codes(lists: Vec<Vec<u32>>) -> Vec<u32> {
5014 let mut lists = lists;
5015 while lists.len() > 1 {
5016 let mut next = Vec::with_capacity(lists.len().div_ceil(2));
5017 for pair in lists.chunks(2) {
5018 match pair {
5019 [left, right] => next.push(merged_pair(left, right)),
5020 [only] => next.push(only.clone()),
5021 _ => {}
5022 }
5023 }
5024 lists = next;
5025 }
5026 lists.pop().unwrap_or_default()
5027}
5028
5029fn merged_pair(left: &[u32], right: &[u32]) -> Vec<u32> {
5030 let mut out = Vec::with_capacity(left.len().saturating_add(right.len()));
5031 let mut at = 0;
5032 let mut to = 0;
5033 while at < left.len() && to < right.len() {
5034 match left[at].cmp(&right[to]) {
5035 Ordering::Less => {
5036 out.push(left[at]);
5037 at += 1;
5038 }
5039 Ordering::Greater => {
5040 out.push(right[to]);
5041 to += 1;
5042 }
5043 Ordering::Equal => {
5044 out.push(left[at]);
5045 at += 1;
5046 to += 1;
5047 }
5048 }
5049 }
5050 out.extend_from_slice(&left[at..]);
5051 out.extend_from_slice(&right[to..]);
5052 out
5053}
5054
5055fn merged_range(ranges: impl Iterator<Item = Range>) -> Range {
5060 let mut merged = Range::default();
5061 let mut first = true;
5062 for range in ranges {
5063 merged.nulls = merged.nulls.saturating_add(range.nulls);
5064 merged.sum = match (merged.sum.take(), range.sum) {
5068 (Some(held), Some(next)) if !first => held.checked_add(next),
5069 (_, next) if first => next,
5070 _ => None,
5071 };
5072 merged.exact = if first { range.exact } else { merged.exact && range.exact };
5073 if first {
5074 merged.low = range.low;
5075 merged.high = range.high;
5076 first = false;
5077 continue;
5078 }
5079 merged.low = match (merged.low.take(), range.low) {
5080 (Some(held), Some(next)) => Some(held.smaller(next)),
5081 _ => None,
5082 };
5083 merged.high = match (merged.high.take(), range.high) {
5084 (Some(held), Some(next)) => Some(held.larger(next)),
5085 _ => None,
5086 };
5087 }
5088 merged
5089}
5090
5091fn shortened(bound: Option<Bound>, high: bool) -> Option<Bound> {
5104 match bound {
5105 Some(Bound::Bytes(mut value)) if value.len() > PART_BOUND_BYTES => {
5106 value.truncate(PART_BOUND_BYTES);
5107 if !high {
5108 return Some(Bound::Bytes(value));
5109 }
5110 while let Some(last) = value.pop() {
5111 if last < u8::MAX {
5112 value.push(last + 1);
5113 return Some(Bound::Bytes(value));
5114 }
5115 }
5116 None
5117 }
5118 other => other,
5119 }
5120}
5121
5122fn encode_part_ranges(ranges: &[Range]) -> Result<Vec<u8>> {
5130 let mut out = Vec::new();
5131 put_u32(
5132 &mut out,
5133 u32::try_from(ranges.len()).map_err(|_| invalid("too many parts in a stripe"))?,
5134 );
5135 for range in ranges {
5136 put_bound(&mut out, shortened(range.low.clone(), false).as_ref())?;
5137 put_bound(&mut out, shortened(range.high.clone(), true).as_ref())?;
5138 put_u32(&mut out, u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?);
5139 }
5140 Ok(out)
5141}
5142
5143fn decode_part_ranges(bytes: &[u8]) -> Result<Vec<Range>> {
5145 let mut cur = Cursor { bytes, at: 0 };
5146 let parts = cur.u32()? as usize;
5147 let mut out = Vec::new();
5148 for _ in 0..parts {
5149 let low = cur.bound()?;
5150 let high = cur.bound()?;
5151 let nulls = cur.u32()? as usize;
5152 out.push(Range { low, high, nulls, exact: false, sum: None });
5153 }
5154 Ok(out)
5155}
5156
5157fn encode_sieves<'a>(sieves: impl Iterator<Item = &'a Option<Sieve>>) -> Result<Vec<u8>> {
5158 let held: Vec<&Option<Sieve>> = sieves.collect();
5159 let mut out = Vec::new();
5160 put_u32(
5161 &mut out,
5162 u32::try_from(held.len()).map_err(|_| invalid("too many parts in a stripe"))?,
5163 );
5164 for sieve in &held {
5165 let length = sieve.as_ref().map_or(0, Sieve::len);
5166 put_u32(&mut out, u32::try_from(length).map_err(|_| invalid("sieve length overflow"))?);
5167 }
5168 for sieve in held.into_iter().flatten() {
5170 out.extend_from_slice(&sieve.to_bytes());
5171 }
5172 Ok(out)
5173}
5174
5175fn decode_sieves(bytes: &[u8]) -> Result<Vec<Option<Sieve>>> {
5181 let parts = u32::from_le_bytes(
5182 bytes
5183 .get(..4)
5184 .ok_or_else(|| invalid("sieve page is truncated"))?
5185 .try_into()
5186 .map_err(|_| invalid("sieve page is truncated"))?,
5187 ) as usize;
5188 let mut lengths = Vec::with_capacity(parts);
5189 for part in 0..parts {
5190 let at = 4 + part * 4;
5191 let field = bytes.get(at..at + 4).ok_or_else(|| invalid("sieve page is truncated"))?;
5192 lengths.push(u32::from_le_bytes(
5193 field.try_into().map_err(|_| invalid("sieve page is truncated"))?,
5194 ) as usize);
5195 }
5196 let mut at = 4 + parts * 4;
5197 let mut out = Vec::with_capacity(parts);
5198 for length in lengths {
5199 if length == 0 {
5200 out.push(None);
5201 continue;
5202 }
5203 let end = at.checked_add(length).ok_or_else(|| invalid("sieve page is truncated"))?;
5204 let field = bytes.get(at..end).ok_or_else(|| invalid("sieve page is truncated"))?;
5205 out.push(Sieve::from_bytes(field));
5206 at = end;
5207 }
5208 if at != bytes.len() {
5209 return Err(invalid("sieve page has trailing bytes"));
5210 }
5211 Ok(out)
5212}
5213
5214fn encode_membership(unique: &[u32]) -> Vec<u8> {
5220 let mut out = Vec::with_capacity(unique.len().saturating_mul(2).saturating_add(5));
5221 put_varint(&mut out, u32::try_from(unique.len()).unwrap_or(u32::MAX));
5222 let mut previous = 0;
5223 for (at, &code) in unique.iter().enumerate() {
5224 put_varint(&mut out, if at == 0 { code } else { code - previous });
5225 previous = code;
5226 }
5227 out
5228}
5229
5230fn take_varint(bytes: &[u8], at: &mut usize) -> Result<u32> {
5231 let mut value = 0_u32;
5232 for shift in (0..35).step_by(7) {
5233 let byte = *bytes.get(*at).ok_or_else(|| invalid("membership varint is truncated"))?;
5234 *at += 1;
5235 let part = u32::from(byte & 0x7f);
5236 if shift == 28 && part > 0x0f {
5237 return Err(invalid("membership varint overflow"));
5238 }
5239 value = value
5240 .checked_add(
5241 part.checked_shl(shift).ok_or_else(|| invalid("membership varint overflow"))?,
5242 )
5243 .ok_or_else(|| invalid("membership varint overflow"))?;
5244 if byte & 0x80 == 0 {
5245 return Ok(value);
5246 }
5247 }
5248 Err(invalid("membership varint is too long"))
5249}
5250
5251fn decode_membership(bytes: &[u8]) -> Result<Vec<u32>> {
5252 let mut at = 0;
5253 let count = take_varint(bytes, &mut at)? as usize;
5254 let mut codes = Vec::with_capacity(count);
5255 let mut previous = 0_u32;
5256 for index in 0..count {
5257 let delta = take_varint(bytes, &mut at)?;
5258 let code = if index == 0 {
5259 delta
5260 } else {
5261 previous.checked_add(delta).ok_or_else(|| invalid("membership code overflow"))?
5262 };
5263 if index > 0 && code <= previous {
5264 return Err(invalid("membership codes are not increasing"));
5265 }
5266 codes.push(code);
5267 previous = code;
5268 }
5269 if at != bytes.len() {
5270 return Err(invalid("membership page has trailing bytes"));
5271 }
5272 Ok(codes)
5273}
5274
5275fn string_dictionary(vector: &Vector) -> Result<Option<Vec<u8>>> {
5276 let mut by_text = HashMap::new();
5277 let mut values = Vec::new();
5278 let mut codes = Vec::with_capacity(vector.len());
5279 let mut plain_bytes = 0_usize;
5280 for row in 0..vector.len() {
5281 let text = vector.text_at(row).unwrap_or("");
5282 plain_bytes = plain_bytes.saturating_add(text.len());
5283 let code = match by_text.get(text) {
5284 Some(&code) => code,
5285 None => {
5286 let code = u32::try_from(values.len())
5287 .map_err(|_| invalid("too many dictionary values"))?;
5288 by_text.insert(text, code);
5289 values.push(text);
5290 code
5291 }
5292 };
5293 codes.push(code);
5294 }
5295 let dictionary_bytes = values.iter().map(|value| value.len()).sum::<usize>();
5296 let encoded = 8_usize
5297 .saturating_add((values.len() + 1).saturating_mul(4))
5298 .saturating_add(dictionary_bytes)
5299 .saturating_add(codes.len().saturating_mul(4));
5300 let plain = (vector.len() + 1).saturating_mul(4).saturating_add(plain_bytes);
5301 if encoded >= plain {
5302 return Ok(None);
5303 }
5304 let mut out = Vec::with_capacity(encoded);
5305 put_u32(
5306 &mut out,
5307 u32::try_from(values.len()).map_err(|_| invalid("too many dictionary values"))?,
5308 );
5309 put_u32(
5310 &mut out,
5311 u32::try_from(dictionary_bytes).map_err(|_| invalid("dictionary payload exceeds 4GiB"))?,
5312 );
5313 let mut offset = 0_u32;
5314 put_u32(&mut out, offset);
5315 for value in &values {
5316 offset = offset
5317 .checked_add(
5318 u32::try_from(value.len()).map_err(|_| invalid("dictionary value is too long"))?,
5319 )
5320 .ok_or_else(|| invalid("dictionary payload exceeds 4GiB"))?;
5321 put_u32(&mut out, offset);
5322 }
5323 for value in values {
5324 out.extend_from_slice(value.as_bytes());
5325 }
5326 for code in codes {
5327 put_u32(&mut out, code);
5328 }
5329 Ok(Some(out))
5330}
5331
5332struct EncodedDictionary {
5333 index: Vec<u8>,
5334 ranks: Vec<u8>,
5335 payload: Vec<Vec<u8>>,
5338}
5339
5340fn head(bytes: &[u8]) -> u64 {
5342 let mut word = [0; 8];
5343 let take = bytes.len().min(8);
5344 word[..take].copy_from_slice(&bytes[..take]);
5345 u64::from_be_bytes(word)
5346}
5347
5348fn rankings(dictionaries: &[Option<GlobalDictionary>]) -> Result<Vec<Vec<(u64, u32)>>> {
5356 let present =
5357 dictionaries.iter().enumerate().filter(|(_, held)| held.is_some()).map(|(at, _)| at);
5358 let present = present.collect::<Vec<_>>();
5359 let mut orders = vec![Vec::new(); dictionaries.len()];
5360 let workers = std::thread::available_parallelism()
5361 .map_or(1, usize::from)
5362 .min(MAX_FREQUENCY_WORKERS)
5363 .min(present.len());
5364 if workers <= 1 {
5365 for at in present {
5366 if let Some(dictionary) = &dictionaries[at] {
5367 orders[at] = dictionary.ranked();
5368 }
5369 }
5370 return Ok(orders);
5371 }
5372 let width = present.len().div_ceil(workers);
5373 let pieces = std::thread::scope(|scope| {
5374 present
5375 .chunks(width)
5376 .map(|columns| {
5377 scope.spawn(|| {
5378 columns
5379 .iter()
5380 .filter_map(|&at| dictionaries[at].as_ref().map(|held| (at, held.ranked())))
5381 .collect::<Vec<_>>()
5382 })
5383 })
5384 .collect::<Vec<_>>()
5385 .into_iter()
5386 .map(|handle| {
5387 handle.join().map_err(|_| Error::internal("a dictionary sort worker panicked"))
5388 })
5389 .collect::<Result<Vec<_>>>()
5390 })?;
5391 for piece in pieces {
5392 for (at, order) in piece {
5393 orders[at] = order;
5394 }
5395 }
5396 Ok(orders)
5397}
5398
5399fn encode_global_dictionary(
5400 dictionary: GlobalDictionary,
5401 order: &[(u64, u32)],
5402) -> Result<EncodedDictionary> {
5403 let values = dictionary.offsets.len() - 1;
5404 if order.len() != values {
5405 return Err(invalid("global dictionary order does not cover its values"));
5406 }
5407 let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
5408 let payload = encode_payload(&dictionary)?;
5409 if payload.len() != blocks {
5410 return Err(invalid("global dictionary payload is not the blocks it says it is"));
5411 }
5412 let (ranks, rank_ends) = encode_ranks(order, code_width(values))?;
5413 let rank_blocks = values.div_ceil(TEXT_RANK_BLOCK);
5414 let offset_bits = offset_width(&dictionary.offsets);
5415 let mut index = Vec::with_capacity(
5416 DICTIONARY_HEADER + offset_bytes(values, offset_bits) + (blocks + rank_blocks) * 16,
5417 );
5418 put_u32(
5419 &mut index,
5420 u32::try_from(values).map_err(|_| invalid("global dictionary has too many values"))?,
5421 );
5422 put_u32(&mut index, TEXT_PAYLOAD_VALUES as u32);
5423 put_u32(
5424 &mut index,
5425 u32::try_from(blocks).map_err(|_| invalid("global dictionary has too many blocks"))?,
5426 );
5427 put_u32(&mut index, offset_bits as u32);
5428 encode_offsets(&dictionary.offsets, offset_bits, &mut index)?;
5429 let mut at = 0_u64;
5433 for block in &payload {
5434 at = at
5435 .checked_add(block.len() as u64)
5436 .ok_or_else(|| invalid("global dictionary payload overflow"))?;
5437 put_u64(&mut index, at);
5438 }
5439 for block in &payload {
5440 put_u64(&mut index, checksum(block));
5441 }
5442 if rank_ends.len() != rank_blocks {
5445 return Err(invalid("global dictionary order is not the blocks it says it is"));
5446 }
5447 for end in &rank_ends {
5448 put_u64(&mut index, *end);
5449 }
5450 let mut at = 0_usize;
5451 for end in &rank_ends {
5452 let end = usize::try_from(*end).map_err(|_| invalid("global dictionary order overflow"))?;
5453 put_u64(&mut index, checksum(&ranks[at..end]));
5454 at = end;
5455 }
5456 Ok(EncodedDictionary { index, ranks, payload })
5457}
5458
5459const PAYLOAD_SAMPLE_BLOCKS: usize = 8;
5466
5467fn payload_shapes() -> Vec<chooser::Settled> {
5493 let integers = vec![integer::Kind::Packed];
5494 [
5495 vec![string::Kind::Front, string::Kind::Lz],
5496 vec![string::Kind::Lz, string::Kind::Fsst],
5497 vec![string::Kind::Lz, string::Kind::Plain],
5498 vec![string::Kind::Fsst],
5499 vec![string::Kind::Plain],
5500 ]
5501 .into_iter()
5502 .map(|strings| chooser::Settled::new(strings, integers.clone()))
5503 .collect()
5504}
5505
5506fn encode_payload(dictionary: &GlobalDictionary) -> Result<Vec<Vec<u8>>> {
5512 let values = dictionary.offsets.len() - 1;
5513 let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
5514 let run = |block: usize| {
5515 let first = block * TEXT_PAYLOAD_VALUES;
5516 let last = (first + TEXT_PAYLOAD_VALUES).min(values);
5517 (first..last)
5518 .map(|value| {
5519 let from = dictionary.offsets[value] as usize;
5520 let to = dictionary.offsets[value + 1] as usize;
5521 &dictionary.payload[from..to]
5522 })
5523 .collect::<Vec<_>>()
5524 };
5525 let shape = (blocks > PAYLOAD_SAMPLE_BLOCKS).then(|| settle_shape(&run, blocks)).transpose()?;
5528 let one = |block: usize| match &shape {
5529 Some(shape) => string::encode_with(&run(block), shape),
5530 None => string::encode(&run(block)),
5531 };
5532 let workers = std::thread::available_parallelism()
5533 .map_or(1, usize::from)
5534 .min(MAX_FREQUENCY_WORKERS)
5535 .min(blocks);
5536 if workers <= 1 {
5537 return (0..blocks).map(one).collect();
5538 }
5539 let next = AtomicUsize::new(0);
5540 let pieces = std::thread::scope(|scope| {
5541 (0..workers)
5542 .map(|_| {
5543 scope.spawn(|| {
5544 let mut mine = Vec::new();
5545 loop {
5546 let block = next.fetch_add(1, Atomic::Relaxed);
5547 if block >= blocks {
5548 break;
5549 }
5550 mine.push((block, one(block)?));
5551 }
5552 Ok(mine)
5553 })
5554 })
5555 .collect::<Vec<_>>()
5556 .into_iter()
5557 .map(|handle| {
5558 handle.join().map_err(|_| Error::internal("a dictionary encode worker panicked"))?
5559 })
5560 .collect::<Result<Vec<_>>>()
5561 })?;
5562 let mut payload = vec![Vec::new(); blocks];
5563 for piece in pieces {
5564 for (block, bytes) in piece {
5565 payload[block] = bytes;
5566 }
5567 }
5568 Ok(payload)
5569}
5570
5571fn settle_shape<'a>(
5579 run: &dyn Fn(usize) -> Vec<&'a [u8]>,
5580 blocks: usize,
5581) -> Result<chooser::Settled> {
5582 let last = blocks - 1;
5583 let sample = (0..PAYLOAD_SAMPLE_BLOCKS)
5584 .map(|region| run(region * last / (PAYLOAD_SAMPLE_BLOCKS - 1)))
5585 .collect::<Vec<_>>();
5586 let mut best: Option<(chooser::Settled, usize)> = None;
5587 for shape in payload_shapes() {
5588 let mut size = 0;
5589 for block in &sample {
5590 size += string::encode_with(block, &shape)?.len();
5591 }
5592 if best.as_ref().is_none_or(|(_, smallest)| size < *smallest) {
5593 best = Some((shape, size));
5594 }
5595 }
5596 best.map(|(shape, _)| shape)
5597 .ok_or_else(|| invalid("no shape applies to a global dictionary payload"))
5598}
5599
5600fn encode_ranks(order: &[(u64, u32)], code_bits: usize) -> Result<(Vec<u8>, Vec<u64>)> {
5607 let mut out = Vec::with_capacity(order.len() * 4);
5608 let mut ends = Vec::with_capacity(order.len().div_ceil(TEXT_RANK_BLOCK));
5609 let mut heads = Vec::with_capacity(TEXT_RANK_BLOCK);
5610 let mut codes = Vec::with_capacity(TEXT_RANK_BLOCK);
5611 for block in order.chunks(TEXT_RANK_BLOCK) {
5612 let base = block.first().map_or(0, |&(head, _)| head);
5615 let span = block.last().map_or(0, |&(head, _)| head.wrapping_sub(base));
5616 let width = (u64::BITS - span.leading_zeros()) as usize;
5617 heads.clear();
5618 codes.clear();
5619 for &(head, code) in block {
5620 heads.push(head.wrapping_sub(base));
5621 codes.push(u64::from(code));
5622 }
5623 put_u64(&mut out, base);
5624 out.push(width as u8);
5625 bitpack::pack_tail(&heads, width, &mut out)
5626 .map_err(|_| invalid("global dictionary heads do not pack"))?;
5627 bitpack::pack_tail(&codes, code_bits, &mut out)
5628 .map_err(|_| invalid("global dictionary codes do not pack"))?;
5629 ends.push(out.len() as u64);
5630 }
5631 Ok((out, ends))
5632}
5633
5634fn open_global_dictionary(
5641 file: Arc<File>,
5642 page: Page,
5643 ty: &LogicalType,
5644 keep_budget: usize,
5645) -> Result<Vector> {
5646 if ty != &LogicalType::Varchar {
5647 return Err(invalid("global dictionary belongs to a non-string column"));
5648 }
5649 let mut header = [0; DICTIONARY_HEADER];
5650 read_at(&file, page.offset, &mut header)?;
5651 let count = u32::from_le_bytes(header[0..4].try_into().expect("four bytes")) as usize;
5652 let per_block = u32::from_le_bytes(header[4..8].try_into().expect("four bytes")) as usize;
5653 let blocks = u32::from_le_bytes(header[8..12].try_into().expect("four bytes")) as usize;
5654 let offset_bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
5655 if per_block != TEXT_PAYLOAD_VALUES {
5656 return Err(invalid("global dictionary block width differs"));
5657 }
5658 if blocks != count.div_ceil(TEXT_PAYLOAD_VALUES) {
5659 return Err(invalid("global dictionary block count differs from its value count"));
5660 }
5661 if offset_bits > u32::BITS as usize {
5662 return Err(invalid("global dictionary packs offsets past a payload"));
5663 }
5664 let offset_len = offset_bytes(count, offset_bits);
5665 let ranks = count;
5670 let rank_blocks = ranks.div_ceil(TEXT_RANK_BLOCK);
5671 let hash_len = blocks
5674 .checked_add(rank_blocks)
5675 .and_then(|words| words.checked_mul(16))
5676 .ok_or_else(|| invalid("global dictionary block count overflow"))?;
5677 let index_len = DICTIONARY_HEADER
5678 .checked_add(offset_len)
5679 .and_then(|len| len.checked_add(hash_len))
5680 .ok_or_else(|| invalid("global dictionary header overflow"))?;
5681 if index_len > page.length as usize {
5682 return Err(invalid("global dictionary offset index exceeds its page"));
5683 }
5684 let mut index = vec![0; index_len];
5685 index[..DICTIONARY_HEADER].copy_from_slice(&header);
5686 read_at(&file, page.offset + DICTIONARY_HEADER as u64, &mut index[DICTIONARY_HEADER..])?;
5687 if checksum(&index) != page.hash {
5688 return Err(invalid("global dictionary index checksum differs"));
5689 }
5690 let offsets = index[DICTIONARY_HEADER..DICTIONARY_HEADER + offset_len].to_vec();
5691 let mut words = index[DICTIONARY_HEADER + offset_len..]
5692 .chunks_exact(8)
5693 .map(|part| u64::from_le_bytes(part.try_into().expect("eight bytes")))
5694 .collect::<Vec<_>>();
5695 let mut hashes = words.split_off(blocks);
5696 let mut rank_ends = hashes.split_off(blocks);
5697 let rank_hashes = rank_ends.split_off(rank_blocks);
5698 let ends = words;
5699 if rank_ends.windows(2).any(|pair| pair[0] >= pair[1]) {
5702 return Err(invalid("global dictionary order blocks do not rise"));
5703 }
5704 let rank_len = usize::try_from(rank_ends.last().copied().unwrap_or_default())
5705 .map_err(|_| invalid("global dictionary rank overflow"))?;
5706 let body_len = index_len
5707 .checked_add(rank_len)
5708 .ok_or_else(|| invalid("global dictionary header overflow"))?;
5709 if body_len > page.length as usize {
5710 return Err(invalid("global dictionary order exceeds its page"));
5711 }
5712 let stored_len = page.length as usize - body_len;
5715 if ends.last().copied().unwrap_or_default() as usize != stored_len
5716 || ends.windows(2).any(|pair| pair[0] > pair[1])
5717 {
5718 return Err(invalid("global dictionary blocks do not bound the payload"));
5719 }
5720 Vector::external_text(
5721 LogicalType::Varchar,
5722 Arc::new(NativeText {
5723 file,
5724 values: count,
5725 offsets,
5726 offset_bits,
5727 ranks,
5728 rank_at: page.offset + index_len as u64,
5729 rank_ends,
5730 rank_hashes,
5731 rank_blocks: (0..rank_blocks).map(|_| OnceLock::new()).collect(),
5732 code_bits: code_width(count),
5733 code_ranks: OnceLock::new(),
5734 payload: page.offset + body_len as u64,
5735 ends,
5736 hashes,
5737 blocks: (0..blocks).map(|_| OnceLock::new()).collect(),
5738 keep_budget,
5739 payload_kept: AtomicUsize::new(0),
5740 searched: Mutex::new(HashMap::new()),
5741 }),
5742 )
5743}
5744
5745fn decode(
5746 ty: &LogicalType,
5747 rows: usize,
5748 bytes: &[u8],
5749 global: Option<Arc<Vector>>,
5750) -> Result<Vector> {
5751 let mut cur = Cursor { bytes, at: 0 };
5752 let codec = cur.u8()?;
5753 let flag = cur.u8()?;
5754 let validity = match flag {
5755 0 => Validity::AllValid,
5756 1 => Validity::AllInvalid,
5757 2 => {
5758 let mask = cur.take(rows.div_ceil(8))?;
5759 Validity::from_iter(rows, |row| mask[row / 8] >> (row % 8) & 1 == 1)
5760 }
5761 _ => return Err(invalid("page validity tag differs")),
5762 };
5763 if codec == 1 {
5764 if ty != &LogicalType::Varchar {
5765 return Err(invalid("dictionary codec belongs to a non-string page"));
5766 }
5767 let count = cur.u32()? as usize;
5768 let payload_len = cur.u32()? as usize;
5769 let offset_bytes = cur.take(
5770 (count + 1)
5771 .checked_mul(4)
5772 .ok_or_else(|| invalid("dictionary offset count overflow"))?,
5773 )?;
5774 let offsets = offset_bytes
5775 .chunks_exact(4)
5776 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
5777 .collect::<Vec<_>>();
5778 let payload = cur.take(payload_len)?.to_vec();
5779 if offsets.first() != Some(&0)
5780 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
5781 || offsets.windows(2).any(|pair| pair[0] > pair[1])
5782 {
5783 return Err(invalid("dictionary offsets do not bound the payload"));
5784 }
5785 let mut strings = StringColumn::over(Buffer::from_vec(payload).into_page());
5788 for pair in offsets.windows(2) {
5789 strings.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
5790 }
5791 let mut codes = Vec::with_capacity(rows);
5792 for _ in 0..rows {
5793 codes.push(cur.u32()?);
5794 }
5795 if codes.iter().any(|code| *code as usize >= count) {
5796 return Err(invalid("dictionary code is out of range"));
5797 }
5798 if cur.at != bytes.len() {
5799 return Err(invalid("dictionary page has trailing bytes"));
5800 }
5801 let dictionary = Vector::flat(LogicalType::Varchar, Data::Varlen(strings))?;
5802 return Ok(Vector::dictionary(codes, dictionary)?.with_validity(validity));
5803 }
5804 if codec == 3 || codec == 4 {
5805 let dictionary = global.ok_or_else(|| invalid("global code page has no dictionary"))?;
5806 let codes = if codec == 4 {
5807 let wide = integer::decode(&bytes[cur.at..])?;
5810 if wide.len() != rows {
5811 return Err(invalid("encoded code page holds the wrong number of rows"));
5812 }
5813 let mut codes = Vec::with_capacity(wide.len());
5820 let mut seen = 0_i64;
5821 for &code in &wide {
5822 seen |= code;
5823 codes.push(code as u32);
5824 }
5825 if seen < 0 || seen > i64::from(u32::MAX) {
5826 return Err(invalid("code is not a code"));
5827 }
5828 codes
5829 } else {
5830 let mut codes = Vec::with_capacity(rows);
5831 for _ in 0..rows {
5832 codes.push(cur.u32()?);
5833 }
5834 if cur.at != bytes.len() {
5835 return Err(invalid("global code page has trailing bytes"));
5836 }
5837 codes
5838 };
5839 let highest = codes.iter().copied().max();
5840 return Ok(Vector::stable_dictionary_validated(codes, dictionary, highest)?
5841 .with_validity(validity));
5842 }
5843 if codec == 6 {
5844 if ty != &LogicalType::Varchar {
5845 return Err(invalid("compressed text codec belongs to a non-string page"));
5846 }
5847 let (payload, ends) = string::decode_flat(&bytes[cur.at..])?.into_parts();
5851 if ends.len() != rows {
5852 return Err(invalid("compressed text page holds the wrong number of rows"));
5853 }
5854 let mut values = StringColumn::over(Buffer::from_vec(payload).into_page());
5857 let mut start = 0;
5858 for end in ends {
5859 let len = end
5860 .checked_sub(start)
5861 .ok_or_else(|| invalid("compressed text value ends before it starts"))?;
5862 values.push_in_place(start, len)?;
5863 start = end;
5864 }
5865 return Ok(Vector::flat(ty.clone(), Data::Varlen(values))?.with_validity(validity));
5866 }
5867 if codec == 5 {
5868 let values = integer::decode(&bytes[cur.at..])?;
5870 if values.len() != rows {
5871 return Err(invalid("cascade page holds the wrong number of rows"));
5872 }
5873 let data = narrowed(ty, values)?;
5874 return Ok(Vector::flat(ty.clone(), data)?.with_validity(validity));
5875 }
5876 if codec == 2 {
5877 let width = u32::from(cur.u8()?);
5878 let base = i128::from_le_bytes(cur.take(16)?.try_into().expect("sixteen bytes"));
5879 let count = cur.u32()? as usize;
5880 let mut words = Vec::with_capacity(count);
5881 for _ in 0..count {
5882 words.push(cur.u64()?);
5883 }
5884 if cur.at != bytes.len() {
5885 return Err(invalid("packed page has trailing bytes"));
5886 }
5887 return Ok(Vector::packed(ty.clone(), words, width, base, rows)?.with_validity(validity));
5888 }
5889 if codec != 0 {
5890 return Err(invalid("page codec is unknown"));
5891 }
5892 let data = match ty {
5893 LogicalType::TinyInt => {
5894 let values = cur.take(rows)?;
5895 Data::Int8(values.iter().map(|item| *item as i8).collect::<Vec<_>>().into())
5896 }
5897 LogicalType::UTinyInt => Data::UInt8(cur.take(rows)?.to_vec().into()),
5898 LogicalType::SmallInt => {
5899 let values =
5900 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
5901 Data::Int16(
5902 values
5903 .chunks_exact(2)
5904 .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
5905 .collect::<Vec<_>>()
5906 .into(),
5907 )
5908 }
5909 LogicalType::USmallInt => {
5910 let values =
5911 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
5912 Data::UInt16(
5913 values
5914 .chunks_exact(2)
5915 .map(|item| u16::from_le_bytes(item.try_into().expect("two bytes")))
5916 .collect::<Vec<_>>()
5917 .into(),
5918 )
5919 }
5920 LogicalType::UInteger => {
5921 let values =
5922 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
5923 Data::UInt32(
5924 values
5925 .chunks_exact(4)
5926 .map(|item| u32::from_le_bytes(item.try_into().expect("four bytes")))
5927 .collect::<Vec<_>>()
5928 .into(),
5929 )
5930 }
5931 LogicalType::UBigInt => {
5932 let values =
5933 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
5934 Data::UInt64(
5935 values
5936 .chunks_exact(8)
5937 .map(|item| u64::from_le_bytes(item.try_into().expect("eight bytes")))
5938 .collect::<Vec<_>>()
5939 .into(),
5940 )
5941 }
5942 LogicalType::Integer | LogicalType::Date => {
5943 let values =
5944 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
5945 Data::Int32(
5946 values
5947 .chunks_exact(4)
5948 .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
5949 .collect::<Vec<_>>()
5950 .into(),
5951 )
5952 }
5953 LogicalType::BigInt | LogicalType::Timestamp => {
5954 let values =
5955 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
5956 Data::Int64(
5957 values
5958 .chunks_exact(8)
5959 .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
5960 .collect::<Vec<_>>()
5961 .into(),
5962 )
5963 }
5964 LogicalType::Boolean => {
5965 let values = cur.take(rows)?;
5966 if values.iter().any(|value| *value > 1) {
5967 return Err(invalid("boolean page has another value"));
5968 }
5969 Data::Bool(values.iter().map(|value| *value == 1).collect::<Vec<_>>().into())
5970 }
5971 LogicalType::Decimal { .. } => match ty.physical() {
5974 PhysicalType::Int16 => {
5975 let values =
5976 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
5977 Data::Int16(
5978 values
5979 .chunks_exact(2)
5980 .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
5981 .collect::<Vec<_>>()
5982 .into(),
5983 )
5984 }
5985 PhysicalType::Int32 => {
5986 let values =
5987 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
5988 Data::Int32(
5989 values
5990 .chunks_exact(4)
5991 .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
5992 .collect::<Vec<_>>()
5993 .into(),
5994 )
5995 }
5996 PhysicalType::Int64 => {
5997 let values =
5998 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
5999 Data::Int64(
6000 values
6001 .chunks_exact(8)
6002 .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
6003 .collect::<Vec<_>>()
6004 .into(),
6005 )
6006 }
6007 _ => {
6008 let values =
6009 cur.take(rows.checked_mul(16).ok_or_else(|| invalid("page size overflow"))?)?;
6010 Data::Int128(
6011 values
6012 .chunks_exact(16)
6013 .map(|item| i128::from_le_bytes(item.try_into().expect("sixteen bytes")))
6014 .collect::<Vec<_>>()
6015 .into(),
6016 )
6017 }
6018 },
6019 LogicalType::Varchar => {
6020 let offset_bytes = cur
6021 .take((rows + 1).checked_mul(4).ok_or_else(|| invalid("offset count overflow"))?)?;
6022 let offsets = offset_bytes
6023 .chunks_exact(4)
6024 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
6025 .collect::<Vec<_>>();
6026 let payload = cur.take(bytes.len() - cur.at)?.to_vec();
6027 if offsets.first() != Some(&0)
6028 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
6029 || offsets.windows(2).any(|pair| pair[0] > pair[1])
6030 {
6031 return Err(invalid("string offsets do not bound the payload"));
6032 }
6033 let mut values = StringColumn::over(Buffer::from_vec(payload).into_page());
6037 for pair in offsets.windows(2) {
6038 values.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
6039 }
6040 Data::Varlen(values)
6041 }
6042 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
6043 };
6044 if cur.at != bytes.len() {
6045 return Err(invalid("page has trailing bytes"));
6046 }
6047 Ok(Vector::flat(ty.clone(), data)?.with_validity(validity))
6048}
6049
6050#[cfg(test)]
6051mod tests {
6052 use std::fs;
6053 use std::io::{Seek, SeekFrom, Write};
6054 use std::path::PathBuf;
6055 use std::time::{SystemTime, UNIX_EPOCH};
6056
6057 use rudb_common::Stat;
6058 use rudb_common::Value;
6059 use rudb_common::bounds::{Frequencies, Op, Remainder, Zones};
6060 use rudb_common::stat::Provenance;
6061
6062 use super::*;
6063
6064 #[test]
6065 fn checksum_matches_fixed_vectors() {
6066 assert_eq!(checksum(b""), 0xef46_db37_51d8_e999);
6067 assert_eq!(checksum(b"a"), 0xd24e_c4f1_a98c_6e5b);
6068 assert_eq!(checksum(b"abc"), 0x44bc_2cf5_ad77_0999);
6069 }
6070
6071 fn path(label: &str) -> PathBuf {
6072 let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
6073 std::env::temp_dir().join(format!("rudb-native-{label}-{}-{stamp}.rdb", std::process::id()))
6074 }
6075
6076 #[test]
6078 fn a_read_at_an_offset_ignores_where_another_thread_left_the_cursor() {
6079 const SPANS: usize = 64;
6080 const SPAN: usize = 512;
6081 let path = path("positional");
6082 let content: Vec<u8> =
6083 (0..SPANS).flat_map(|span| std::iter::repeat_n(span as u8, SPAN)).collect();
6084 fs::write(&path, &content).expect("the file is written");
6085 let file = Arc::new(File::open(&path).expect("the file opens"));
6086 std::thread::scope(|scope| {
6087 for _ in 0..8 {
6088 let file = Arc::clone(&file);
6089 scope.spawn(move || {
6090 for _ in 0..64 {
6091 for span in 0..SPANS {
6092 let mut bytes = [0_u8; SPAN];
6093 read_at(&file, (span * SPAN) as u64, &mut bytes)
6094 .expect("the span reads");
6095 assert!(
6096 bytes.iter().all(|byte| *byte == span as u8),
6097 "span {span} came back as {}",
6098 bytes[0],
6099 );
6100 }
6101 }
6102 });
6103 }
6104 });
6105 let mut past = [0_u8; SPAN];
6106 let end = (SPANS * SPAN) as u64;
6107 let error = read_at(&file, end, &mut past).expect_err("a read past the end is refused");
6108 assert!(error.message().contains("ends before its declared length"), "{error}");
6109 drop(file);
6110 let _ = fs::remove_file(&path);
6111 }
6112
6113 #[test]
6119 fn a_writer_puts_a_page_where_it_said_it_did_wherever_the_cursor_has_got_to() {
6120 let path = path("cursor");
6121 let mut writer = Writer::create(
6122 &path,
6123 "items",
6124 vec![
6125 Field::required("id", LogicalType::Integer),
6126 Field::new("text", LogicalType::Varchar),
6127 ],
6128 )
6129 .expect("new file");
6130 writer.append(&sample()).expect("first part");
6131 writer.file.seek(SeekFrom::Start(0)).expect("the cursor goes back to the header");
6132 writer.append(&sample()).expect("second part");
6133 writer.file.seek(SeekFrom::Start(1)).expect("and somewhere useless again");
6134 writer.finish().expect("commit");
6135 let reader = Reader::open(&path).expect("reopen from disk");
6136 assert_eq!(reader.table().rows(), 6);
6137 let ids = reader.read(0, &[0]).expect("the integer page reads back");
6138 assert_eq!(ids.value_at(0, 0), Value::Integer(4));
6139 assert_eq!(ids.value_at(2, 0), Value::Integer(-2));
6140 let text = reader.read(1, &[1]).expect("the text page reads back");
6141 assert_eq!(text.value_at(1, 0), Value::Null);
6142 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
6143 let end = reader.table().stripes().iter().flat_map(|stripe| {
6146 stripe
6147 .pages
6148 .iter()
6149 .map(|page| page.offset + u64::from(page.length))
6150 .chain(std::iter::once(stripe.index.offset + u64::from(stripe.index.length)))
6151 });
6152 let last = end.fold(HEADER, u64::max);
6153 let directory = fs::metadata(&path).expect("the file is there").len();
6154 assert!(last <= directory, "a page runs to {last} in a file of {directory} bytes");
6155 fs::remove_file(path).expect("remove scratch file");
6156 }
6157
6158 fn dictionary_index_len(header: &[u8; DICTIONARY_HEADER]) -> u64 {
6164 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
6165 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
6166 let bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
6167 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
6168 DICTIONARY_HEADER as u64
6169 + offset_bytes(count as usize, bits) as u64
6170 + (blocks + rank_blocks) * 16
6171 }
6172
6173 fn last_rank_end(file: &File, offset: u64, header: &[u8; DICTIONARY_HEADER]) -> u64 {
6175 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
6176 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
6177 let bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
6178 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
6179 let at = offset
6180 + DICTIONARY_HEADER as u64
6181 + offset_bytes(count as usize, bits) as u64
6182 + blocks * 16
6183 + (rank_blocks - 1) * 8;
6184 let mut end = [0; 8];
6185 read_at(file, at, &mut end).expect("the last rank block end");
6186 u64::from_le_bytes(end)
6187 }
6188
6189 fn sample() -> Chunk {
6190 Chunk::new(vec![
6191 Vector::from_values(
6192 LogicalType::Integer,
6193 &[Value::Integer(4), Value::Integer(9), Value::Integer(-2)],
6194 )
6195 .expect("integers"),
6196 Vector::from_values(
6197 LogicalType::Varchar,
6198 &[
6199 Value::Varchar("alpha".into()),
6200 Value::Null,
6201 Value::Varchar("long text after a slash".into()),
6202 ],
6203 )
6204 .expect("strings"),
6205 ])
6206 .expect("matching rows")
6207 }
6208
6209 fn sample_ids() -> Chunk {
6210 Chunk::new(vec![
6211 Vector::flat(LogicalType::Integer, Data::Int32(vec![7, 8, 9].into()))
6212 .expect("integers"),
6213 ])
6214 .expect("one column")
6215 }
6216
6217 #[test]
6218 fn the_planner_gets_the_null_count_off_the_same_directory_the_bounds_are_in() {
6219 let path = path("nulls_for_the_planner");
6222 let mut writer =
6223 Writer::create(&path, "items", vec![Field::new("a", LogicalType::Integer)])
6224 .expect("new file");
6225 let rows = Chunk::new(vec![
6226 Vector::from_values(
6227 LogicalType::Integer,
6228 &[
6229 Value::Integer(4),
6230 Value::Null,
6231 Value::Integer(9),
6232 Value::Null,
6233 Value::Integer(1),
6234 Value::Integer(2),
6235 ],
6236 )
6237 .expect("integers"),
6238 ])
6239 .expect("one column");
6240 writer.append(&rows).expect("the only part");
6241 writer.finish().expect("commit");
6242 let reader = Reader::open(&path).expect("reopen from disk");
6243 let stripes = Stripes::new(reader);
6244 let column = stripes.column("a").expect("the file has that column");
6245 assert_eq!(stripes.nulls(column), Stat::exact(2, Provenance::NullCount));
6246 assert_eq!(stripes.nulls(column + 1), Stat::Unknown);
6249 fs::remove_file(&path).expect("clean up");
6250 }
6251
6252 #[test]
6253 fn the_planner_gets_a_row_count_per_value_off_a_complete_synopsis() {
6254 let path = path("frequencies_for_the_planner");
6259 let mut writer =
6260 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
6261 .expect("new file");
6262 let rows = Chunk::new(vec![
6263 Vector::from_values(
6264 LogicalType::Integer,
6265 &[
6266 Value::Integer(4),
6267 Value::Integer(4),
6268 Value::Integer(4),
6269 Value::Integer(9),
6270 Value::Integer(9),
6271 Value::Integer(1),
6272 ],
6273 )
6274 .expect("integers"),
6275 ])
6276 .expect("one column");
6277 writer.append(&rows).expect("the only part");
6278 writer.finish().expect("commit");
6279 let reader = Reader::open(&path).expect("reopen from disk");
6280 let common = Common::new(reader);
6281 assert_eq!(common.rows(), 6);
6282 let column = common.column("id").expect("the file has that column");
6283 assert_eq!(common.column("nothing"), None);
6284 assert_eq!(
6285 common.rows_with(column, &Bound::Int(4)),
6286 Stat::exact(3, Provenance::FrequencySynopsis)
6287 );
6288 assert_eq!(
6290 common.rows_with(column, &Bound::Int(7)),
6291 Stat::exact(0, Provenance::FrequencySynopsis)
6292 );
6293 assert_eq!(common.rows_with(column, &Bound::Bytes(b"four".to_vec())), Stat::Unknown);
6296 assert_eq!(common.remainder(column), None);
6299 fs::remove_file(&path).expect("clean up");
6300 }
6301
6302 #[test]
6303 fn the_planner_gets_an_exact_count_for_a_leading_value_of_an_incomplete_synopsis() {
6304 let path = path("frequency_prefix_for_the_planner");
6311 let mut writer =
6312 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
6313 .expect("new file");
6314 let mut values = vec![Value::Integer(1); 10_000];
6315 for _ in 0..10 {
6316 values.extend((0..600).map(|tail| Value::Integer(1_000 + tail)));
6317 }
6318 for part in values.chunks(8_000) {
6321 let rows = Chunk::new(vec![
6322 Vector::from_values(LogicalType::Integer, part).expect("integers"),
6323 ])
6324 .expect("one column");
6325 writer.append(&rows).expect("a part");
6326 }
6327 writer.finish().expect("commit");
6328 let reader = Reader::open(&path).expect("reopen from disk");
6329 let prefix =
6330 reader.frequency_prefix(0).expect("a readable synopsis").expect("the column has one");
6331 assert_eq!(prefix.entries.len(), 512);
6334 assert_eq!(prefix.omitted_max, 10);
6335 let common = Common::new(reader);
6336 assert_eq!(common.rows(), 16_000);
6337 let column = common.column("id").expect("the file has that column");
6338 assert_eq!(
6339 common.rows_with(column, &Bound::Int(1)),
6340 Stat::exact(10_000, Provenance::FrequencySynopsis)
6341 );
6342 assert_eq!(
6344 common.rows_with(column, &Bound::Int(1_100)),
6345 Stat::exact(10, Provenance::FrequencySynopsis)
6346 );
6347 assert_eq!(common.rows_with(column, &Bound::Int(1_550)), Stat::Unknown);
6350 assert_eq!(common.rows_with(column, &Bound::Int(9_999)), Stat::Unknown);
6353 let remainder = common.remainder(column).expect("the list is a prefix");
6357 assert_eq!(remainder, Remainder { rows: 890, listed: 512, most: 10 });
6358 assert_eq!(remainder.rows / (601 - remainder.listed), 10);
6359 fs::remove_file(&path).expect("clean up");
6360 }
6361
6362 #[test]
6363 fn committed_file_reopens_and_reads_only_requested_columns() {
6364 let path = path("reopen");
6365 let mut writer = Writer::create(
6366 &path,
6367 "items",
6368 vec![
6369 Field::required("id", LogicalType::Integer),
6370 Field::new("text", LogicalType::Varchar),
6371 ],
6372 )
6373 .expect("new file");
6374 writer.append(&sample()).expect("first part");
6375 writer.append(&sample()).expect("second part");
6376 writer.finish().expect("commit");
6377 let reader = Reader::open(&path).expect("reopen from disk");
6378 assert_eq!(reader.table().rows(), 6);
6379 assert_eq!(reader.table().stripes().len(), 1);
6382 assert_eq!(reader.parts(), 2);
6383 assert_eq!(reader.part_rows(0), 3);
6384 assert_eq!(reader.part_rows(1), 3);
6385 let text = reader.read(1, &[1]).expect("only text page");
6386 assert_eq!(text.width(), 1);
6387 assert_eq!(text.value_at(1, 0), Value::Null);
6388 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
6389 let sparse = reader.read_sparse(1, &[1]).expect("one part without its whole page");
6390 assert_eq!(sparse.width(), 1);
6391 assert_eq!(sparse.value_at(1, 0), Value::Null);
6392 assert_eq!(sparse.value_at(2, 0), Value::Varchar("long text after a slash".into()));
6393 assert!(!reader.skips_codes(0, 1, &[0]).expect("alpha is in the stripe"));
6394 assert!(!reader.skips_codes(0, 1, &[2]).expect("long text is in the stripe"));
6395 assert!(reader.skips_codes(0, 1, &[3]).expect("unknown code is absent"));
6396 let count = reader.read(0, &[]).expect("no page is needed for count");
6397 assert_eq!(count.len(), 3);
6398 assert!(reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }]));
6399 assert!(!reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(0) }]));
6400 let integers = reader.top_frequencies(0, 1).expect("valid integer synopsis").expect("kept");
6401 assert_eq!(
6402 integers,
6403 vec![(Value::Integer(-2), 2), (Value::Integer(4), 2), (Value::Integer(9), 2),]
6404 );
6405 let strings = reader.top_frequencies(1, 1).expect("valid string synopsis").expect("kept");
6406 assert_eq!(strings.len(), 3);
6407 assert!(strings.contains(&(Value::Null, 2)));
6408 assert!(strings.contains(&(Value::Varchar("alpha".into()), 2)));
6409 assert!(strings.contains(&(Value::Varchar("long text after a slash".into()), 2)));
6410 fs::remove_file(path).expect("remove scratch file");
6411 }
6412
6413 #[test]
6421 fn runs_handed_over_out_of_order_still_read_back_in_source_order() {
6422 let path = path("interleaved-runs");
6423 let mut writer =
6424 Writer::create(&path, "interleaved", vec![Field::new("v", LogicalType::BigInt)])
6425 .expect("new file");
6426 for morsel in [2_u64, 0, 3, 1] {
6427 let parts = (0..4_u64)
6428 .map(|chunk| {
6429 let first = i64::try_from(morsel * 32 + chunk * 8).expect("small");
6430 let values =
6431 (0..8_i64).map(|row| Value::BigInt(first + row)).collect::<Vec<_>>();
6432 let column =
6433 Vector::from_values(LogicalType::BigInt, &values).expect("a column");
6434 ((morsel, chunk), Chunk::new(vec![column]).expect("one column"))
6435 })
6436 .collect::<Vec<_>>();
6437 writer.append_stripe(parts).expect("a stripe");
6438 }
6439 writer.finish().expect("commit");
6440
6441 let reader = Reader::open(&path).expect("valid directory");
6442 assert_eq!(reader.table().stripes().len(), 4, "a run is a stripe of its own");
6443 assert_eq!(reader.table().rows(), 128);
6444 for part in 0..16_usize {
6445 let read = reader.read(part, &[0]).expect("a part back");
6446 for row in 0..8_usize {
6447 let want = i64::try_from(part * 8 + row).expect("small");
6448 assert_eq!(read.value_at(row, 0), Value::BigInt(want), "part {part} row {row}");
6449 }
6450 }
6451 fs::remove_file(path).expect("remove scratch file");
6452 }
6453
6454 #[test]
6457 fn runs_that_overlap_each_other_are_refused_at_commit() {
6458 let path = path("overlapping-runs");
6459 let mut writer =
6460 Writer::create(&path, "overlapping", vec![Field::new("v", LogicalType::BigInt)])
6461 .expect("new file");
6462 let one = |order: (u64, u64)| {
6463 let column =
6464 Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)]).expect("a column");
6465 (order, Chunk::new(vec![column]).expect("one column"))
6466 };
6467 writer.append_stripe(vec![one((0, 0)), one((0, 2))]).expect("a stripe");
6470 writer.append_stripe(vec![one((0, 1))]).expect("a stripe");
6471 let error = writer.finish().expect_err("the runs overlap");
6472 assert!(error.message().contains("source order"), "{error}");
6473 fs::remove_file(path).expect("remove scratch file");
6474 }
6475
6476 #[test]
6479 fn a_run_longer_than_a_stripe_is_refused() {
6480 let path = path("overlong-run");
6481 let mut writer =
6482 Writer::create(&path, "overlong", vec![Field::new("v", LogicalType::BigInt)])
6483 .expect("new file");
6484 let parts = (0..=STRIPE_PARTS)
6485 .map(|at| {
6486 let column = Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)])
6487 .expect("a column");
6488 let chunk = Chunk::new(vec![column]).expect("one column");
6489 ((0, u64::try_from(at).expect("small")), chunk)
6490 })
6491 .collect::<Vec<_>>();
6492 let error = writer.append_stripe(parts).expect_err("one part too many");
6493 assert!(error.message().contains("more parts than it holds"), "{error}");
6494 fs::remove_file(path).expect("remove scratch file");
6495 }
6496
6497 #[test]
6503 fn parts_past_the_stripe_bound_start_a_new_stripe() {
6504 let path = path("stripe-bound");
6505 let mut writer = Writer::create(
6506 &path,
6507 "items",
6508 vec![
6509 Field::required("id", LogicalType::Integer),
6510 Field::new("text", LogicalType::Varchar),
6511 ],
6512 )
6513 .expect("new file");
6514 let parts = STRIPE_PARTS * 2 + 3;
6515 for part in 0..parts {
6516 let id = part as i32;
6517 let chunk = Chunk::new(vec![
6518 Vector::from_values(
6519 LogicalType::Integer,
6520 &[Value::Integer(id), Value::Integer(-id)],
6521 )
6522 .expect("integers"),
6523 Vector::from_values(
6524 LogicalType::Varchar,
6525 &[Value::Varchar(format!("value {part}")), Value::Null],
6526 )
6527 .expect("strings"),
6528 ])
6529 .expect("matching rows");
6530 writer.append(&chunk).expect("one part");
6531 }
6532 writer.finish().expect("commit");
6533
6534 let reader = Reader::open(&path).expect("reopen from disk");
6535 assert_eq!(reader.parts(), parts);
6536 assert_eq!(reader.table().rows(), parts * 2);
6537 assert_eq!(reader.table().stripes().len(), parts.div_ceil(STRIPE_PARTS));
6538 assert_eq!(reader.table().stripes()[0].parts(), STRIPE_PARTS);
6539 assert_eq!(reader.table().stripes()[0].rows(), STRIPE_PARTS * 2);
6540 assert_eq!(reader.table().stripes()[2].parts(), 3);
6541 for part in (0..parts).rev() {
6544 let dense = reader.read(part, &[0, 1]).expect("a whole page read");
6545 let sparse = reader.read_sparse(part, &[0, 1]).expect("one part read");
6546 for chunk in [&dense, &sparse] {
6547 assert_eq!(chunk.len(), 2, "part {part} has its own row count");
6548 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
6549 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
6550 assert_eq!(chunk.value_at(0, 1), Value::Varchar(format!("value {part}")));
6551 assert_eq!(chunk.value_at(1, 1), Value::Null);
6552 }
6553 }
6554 let above = [Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }];
6557 assert!(reader.skips(0, &above), "the first stripe stops at 63");
6558 assert!(!reader.skips(STRIPE_PARTS * 2, &above), "the third stripe reaches 130");
6559 fs::remove_file(path).expect("remove scratch file");
6560 }
6561
6562 fn scattered(n: i64) -> i64 {
6564 n.wrapping_mul(-7_046_029_254_386_353_131)
6565 }
6566
6567 #[test]
6573 fn a_part_is_skipped_when_its_sieve_does_not_hold_the_constant() {
6574 let path = path("sieve-skip");
6575 let mut writer =
6576 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
6577 .expect("new file");
6578 let parts = STRIPE_PARTS + 3;
6579 let per_part = 128;
6583 for part in 0..parts {
6584 let held: Vec<Value> = (0..per_part)
6585 .map(|row| Value::BigInt(scattered((part * per_part + row) as i64)))
6586 .collect();
6587 let chunk =
6588 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
6589 .expect("one column");
6590 writer.append(&chunk).expect("one part");
6591 }
6592 writer.finish().expect("commit");
6593
6594 let reader = Reader::open(&path).expect("reopen from disk");
6595 let probe = |value: i64| Probe {
6596 column: 0,
6597 op: Op::Equal,
6598 value: Bound::Int(i128::from(scattered(value))),
6599 };
6600 for wanted in [0_i64, (per_part + 1) as i64, (parts * per_part - 1) as i64] {
6601 let tests = [probe(wanted)];
6602 let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &tests)).collect();
6603 let home = wanted as usize / per_part;
6604 assert!(kept.contains(&home), "the part holding {wanted} is read");
6605 assert!(kept.len() <= 2, "{wanted} keeps {kept:?}, which is more than one stray part");
6609 }
6610 let absent = [probe((parts * per_part) as i64 + 1)];
6611 let kept = (0..parts).filter(|&part| !reader.skips(part, &absent)).count();
6612 assert!(kept <= 1, "{kept} parts of {parts} kept a value no part holds");
6613 let tests = [probe(0)];
6616 assert!(
6617 reader.table().stripes().iter().all(|stripe| !stripe.zone.skips(&tests)),
6618 "the bounds rule out no stripe at all"
6619 );
6620 fs::remove_file(path).expect("remove scratch file");
6621 }
6622
6623 #[test]
6629 fn a_part_is_skipped_when_its_own_bounds_rule_out_a_comparison_the_stripe_keeps() {
6630 let path = path("part-range-skip");
6631 let mut writer =
6632 Writer::create(&path, "hits", vec![Field::required("at", LogicalType::BigInt)])
6633 .expect("new file");
6634 let parts = STRIPE_PARTS + 3;
6635 let per_part = 128;
6636 for part in 0..parts {
6637 let held: Vec<Value> = (0..per_part)
6641 .map(|row| {
6642 Value::BigInt((part * 1_000) as i64 + (scattered(row as i64).rem_euclid(900)))
6643 })
6644 .collect();
6645 let chunk =
6646 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
6647 .expect("one column");
6648 writer.append(&chunk).expect("one part");
6649 }
6650 writer.finish().expect("commit");
6651
6652 let reader = Reader::open(&path).expect("reopen from disk");
6653 let under = [Probe { column: 0, op: Op::Less, value: Bound::Int(3_000) }];
6654 let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &under)).collect();
6655 assert_eq!(kept, vec![0, 1, 2], "only the three parts that start under three thousand");
6656 assert!(!reader.stripe_skips(0, &under), "the stripe reaches from zero and keeps itself");
6658 fs::remove_file(path).expect("remove scratch file");
6659 }
6660
6661 #[test]
6665 fn a_part_is_waved_through_when_its_own_bounds_pass_a_comparison_the_stripe_cannot() {
6666 let path = path("part-range-certain");
6667 let mut writer =
6668 Writer::create(&path, "hits", vec![Field::required("at", LogicalType::BigInt)])
6669 .expect("new file");
6670 let parts = STRIPE_PARTS + 3;
6671 let per_part = 128;
6672 for part in 0..parts {
6673 let held: Vec<Value> = (0..per_part)
6674 .map(|row| {
6675 Value::BigInt((part * 1_000) as i64 + (scattered(row as i64).rem_euclid(900)))
6676 })
6677 .collect();
6678 let chunk =
6679 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
6680 .expect("one column");
6681 writer.append(&chunk).expect("one part");
6682 }
6683 writer.finish().expect("commit");
6684
6685 let reader = Reader::open(&path).expect("reopen from disk");
6686 let under = [Probe { column: 0, op: Op::Less, value: Bound::Int(3_000) }];
6687 let waved: Vec<usize> = (0..parts).filter(|&part| reader.certain(part, &under)).collect();
6688 assert_eq!(waved, vec![0, 1, 2], "the three parts that end under three thousand");
6689 assert!(!reader.stripe_skips(0, &under), "the stripe straddles the comparison");
6692 fs::remove_file(path).expect("remove scratch file");
6693 }
6694
6695 #[test]
6698 fn a_stripe_of_one_part_writes_no_range_page_and_a_stripe_of_many_does() {
6699 for (parts, wanted) in [(1_usize, false), (STRIPE_PARTS, true)] {
6700 let path = path("part-range-page");
6701 let mut writer =
6702 Writer::create(&path, "hits", vec![Field::required("at", LogicalType::BigInt)])
6703 .expect("new file");
6704 for part in 0..parts {
6705 let held: Vec<Value> = (0..128)
6706 .map(|row| {
6707 Value::BigInt((part * 1_000) as i64 + scattered(row as i64).rem_euclid(900))
6708 })
6709 .collect();
6710 let chunk = Chunk::new(vec![
6711 Vector::from_values(LogicalType::BigInt, &held).expect("numbers"),
6712 ])
6713 .expect("one column");
6714 writer.append(&chunk).expect("one part");
6715 }
6716 writer.finish().expect("commit");
6717 let reader = Reader::open(&path).expect("reopen from disk");
6718 let bytes = reader.layout().columns[0].part_ranges;
6719 assert_eq!(bytes > 0, wanted, "{parts} parts wrote {bytes} bytes of ranges");
6720 fs::remove_file(path).expect("remove scratch file");
6721 }
6722 }
6723
6724 #[test]
6727 fn a_string_end_that_is_cut_down_still_covers_the_value_it_came_from() {
6728 let long = vec![b'a'; PART_BOUND_BYTES * 2];
6729 let low = shortened(Some(Bound::Bytes(long.clone())), false).expect("a low end");
6730 let high = shortened(Some(Bound::Bytes(long.clone())), true).expect("a high end");
6731 let Bound::Bytes(low) = low else { panic!("a string stays a string") };
6732 let Bound::Bytes(high) = high else { panic!("a string stays a string") };
6733 assert!(low.len() <= PART_BOUND_BYTES && high.len() <= PART_BOUND_BYTES);
6734 assert!(low.as_slice() <= long.as_slice(), "the low end is at or under the value");
6735 assert!(high.as_slice() >= long.as_slice(), "the high end is at or over the value");
6736 }
6737
6738 #[test]
6741 fn a_string_end_with_no_room_to_step_up_gives_up_the_bound() {
6742 let long = vec![u8::MAX; PART_BOUND_BYTES * 2];
6743 assert_eq!(shortened(Some(Bound::Bytes(long.clone())), true), None);
6744 let low = shortened(Some(Bound::Bytes(long)), false).expect("a low end is still a prefix");
6745 assert_eq!(low, Bound::Bytes(vec![u8::MAX; PART_BOUND_BYTES]));
6746 }
6747
6748 #[test]
6758 fn a_sieve_larger_than_the_part_it_indexes_is_not_written() {
6759 let path = path("sieve-pays");
6760 let fields = vec![
6761 Field::required("spread", LogicalType::BigInt),
6762 Field::required("repeated", LogicalType::BigInt),
6763 ];
6764 let mut writer = Writer::create(&path, "hits", fields).expect("new file");
6765 let parts = 3;
6766 let per_part = 1024;
6767 for part in 0..parts {
6768 let base = (part * per_part) as i64;
6769 let spread: Vec<Value> =
6770 (0..per_part).map(|row| Value::BigInt(scattered(base + row as i64))).collect();
6771 let repeated: Vec<Value> =
6772 (0..per_part).map(|row| Value::BigInt(scattered((row / 256) as i64))).collect();
6773 let chunk = Chunk::new(vec![
6774 Vector::from_values(LogicalType::BigInt, &spread).expect("numbers"),
6775 Vector::from_values(LogicalType::BigInt, &repeated).expect("numbers"),
6776 ])
6777 .expect("two columns");
6778 writer.append(&chunk).expect("one part");
6779 }
6780 writer.finish().expect("commit");
6781
6782 let reader = Reader::open(&path).expect("reopen from disk");
6783 let layout = reader.layout();
6784 let spread = &layout.columns[0];
6785 let repeated = &layout.columns[1];
6786 assert!(spread.sieves > 0, "a column whose parts are worth a filter keeps one");
6787 assert_eq!(
6788 repeated.sieves, 0,
6789 "a column whose filter costs more than its parts keeps none"
6790 );
6791 for column in &layout.columns {
6794 assert!(
6795 column.sieves < column.pages,
6796 "{} spends {} on sieves over {} of data",
6797 column.name,
6798 column.sieves,
6799 column.pages
6800 );
6801 }
6802 let absent = [Probe {
6804 column: 0,
6805 op: Op::Equal,
6806 value: Bound::Int(i128::from(scattered((parts * per_part) as i64 + 1))),
6807 }];
6808 assert!((0..parts).all(|part| reader.skips(part, &absent)), "no part holds it");
6809 fs::remove_file(path).expect("remove scratch file");
6810 }
6811
6812 #[test]
6818 fn a_damaged_sieve_page_is_read_through_rather_than_refused() {
6819 let path = path("sieve-damaged");
6820 let mut writer =
6821 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
6822 .expect("new file");
6823 let rows = 128;
6824 let held: Vec<Value> = (0..rows).map(|row| Value::BigInt(scattered(row))).collect();
6825 let chunk =
6826 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
6827 .expect("one column");
6828 writer.append(&chunk).expect("one part");
6829 writer.finish().expect("commit");
6830
6831 let page =
6832 Reader::open(&path).expect("reopen").table.stripes[0].sieves[0].expect("a sieve page");
6833 let mut file = OpenOptions::new().write(true).open(&path).expect("open the sieve page");
6834 file.seek(SeekFrom::Start(page.offset + u64::from(page.length) - 1)).expect("seek");
6835 file.write_all(&[0xff]).expect("damage one byte");
6836 drop(file);
6837
6838 let reader = Reader::open(&path).expect("reopen the damaged file");
6839 let absent =
6840 [Probe { column: 0, op: Op::Equal, value: Bound::Int(i128::from(scattered(99))) }];
6841 assert!(!reader.skips(0, &absent), "a sieve that cannot be read skips nothing");
6842 assert_eq!(
6843 reader.read(0, &[0]).expect("the rows are untouched").len(),
6844 usize::try_from(rows).expect("a small count")
6845 );
6846 fs::remove_file(path).expect("remove scratch file");
6847 }
6848
6849 #[test]
6860 fn workers_that_want_the_same_stripe_read_it_once() {
6861 let path = path("single-flight");
6862 let mut writer =
6863 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
6864 .expect("new file");
6865 for part in 0..STRIPE_PARTS {
6866 let id = part as i32;
6867 let chunk = Chunk::new(vec![
6868 Vector::from_values(
6869 LogicalType::Integer,
6870 &[Value::Integer(id), Value::Integer(-id)],
6871 )
6872 .expect("integers"),
6873 ])
6874 .expect("matching rows");
6875 writer.append(&chunk).expect("one part");
6876 }
6877 writer.finish().expect("commit");
6878
6879 let reader = Reader::open(&path).expect("reopen from disk");
6880 assert_eq!(reader.table().stripes().len(), 1, "one stripe is the point of the test");
6881 let barrier = std::sync::Barrier::new(8);
6882 std::thread::scope(|scope| {
6883 for worker in 0..8 {
6884 let reader = &reader;
6885 let barrier = &barrier;
6886 scope.spawn(move || {
6887 barrier.wait();
6888 for part in (worker..STRIPE_PARTS).step_by(8) {
6889 let chunk = reader.read(part, &[0]).expect("a whole page read");
6890 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
6891 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
6892 }
6893 });
6894 }
6895 });
6896 assert_eq!(reader.pages.load(Atomic::Relaxed), 1, "one stripe, one page read, whoever won");
6897 fs::remove_file(path).expect("remove scratch file");
6898 }
6899
6900 #[test]
6913 fn opening_costs_the_same_over_a_thousand_times_the_rows() {
6914 let opened = |label: &str, rows_per_part: i32| {
6915 let path = path(label);
6916 let mut writer =
6917 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
6918 .expect("new file");
6919 for part in 0..STRIPE_PARTS * 3 {
6920 let values = (0..rows_per_part)
6924 .map(|row| {
6925 Value::Integer((part as i32 * rows_per_part + row).wrapping_mul(2_654_435))
6926 })
6927 .collect::<Vec<_>>();
6928 let chunk = Chunk::new(vec![
6929 Vector::from_values(LogicalType::Integer, &values).expect("integers"),
6930 ])
6931 .expect("matching rows");
6932 writer.append(&chunk).expect("one part");
6933 }
6934 writer.finish().expect("commit");
6935 let reader = Reader::open(&path).expect("reopen from disk");
6936 let size = fs::metadata(&path).expect("the file is there").len();
6937 let out = (reader.reads(), reader.table().stripes().len(), size);
6938 fs::remove_file(path).expect("remove scratch file");
6939 out
6940 };
6941
6942 let (thin, thin_stripes, thin_size) = opened("open-thin", 1);
6943 let (fat, fat_stripes, fat_size) = opened("open-fat", 1000);
6944 assert_eq!(
6945 thin_stripes, fat_stripes,
6946 "the same stripe count is what makes this a fair ask"
6947 );
6948 assert!(
6949 fat_size > thin_size * 50,
6950 "the fat file has to actually be larger, and it is {fat_size} against {thin_size}"
6951 );
6952
6953 assert_eq!(thin.opening.reads, fat.opening.reads, "the same reads either way");
6954 assert_eq!(thin.pages, 0, "opening read a page");
6955 assert_eq!(fat.pages, 0, "opening read a page");
6956 assert_eq!(thin.indexes, 0, "opening read an index");
6957 assert_eq!(fat.indexes, 0, "opening read an index");
6958 assert!(
6961 fat.opening.bytes < thin.opening.bytes * 2,
6962 "opening the thin file read {} bytes and the fat one read {}",
6963 thin.opening.bytes,
6964 fat.opening.bytes
6965 );
6966 }
6967
6968 #[test]
6976 fn two_opens_of_one_file_cost_the_same_and_the_second_is_not_cheaper() {
6977 let path = path("open-twice");
6978 let mut writer =
6979 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
6980 .expect("new file");
6981 for part in 0..STRIPE_PARTS * 3 {
6982 let chunk = Chunk::new(vec![
6983 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
6984 .expect("integers"),
6985 ])
6986 .expect("matching rows");
6987 writer.append(&chunk).expect("one part");
6988 }
6989 writer.finish().expect("commit");
6990
6991 let first = Reader::open(&path).expect("open");
6992 for part in 0..first.parts() {
6995 first.read(part, &[0]).expect("a part");
6996 }
6997 assert!(first.reads().pages > 0, "the scan has to have read something");
6998 let second = Reader::open(&path).expect("open again");
6999
7000 assert_eq!(first.reads().opening, second.reads().opening);
7001 assert_eq!(
7002 second.reads().pages,
7003 0,
7004 "the second open read a page off the back of the first"
7005 );
7006 assert_eq!(second.reads().indexes, 0, "the second open read an index it inherited");
7007 fs::remove_file(path).expect("remove scratch file");
7008 }
7009
7010 #[test]
7018 fn an_index_is_read_once_per_stripe_however_often_the_page_is_evicted() {
7019 let path = path("index-cache");
7020 let mut writer =
7021 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
7022 .expect("new file");
7023 let parts = STRIPE_PARTS * (CACHED_STRIPES_PER_COLUMN + 2);
7024 for part in 0..parts {
7025 let id = part as i32;
7026 let chunk = Chunk::new(vec![
7027 Vector::from_values(LogicalType::Integer, &[Value::Integer(id)]).expect("integers"),
7028 ])
7029 .expect("matching rows");
7030 writer.append(&chunk).expect("one part");
7031 }
7032 writer.finish().expect("commit");
7033
7034 let reader = Reader::open(&path).expect("reopen from disk");
7035 let stripes = reader.table().stripes().len();
7036 assert!(stripes > CACHED_STRIPES_PER_COLUMN, "the page cache has to be too small for this");
7037 for _ in 0..2 {
7039 for part in 0..parts {
7040 let chunk = reader.read(part, &[0]).expect("a part");
7041 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
7042 }
7043 }
7044 assert_eq!(reader.indexes.load(Atomic::Relaxed), stripes, "one index read per stripe");
7045 assert!(
7046 reader.pages.load(Atomic::Relaxed) > stripes,
7047 "the pages are the ones that get read again, which is what makes the index count mean \
7048 something"
7049 );
7050 fs::remove_file(path).expect("remove scratch file");
7051 }
7052
7053 #[test]
7062 fn a_worker_per_stripe_reads_its_page_once_when_the_cache_was_told_to_expect_it() {
7063 let workers = CACHED_STRIPES_PER_COLUMN + 4;
7064 let path = path("stripe-per-worker");
7065 let mut writer =
7066 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
7067 .expect("new file");
7068 for part in 0..STRIPE_PARTS * workers {
7069 let chunk = Chunk::new(vec![
7070 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
7071 .expect("integers"),
7072 ])
7073 .expect("matching rows");
7074 writer.append(&chunk).expect("one part");
7075 }
7076 writer.finish().expect("commit");
7077
7078 let read = |told: bool| {
7079 let reader = Reader::open(&path).expect("reopen from disk");
7080 assert_eq!(reader.table().stripes().len(), workers, "a stripe per worker");
7081 if told {
7082 reader.keep_stripes(workers);
7083 }
7084 let barrier = std::sync::Barrier::new(workers);
7085 std::thread::scope(|scope| {
7086 for (worker, run) in reader.stripe_parts().into_iter().enumerate() {
7087 let reader = &reader;
7088 let barrier = &barrier;
7089 scope.spawn(move || {
7090 for part in run {
7091 barrier.wait();
7092 let chunk = reader.read(part, &[0]).expect("a part of my own stripe");
7093 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
7094 }
7095 assert!(worker < workers);
7096 });
7097 }
7098 });
7099 reader.pages.load(Atomic::Relaxed)
7100 };
7101
7102 assert_eq!(read(true), workers, "one page read per stripe and no more");
7103 assert!(read(false) > workers, "a cache that small is read again on every part");
7104 fs::remove_file(path).expect("remove scratch file");
7105 }
7106
7107 #[test]
7112 fn a_damaged_index_page_is_an_error() {
7113 let path = path("damaged-index");
7114 let mut writer =
7115 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
7116 .expect("new file");
7117 writer.append(&sample_ids()).expect("first part");
7118 writer.append(&sample_ids()).expect("second part");
7119 writer.finish().expect("commit");
7120
7121 let reader = Reader::open(&path).expect("valid directory");
7122 let index = reader.table.stripes[0].index;
7123 let mut byte = [0; 1];
7124 read_at(&reader.file, index.offset, &mut byte).expect("the first part length");
7125 let mut file = OpenOptions::new().write(true).open(&path).expect("open index page");
7126 file.seek(SeekFrom::Start(index.offset)).expect("index start");
7127 file.write_all(&[!byte[0]]).expect("damage the first part length");
7128 let error = reader.read(1, &[0]).expect_err("a damaged index must not be used");
7129 assert!(error.message().contains("index page section checksum differs"), "{error}");
7130 fs::remove_file(path).expect("remove scratch file");
7131 }
7132
7133 #[test]
7140 fn every_integer_width_round_trips_through_a_page() {
7141 let path = path("integer-widths");
7142 let columns = [
7143 (LogicalType::TinyInt, vec![Value::TinyInt(i8::MIN), Value::TinyInt(i8::MAX)]),
7144 (LogicalType::UTinyInt, vec![Value::UTinyInt(0), Value::UTinyInt(u8::MAX)]),
7145 (LogicalType::SmallInt, vec![Value::SmallInt(i16::MIN), Value::SmallInt(i16::MAX)]),
7146 (LogicalType::USmallInt, vec![Value::USmallInt(0), Value::USmallInt(u16::MAX)]),
7147 (LogicalType::Integer, vec![Value::Integer(i32::MIN), Value::Integer(i32::MAX)]),
7148 (LogicalType::UInteger, vec![Value::UInteger(0), Value::UInteger(u32::MAX)]),
7149 (LogicalType::BigInt, vec![Value::BigInt(i64::MIN), Value::BigInt(i64::MAX)]),
7150 (LogicalType::UBigInt, vec![Value::UBigInt(0), Value::UBigInt(u64::MAX)]),
7151 ];
7152 let fields = columns
7153 .iter()
7154 .enumerate()
7155 .map(|(at, (ty, _))| Field::required(format!("c{at}"), ty.clone()))
7156 .collect::<Vec<_>>();
7157 let vectors = columns
7158 .iter()
7159 .map(|(ty, values)| Vector::from_values(ty.clone(), values).expect("a vector"))
7160 .collect::<Vec<_>>();
7161 let mut writer = Writer::create(&path, "widths", fields).expect("new file");
7162 writer.append(&Chunk::new(vectors).expect("matching rows")).expect("one stripe");
7163 writer.finish().expect("commit");
7164
7165 let reader = Reader::open(&path).expect("reopen from disk");
7166 let wanted = (0..columns.len()).collect::<Vec<_>>();
7167 let read = reader.read(0, &wanted).expect("every column");
7168 assert_eq!(read.len(), 2);
7169 for (at, (ty, values)) in columns.iter().enumerate() {
7171 assert_eq!(read.value_at(0, at), values[0], "the low end of {ty}");
7172 assert_eq!(read.value_at(1, at), values[1], "the high end of {ty}");
7173 }
7174 fs::remove_file(path).expect("remove scratch file");
7175 }
7176
7177 #[test]
7178 fn numeric_frequency_candidates_keep_bounded_row_ordinals() {
7179 let path = path("frequency-ordinals");
7180 let mut writer =
7181 Writer::create(&path, "items", vec![Field::required("id", LogicalType::BigInt)])
7182 .expect("new file");
7183 let mut values = Vec::new();
7184 for leader in 0..10_i64 {
7185 values.extend(std::iter::repeat_n(leader, 100));
7186 }
7187 values.extend(1_000_i64..41_000);
7188 for part in values.chunks(1_024) {
7189 let vector = Vector::flat(LogicalType::BigInt, Data::Int64(part.to_vec().into()))
7190 .expect("big integers");
7191 writer.append(&Chunk::new(vec![vector]).expect("one column")).expect("one stripe");
7192 }
7193 writer.finish().expect("commit");
7194
7195 let reader = Reader::open(&path).expect("reopen from disk");
7196 let occurrences =
7197 reader.frequency_occurrences(0).expect("valid metadata").expect("bounded ordinals");
7198 assert!(occurrences.omitted_max < 100);
7199 assert!(occurrences.ordinals.len() <= FREQUENCY_ORDINALS);
7200 assert!(occurrences.ordinals.windows(2).all(|pair| pair[0] < pair[1]));
7201 assert_eq!(&occurrences.ordinals[..1_000], &(0_u64..1_000).collect::<Vec<_>>());
7202 fs::remove_file(path).expect("remove scratch file");
7203 }
7204
7205 #[test]
7211 fn a_file_from_another_format_says_which_format_it_is() {
7212 let older = path("older-format");
7213 let mut writer =
7214 Writer::create(&older, "items", vec![Field::new("id", LogicalType::Integer)])
7215 .expect("new file");
7216 let chunk = Chunk::new(vec![
7217 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
7218 .expect("integers"),
7219 ])
7220 .expect("chunk");
7221 writer.append(&chunk).expect("page written");
7222 writer.finish().expect("commit");
7223
7224 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
7225 file.seek(SeekFrom::Start(8)).expect("the version follows the magic");
7226 file.write_all(&(FORMAT - 1).to_le_bytes()).expect("write an older version");
7227 drop(file);
7228 let complaint = Reader::open(&older).expect_err("an older format is refused").to_string();
7229 assert!(complaint.contains(&format!("format {}", FORMAT - 1)), "{complaint}");
7230 assert!(complaint.contains(&format!("format {FORMAT}")), "{complaint}");
7231
7232 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
7233 file.seek(SeekFrom::Start(0)).expect("the magic is first");
7234 file.write_all(b"NOTRUDB!").expect("write another engine's magic");
7235 drop(file);
7236 let complaint = Reader::open(&older).expect_err("a foreign file is refused").to_string();
7237 assert!(complaint.contains("magic"), "{complaint}");
7238 assert!(!complaint.contains("format"), "a version has nothing to do with it: {complaint}");
7239 fs::remove_file(older).expect("remove scratch file");
7240 }
7241
7242 #[test]
7243 fn an_unfinished_or_damaged_file_does_not_answer_with_partial_rows() {
7244 let unfinished = path("unfinished");
7245 let mut writer =
7246 Writer::create(&unfinished, "items", vec![Field::new("id", LogicalType::Integer)])
7247 .expect("new file");
7248 let chunk = Chunk::new(vec![
7249 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
7250 .expect("integers"),
7251 ])
7252 .expect("chunk");
7253 writer.append(&chunk).expect("page written");
7254 drop(writer);
7255 assert!(Reader::open(&unfinished).is_err(), "no directory was committed");
7256 fs::remove_file(unfinished).expect("remove scratch file");
7257
7258 let damaged = path("damaged");
7259 let mut writer =
7260 Writer::create(&damaged, "items", vec![Field::new("id", LogicalType::Integer)])
7261 .expect("new file");
7262 writer.append(&chunk).expect("page written");
7263 writer.finish().expect("commit");
7264 let reader = Reader::open(&damaged).expect("valid directory");
7265 let mut file =
7266 OpenOptions::new().write(true).open(&damaged).expect("open for a damaged page");
7267 file.seek(SeekFrom::Start(HEADER + 1)).expect("inside first page");
7268 file.write_all(&[255]).expect("damage one byte");
7269 assert!(reader.read(0, &[0]).is_err(), "page checksum rejects corruption");
7270 fs::remove_file(damaged).expect("remove scratch file");
7271 }
7272
7273 #[test]
7274 fn damaged_lazy_dictionary_payload_is_an_error() {
7275 let path = path("damaged-dictionary");
7276 let mut writer = Writer::create(
7277 &path,
7278 "items",
7279 vec![
7280 Field::required("id", LogicalType::Integer),
7281 Field::new("text", LogicalType::Varchar),
7282 ],
7283 )
7284 .expect("new file");
7285 writer.append(&sample()).expect("stripe written");
7286 writer.finish().expect("commit");
7287
7288 let reader = Reader::open(&path).expect("valid directory");
7289 let dictionary = reader.table.dictionaries[1].expect("string dictionary page");
7290 let mut header = [0; DICTIONARY_HEADER];
7293 read_at(&reader.file, dictionary.offset, &mut header).expect("dictionary header");
7294 let index_len = dictionary_index_len(&header);
7295 let rank_len = last_rank_end(&reader.file, dictionary.offset, &header);
7296 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
7297 file.seek(SeekFrom::Start(dictionary.offset + index_len + rank_len))
7298 .expect("inside dictionary payload");
7299 file.write_all(&[255]).expect("damage dictionary payload");
7300
7301 let chunk = reader.read(0, &[1]).expect("code page and dictionary index remain valid");
7302 let error =
7303 chunk.validate_external().expect_err("payload corruption must reach the caller");
7304 assert!(error.message().contains("payload checksum differs"), "{error}");
7305 fs::remove_file(path).expect("remove scratch file");
7306 }
7307
7308 #[test]
7318 fn a_column_of_all_different_values_is_written_without_a_dictionary() {
7319 let path = path("dictionary-decide");
7320 let rows = 20_000;
7321 let unique =
7323 |row: usize| format!("{row:09} a value that appears exactly once in the table");
7324 let repeated = |row: usize| unique(row / 40);
7326 let mut writer = Writer::create(
7327 &path,
7328 "items",
7329 vec![
7330 Field::required("unique", LogicalType::Varchar),
7331 Field::required("repeated", LogicalType::Varchar),
7332 ],
7333 )
7334 .expect("new file");
7335 for part in (0..rows).step_by(1_000) {
7336 let span = part..(part + 1_000).min(rows);
7337 let left = span.clone().map(|row| Value::Varchar(unique(row))).collect::<Vec<_>>();
7338 let right = span.map(|row| Value::Varchar(repeated(row))).collect::<Vec<_>>();
7339 writer
7340 .append(
7341 &Chunk::new(vec![
7342 Vector::from_values(LogicalType::Varchar, &left).expect("strings"),
7343 Vector::from_values(LogicalType::Varchar, &right).expect("strings"),
7344 ])
7345 .expect("two columns"),
7346 )
7347 .expect("a part");
7348 }
7349 writer.finish().expect("commit");
7350
7351 let reader = Reader::open(&path).expect("reopen from disk");
7352 assert!(
7353 reader.table.dictionaries[0].is_none(),
7354 "a column with no repeats has nothing to say twice"
7355 );
7356 assert!(
7357 reader.table.dictionaries[1].is_some(),
7358 "a column whose values come round again keeps its dictionary"
7359 );
7360 let mut first = 0;
7361 for part in 0..reader.parts() {
7362 let chunk = reader.read(part, &[0, 1]).expect("a part");
7363 for row in 0..chunk.len() {
7364 assert_eq!(chunk.value_at(row, 0), Value::Varchar(unique(first + row)));
7365 assert_eq!(chunk.value_at(row, 1), Value::Varchar(repeated(first + row)));
7366 }
7367 first += chunk.len();
7368 }
7369 assert_eq!(first, rows, "every row was read back");
7370 let raw = (0..rows).map(|row| unique(row).len()).sum::<usize>();
7371 let size = fs::metadata(&path).expect("the file is there").len() as usize;
7372 assert!(size < raw, "a column without a dictionary is still encoded: {size} against {raw}");
7373 fs::remove_file(path).expect("remove scratch file");
7374 }
7375
7376 #[test]
7389 fn a_dictionary_over_many_blocks_checks_every_block_of_it() {
7390 let path = path("dictionary-blocks");
7391 let value = |row: usize| {
7392 let row = row.saturating_sub(8_000);
7393 format!("{row:07} a value long enough to be worth a payload block")
7394 };
7395 let parts = 40;
7396 let per_part = 1000;
7397 let mut writer =
7398 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
7399 .expect("new file");
7400 for part in 0..parts {
7401 let values = (0..per_part)
7402 .map(|row| Value::Varchar(value(part * per_part + row)))
7403 .collect::<Vec<_>>();
7404 let chunk = Chunk::new(vec![
7405 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
7406 ])
7407 .expect("matching rows");
7408 writer.append(&chunk).expect("a part");
7409 }
7410 writer.finish().expect("commit");
7411
7412 let reader = Reader::open(&path).expect("reopen from disk");
7413 let dictionary = reader.table.dictionaries[0].expect("string dictionary page");
7414 assert!(
7415 parts * per_part > TEXT_PAYLOAD_VALUES * 4,
7416 "the dictionary has to be several blocks for this to be testing anything"
7417 );
7418 for part in [0, parts - 1] {
7419 let chunk = reader.read(part, &[0]).expect("a part");
7420 chunk.validate_external().expect("every payload block checks out");
7421 assert_eq!(chunk.value_at(0, 0), Value::Varchar(value(part * per_part)));
7422 }
7423
7424 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
7425 file.seek(SeekFrom::Start(dictionary.offset + u64::from(dictionary.length) - 4))
7426 .expect("the last bytes of the page are payload");
7427 file.write_all(&[255]).expect("damage the last payload block");
7428 let reader = Reader::open(&path).expect("the directory and the index are untouched");
7429 let chunk = reader.read(parts - 1, &[0]).expect("the code page remains valid");
7430 let error = chunk.validate_external().expect_err("the damage must reach the caller");
7431 assert!(error.message().contains("payload checksum differs"), "{error}");
7432 fs::remove_file(path).expect("remove scratch file");
7433 }
7434
7435 #[test]
7449 fn values_of_different_lengths_read_back_out_of_packed_offsets() {
7450 let path = path("dictionary-offsets");
7451 let value = |row: usize| {
7452 let row = row % 5_000;
7453 if row % 511 == 3 { String::new() } else { "x".repeat(row % 97) + &format!("{row:05}") }
7454 };
7455 let rows = 6_000;
7456 let mut writer =
7457 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
7458 .expect("new file");
7459 let values = (0..rows).map(|row| Value::Varchar(value(row))).collect::<Vec<_>>();
7460 for part in values.chunks(1_000) {
7461 let chunk =
7462 Chunk::new(vec![Vector::from_values(LogicalType::Varchar, part).expect("strings")])
7463 .expect("matching rows");
7464 writer.append(&chunk).expect("a part");
7465 }
7466 writer.finish().expect("commit");
7467
7468 let reader = Reader::open(&path).expect("reopen from disk");
7469 assert!(
7470 rows > TEXT_PAYLOAD_VALUES * 4,
7471 "the dictionary has to be several blocks for this to be testing anything"
7472 );
7473 for part in 0..rows / 1_000 {
7474 let chunk = reader.read(part, &[0]).expect("a part");
7475 for row in 0..1_000 {
7476 let row = part * 1_000 + row;
7477 assert_eq!(
7478 chunk.value_at(row % 1_000, 0),
7479 Value::Varchar(value(row)),
7480 "value {row}"
7481 );
7482 }
7483 }
7484 fs::remove_file(path).expect("remove scratch file");
7485 }
7486
7487 #[test]
7499 fn a_global_dictionary_is_opened_once_however_many_workers_ask_at_once() {
7500 let path = path("dictionary-once");
7501 let parts = 8;
7502 let per_part = 500;
7503 let value =
7504 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
7505 let mut writer =
7506 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
7507 .expect("new file");
7508 for part in 0..parts {
7509 let values = (0..per_part)
7510 .map(|row| Value::Varchar(value(part * per_part + row)))
7511 .collect::<Vec<_>>();
7512 let chunk = Chunk::new(vec![
7513 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
7514 ])
7515 .expect("matching rows");
7516 writer.append(&chunk).expect("a part");
7517 }
7518 writer.finish().expect("commit");
7519
7520 let reader = Reader::open(&path).expect("reopen from disk");
7521 assert!(reader.table.dictionaries[0].is_some(), "the column has to have one to share");
7522 assert_eq!(reader.reads().dictionaries, 0, "opening the file does not open a dictionary");
7523
7524 let workers = 16;
7525 let gate = std::sync::Barrier::new(workers);
7526 std::thread::scope(|scope| {
7527 for worker in 0..workers {
7528 let reader = reader.clone();
7529 let gate = &gate;
7530 scope.spawn(move || {
7531 gate.wait();
7532 let chunk = reader.read(worker % parts, &[0]).expect("a part");
7533 assert_eq!(
7534 chunk.value_at(0, 0),
7535 Value::Varchar(value((worker % parts) * per_part))
7536 );
7537 });
7538 }
7539 });
7540
7541 assert_eq!(reader.reads().dictionaries, 1, "sixteen workers, one dictionary, one open");
7542 fs::remove_file(path).expect("remove scratch file");
7543 }
7544
7545 #[test]
7550 fn a_damaged_sorted_order_is_an_error() {
7551 let path = path("damaged-order");
7552 let mut writer = Writer::create(
7553 &path,
7554 "items",
7555 vec![
7556 Field::required("id", LogicalType::Integer),
7557 Field::new("text", LogicalType::Varchar),
7558 ],
7559 )
7560 .expect("new file");
7561 writer.append(&sample()).expect("stripe written");
7562 writer.finish().expect("commit");
7563
7564 let reader = Reader::open(&path).expect("valid directory");
7565 let page = reader.table.dictionaries[1].expect("string dictionary page");
7566 let mut header = [0; DICTIONARY_HEADER];
7567 read_at(&reader.file, page.offset, &mut header).expect("dictionary header");
7568 let index_len = dictionary_index_len(&header);
7569 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
7570 file.seek(SeekFrom::Start(page.offset + index_len)).expect("the first head");
7571 file.write_all(&[255]).expect("damage the order");
7572
7573 let dictionary = reader.dictionary(1).expect("read").expect("a string column has one");
7574 let error = dictionary.compare_rank(0, b"anything").expect_err("a damaged order is caught");
7575 assert!(error.message().contains("rank checksum differs"), "{error}");
7576 fs::remove_file(path).expect("remove scratch file");
7577 }
7578
7579 #[test]
7583 fn a_global_dictionary_carries_the_sorted_order_of_its_values() {
7584 let spellings = ["overlong1z", "b", "", "overlong1a", "overlong", "ab", "a", "overlong1"];
7587 let path = path("dictionary-order");
7588 let mut writer =
7589 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
7590 .expect("new file");
7591 writer
7592 .append(
7593 &Chunk::new(vec![
7594 Vector::from_values(
7595 LogicalType::Varchar,
7596 &spellings.map(|text| Value::Varchar(text.into())),
7597 )
7598 .expect("strings"),
7599 ])
7600 .expect("one column"),
7601 )
7602 .expect("stripe written");
7603 writer.finish().expect("commit");
7604
7605 let reader = Reader::open(&path).expect("valid directory");
7606 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
7607 let count = dictionary.ranks().expect("a v10 file stores one");
7608 assert_eq!(count, spellings.len(), "every distinct value has a rank");
7609 let order = (0..count)
7610 .map(|rank| dictionary.code_at_rank(rank).expect("a code"))
7611 .collect::<Vec<_>>();
7612 let mut seen = order.clone();
7613 seen.sort_unstable();
7614 assert_eq!(seen, (0..spellings.len() as u32).collect::<Vec<_>>(), "a permutation of codes");
7615
7616 let ranked = order
7617 .iter()
7618 .map(|&code| {
7619 dictionary.try_bytes_at(code as usize).expect("read").expect("a value").to_vec()
7620 })
7621 .collect::<Vec<_>>();
7622 let mut expected = spellings.map(|text| text.as_bytes().to_vec()).to_vec();
7623 expected.sort();
7624 assert_eq!(ranked, expected, "rank order is value order");
7625
7626 for (rank, value) in expected.iter().enumerate() {
7629 assert_eq!(
7630 dictionary.compare_rank(rank, value).expect("compare"),
7631 Ordering::Equal,
7632 "rank {rank} is its own value"
7633 );
7634 if rank > 0 {
7635 assert_eq!(
7636 dictionary.compare_rank(rank - 1, value).expect("compare"),
7637 Ordering::Less,
7638 "rank {rank} follows the one before it"
7639 );
7640 }
7641 }
7642 fs::remove_file(path).expect("remove scratch file");
7643 }
7644
7645 #[test]
7653 fn a_dictionary_sweep_reads_every_value_and_keeps_it_under_the_budget() {
7654 let path = path("dictionary-sweep");
7655 let spellings = (0..2_500)
7658 .map(|index| Value::Varchar(format!("value {index:08} {}", "x".repeat(index % 40))))
7659 .collect::<Vec<_>>();
7660 let mut writer =
7661 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
7662 .expect("new file");
7663 for part in spellings.chunks(1_024) {
7666 writer
7667 .append(
7668 &Chunk::new(vec![
7669 Vector::from_values(LogicalType::Varchar, part).expect("strings"),
7670 ])
7671 .expect("one column"),
7672 )
7673 .expect("stripe written");
7674 }
7675 writer.finish().expect("commit");
7676
7677 let reader = Reader::open(&path).expect("valid directory");
7678 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
7679 assert_eq!(dictionary.len(), spellings.len(), "every value is distinct");
7680
7681 let resting = dictionary.footprint();
7682 let mut swept: Vec<Vec<u8>> = Vec::new();
7683 let mut at = 0;
7684 let mut calls = 0;
7685 while at < dictionary.len() {
7686 let stopped = dictionary
7687 .sweep_text(at, dictionary.len(), &mut |index: usize, text: &[u8]| {
7688 assert_eq!(index, swept.len(), "a sweep hands its values over in order");
7689 swept.push(text.to_vec());
7690 Ok(())
7691 })
7692 .expect("a sweep reads");
7693 assert!(stopped > at, "a sweep moves");
7694 at = stopped;
7695 calls += 1;
7696 }
7697 assert_eq!(calls, 3, "a sweep hands over one block at a time");
7698 let after = dictionary.footprint();
7699 assert!(after > resting, "a sweep under the budget keeps what it decoded");
7700
7701 let read = (0..dictionary.len())
7702 .map(|code| dictionary.try_bytes_at(code).expect("read").expect("a value").to_vec())
7703 .collect::<Vec<_>>();
7704 assert_eq!(swept, read, "a sweep answers what a point read answers");
7705 assert_eq!(dictionary.footprint(), after, "a point read of a kept block decodes nothing");
7706 fs::remove_file(path).expect("remove scratch file");
7707 }
7708
7709 #[test]
7720 fn a_sweep_over_a_block_with_a_short_second_run_reads_what_a_point_read_reads() {
7721 let path = path("dictionary-sweep-short-run");
7722 let spellings = (0..2_800)
7723 .map(|index| Value::Varchar(format!("value {index:08} {}", "x".repeat(index % 40))))
7724 .collect::<Vec<_>>();
7725 let mut writer =
7726 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
7727 .expect("new file");
7728 for part in spellings.chunks(1_024) {
7729 writer
7730 .append(
7731 &Chunk::new(vec![
7732 Vector::from_values(LogicalType::Varchar, part).expect("strings"),
7733 ])
7734 .expect("one column"),
7735 )
7736 .expect("stripe written");
7737 }
7738 writer.finish().expect("commit");
7739
7740 let reader = Reader::open(&path).expect("valid directory");
7741 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
7742 assert_eq!(dictionary.len(), spellings.len(), "every value is distinct");
7743 let last = dictionary.len() % TEXT_PAYLOAD_VALUES;
7744 assert!(last > TEXT_OFFSET_RUN, "the last block has to reach into a second run of offsets");
7745 assert!(last < TEXT_PAYLOAD_VALUES, "and that second run has to be short of a whole one");
7746
7747 let mut swept: Vec<Vec<u8>> = Vec::new();
7748 let mut at = 0;
7749 while at < dictionary.len() {
7750 let stopped = dictionary
7751 .sweep_text(at, dictionary.len(), &mut |index: usize, text: &[u8]| {
7752 assert_eq!(index, swept.len(), "a sweep hands its values over in order");
7753 swept.push(text.to_vec());
7754 Ok(())
7755 })
7756 .expect("a sweep reads");
7757 assert!(stopped > at, "a sweep moves");
7758 at = stopped;
7759 }
7760 let read = (0..dictionary.len())
7761 .map(|code| dictionary.try_bytes_at(code).expect("read").expect("a value").to_vec())
7762 .collect::<Vec<_>>();
7763 assert_eq!(swept, read, "a sweep answers what a point read answers");
7764 fs::remove_file(path).expect("remove scratch file");
7765 }
7766
7767 #[test]
7777 fn narrowing_a_page_takes_what_fits_and_refuses_what_does_not() {
7778 assert_eq!(fit::<i8>(&[]).expect("an empty page fits anything"), Vec::<i8>::new());
7779 assert_eq!(fit::<i8>(&[-128, 0, 127]).expect("the edges fit"), vec![-128_i8, 0, 127]);
7780 fit::<i8>(&[128]).expect_err("one past the top does not fit");
7781 fit::<i8>(&[-129]).expect_err("one past the bottom does not fit");
7782 assert_eq!(fit::<u8>(&[0, 255]).expect("the edges fit"), vec![0_u8, 255]);
7783 fit::<u8>(&[256]).expect_err("one past the top does not fit");
7784 fit::<u8>(&[-1]).expect_err("a negative does not fit an unsigned page");
7785 assert_eq!(
7786 fit::<i16>(&[-32_768, 0, 32_767]).expect("the edges fit"),
7787 vec![-32_768_i16, 0, 32_767]
7788 );
7789 fit::<i16>(&[32_768]).expect_err("one past the top does not fit");
7790 fit::<i16>(&[-32_769]).expect_err("one past the bottom does not fit");
7791 assert_eq!(fit::<u16>(&[0, 65_535]).expect("the edges fit"), vec![0_u16, 65_535]);
7792 fit::<u16>(&[65_536]).expect_err("one past the top does not fit");
7793 fit::<u16>(&[-1]).expect_err("a negative does not fit an unsigned page");
7794 assert_eq!(
7795 fit::<i32>(&[i64::from(i32::MIN), 0, i64::from(i32::MAX)]).expect("the edges fit"),
7796 vec![i32::MIN, 0, i32::MAX]
7797 );
7798 fit::<i32>(&[i64::from(i32::MAX) + 1]).expect_err("one past the top does not fit");
7799 fit::<i32>(&[i64::from(i32::MIN) - 1]).expect_err("one past the bottom does not fit");
7800 assert_eq!(
7801 fit::<u32>(&[0, 4_294_967_295]).expect("the edges fit"),
7802 vec![0_u32, 4_294_967_295]
7803 );
7804 fit::<u32>(&[4_294_967_296]).expect_err("one past the top does not fit");
7805 fit::<u32>(&[-1]).expect_err("a negative does not fit an unsigned page");
7806
7807 fit::<i8>(&[0, 1, 2, 128, 3]).expect_err("one bad value spoils the page");
7810 }
7811
7812 #[test]
7819 fn the_residue_agrees_with_a_checked_conversion_everywhere() {
7820 for value in -70_000_i64..70_000 {
7821 assert_eq!(fit::<i8>(&[value]).is_ok(), i8::try_from(value).is_ok(), "{value} as i8");
7822 assert_eq!(fit::<u8>(&[value]).is_ok(), u8::try_from(value).is_ok(), "{value} as u8");
7823 assert_eq!(fit::<i16>(&[value]).is_ok(), i16::try_from(value).is_ok(), "{value} i16");
7824 assert_eq!(fit::<u16>(&[value]).is_ok(), u16::try_from(value).is_ok(), "{value} u16");
7825 }
7826 let wide = [i64::MIN, i64::MIN + 1, i64::from(i32::MIN), 0, i64::from(u32::MAX), i64::MAX];
7827 for edge in wide {
7828 for step in -2_i64..=2 {
7829 let value = edge.saturating_add(step);
7830 assert_eq!(
7831 fit::<i32>(&[value]).is_ok(),
7832 i32::try_from(value).is_ok(),
7833 "{value} as i32"
7834 );
7835 assert_eq!(
7836 fit::<u32>(&[value]).is_ok(),
7837 u32::try_from(value).is_ok(),
7838 "{value} as u32"
7839 );
7840 }
7841 }
7842 }
7843
7844 #[test]
7852 fn a_dictionary_at_its_budget_sweeps_without_keeping() {
7853 let path = path("dictionary-budget");
7854 let spellings = (0..2_500)
7855 .map(|index| Value::Varchar(format!("value {index:08} {}", "y".repeat(index % 40))))
7856 .collect::<Vec<_>>();
7857 let mut writer =
7858 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
7859 .expect("new file");
7860 for part in spellings.chunks(1_024) {
7861 writer
7862 .append(
7863 &Chunk::new(vec![
7864 Vector::from_values(LogicalType::Varchar, part).expect("strings"),
7865 ])
7866 .expect("one column"),
7867 )
7868 .expect("stripe written");
7869 }
7870 writer.finish().expect("commit");
7871
7872 let reader = Reader::open(&path).expect("valid directory");
7873 let page = reader.table.dictionaries[0].expect("a string column has one");
7874 let file = Arc::clone(&reader.file);
7875 let starved = open_global_dictionary(file, page, &LogicalType::Varchar, 0)
7876 .expect("a dictionary opens whatever it may keep");
7877
7878 let resting = starved.footprint();
7879 let mut swept: Vec<Vec<u8>> = Vec::new();
7880 let mut at = 0;
7881 while at < starved.len() {
7882 at = starved
7883 .sweep_text(at, starved.len(), &mut |_index: usize, text: &[u8]| {
7884 swept.push(text.to_vec());
7885 Ok(())
7886 })
7887 .expect("a sweep reads");
7888 }
7889 assert_eq!(swept.len(), spellings.len(), "a starved sweep still reads every value");
7890 assert_eq!(starved.footprint(), resting, "and keeps no block it decoded");
7891
7892 let generous = reader.dictionary(0).expect("read").expect("a string column has one");
7893 let read = (0..generous.len())
7894 .map(|code| generous.try_bytes_at(code).expect("read").expect("a value").to_vec())
7895 .collect::<Vec<_>>();
7896 assert_eq!(swept, read, "a starved sweep answers what a point read answers");
7897 fs::remove_file(path).expect("remove scratch file");
7898 }
7899
7900 #[test]
7901 fn damaged_membership_cannot_skip_a_string_page() {
7902 let path = path("damaged-membership");
7903 let mut writer = Writer::create(
7904 &path,
7905 "items",
7906 vec![
7907 Field::required("id", LogicalType::Integer),
7908 Field::new("text", LogicalType::Varchar),
7909 ],
7910 )
7911 .expect("new file");
7912 writer.append(&sample()).expect("stripe written");
7913 writer.finish().expect("commit");
7914
7915 let reader = Reader::open(&path).expect("valid directory");
7916 let membership = reader.table.stripes[0].memberships[1].expect("string membership");
7917 let mut file = OpenOptions::new().write(true).open(&path).expect("open membership page");
7918 file.seek(SeekFrom::Start(membership.offset)).expect("membership start");
7919 file.write_all(&[255]).expect("damage membership");
7920 let error = reader.skips_codes(0, 1, &[3]).expect_err("corruption must not skip rows");
7921 assert!(error.message().contains("membership page checksum differs"), "{error}");
7922 fs::remove_file(path).expect("remove scratch file");
7923 }
7924
7925 #[test]
7926 fn membership_delta_stream_is_sorted_exact_and_bounded() {
7927 let unique = unique_codes(&[900, 4, 4, 72, 9, u32::MAX]);
7928 assert_eq!(unique, [4, 9, 72, 900, u32::MAX]);
7929 let encoded = encode_membership(&unique);
7930 assert_eq!(
7931 decode_membership(&encoded).expect("valid membership"),
7932 [4, 9, 72, 900, u32::MAX]
7933 );
7934 let merged = merged_codes(vec![vec![4, 900], vec![9, 900, u32::MAX], vec![72]]);
7937 assert_eq!(merged, [4, 9, 72, 900, u32::MAX]);
7938 assert_eq!(
7939 decode_membership(&encode_membership(&merged)).expect("valid membership"),
7940 unique
7941 );
7942 assert!(decode_membership(&[1, 0x80]).is_err(), "a truncated varint is invalid");
7943 assert!(
7944 decode_membership(&[1, 0xff, 0xff, 0xff, 0xff, 0x10]).is_err(),
7945 "a value past u32 is invalid"
7946 );
7947 }
7948
7949 #[test]
7950 fn a_global_dictionary_may_be_larger_than_one_column_page() {
7951 let dictionary = Page {
7952 offset: HEADER,
7953 length: u32::try_from(MAX_PAGE + 1).expect("the page bound fits on disk"),
7954 hash: 0,
7955 };
7956 let table = Table {
7957 name: "items".to_owned(),
7958 fields: vec![Field::new("text", LogicalType::Varchar)],
7959 stripes: Vec::new(),
7960 rows: 0,
7961 dictionaries: vec![Some(dictionary)],
7962 distincts: vec![None],
7963 frequencies: vec![None],
7964 clustering: None,
7965 };
7966 let directory = encode_directory(&table).expect("directory");
7967 let file_size = dictionary.offset + u64::from(dictionary.length) + 1;
7968
7969 let decoded = decode_directory(&directory, file_size).expect("large lazy dictionary");
7970 assert_eq!(decoded.dictionaries[0].expect("dictionary").length, dictionary.length);
7971 }
7972
7973 #[test]
7974 fn a_column_with_one_value_everywhere_costs_almost_nothing_a_row() {
7975 let path = path("constant-codes");
7976 let mut writer =
7977 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
7978 .expect("new file");
7979 let empty = vec![Value::Varchar(String::new()); 1024];
7980 for _ in 0..4 {
7981 let column = Vector::from_values(LogicalType::Varchar, &empty).expect("strings");
7982 writer.append(&Chunk::new(vec![column]).expect("one column")).expect("a part");
7983 }
7984 writer.finish().expect("commit");
7985
7986 let reader = Reader::open(&path).expect("valid directory");
7987 let pages = reader.layout().columns.first().expect("one column").pages;
7988 assert!(pages < 256, "{pages} bytes of pages for 4,096 rows of one value");
7992 let read = reader.read(3, &[0]).expect("the last part back");
7993 assert_eq!(read.value_at(0, 0), Value::Varchar(String::new()));
7994 assert_eq!(read.value_at(1023, 0), Value::Varchar(String::new()));
7995 fs::remove_file(path).expect("remove scratch file");
7996 }
7997
7998 #[test]
7999 fn a_cascade_value_too_wide_for_its_column_is_refused_rather_than_cut() {
8000 let over = vec![i64::from(i32::MAX) + 1];
8003 let error = narrowed(&LogicalType::Integer, over).expect_err("a page that disagrees");
8004 assert!(format!("{error}").contains("not of its type"), "{error}");
8005 assert!(narrowed(&LogicalType::BigInt, vec![i64::MIN]).is_ok(), "bigint holds all of i64");
8006 assert!(narrowed(&LogicalType::Varchar, vec![0]).is_err(), "strings are not integers");
8007 }
8008
8009 #[test]
8010 fn a_code_stream_the_cascade_cannot_shrink_is_left_alone() {
8011 let mut state: u32 = 0x9e37_79b9;
8015 let spread: Vec<u32> = (0..1024)
8016 .map(|_| {
8017 state ^= state << 13;
8018 state ^= state >> 17;
8019 state ^= state << 5;
8020 state
8021 })
8022 .collect();
8023 assert_eq!(encoded_codes(&spread).expect("no failure"), None);
8024 let near: Vec<u32> = (0..1024).collect();
8025 let coded = encoded_codes(&near).expect("no failure").expect("counting up is packable");
8026 assert!(coded.len() < near.len() * 4, "{} bytes for a run of 1,024", coded.len());
8027 }
8028
8029 #[test]
8035 fn two_writes_of_the_same_rows_give_the_same_bytes() {
8036 fn written(path: &PathBuf) {
8037 let fields = (0..40)
8038 .map(|column| {
8039 let ty =
8040 if column % 4 == 0 { LogicalType::Varchar } else { LogicalType::BigInt };
8041 Field::new(format!("c{column}"), ty)
8042 })
8043 .collect::<Vec<_>>();
8044 let mut writer = Writer::create(path, "wide", fields).expect("new file");
8045 for part in 0..70_u64 {
8046 let columns = (0..40)
8047 .map(|column| {
8048 let values = (0..64_u64)
8049 .map(|row| {
8050 let seed = part.wrapping_mul(31).wrapping_add(row);
8051 if column % 4 == 0 {
8052 Value::Varchar(format!("v{}", seed % 17))
8053 } else {
8054 Value::BigInt(i64::try_from(seed % 97).expect("small"))
8055 }
8056 })
8057 .collect::<Vec<_>>();
8058 let ty = if column % 4 == 0 {
8059 LogicalType::Varchar
8060 } else {
8061 LogicalType::BigInt
8062 };
8063 Vector::from_values(ty, &values).expect("a column")
8064 })
8065 .collect::<Vec<_>>();
8066 writer.append(&Chunk::new(columns).expect("forty columns")).expect("a part");
8067 }
8068 writer.finish().expect("commit");
8069 }
8070
8071 let first = path("repeatable-one");
8072 let second = path("repeatable-two");
8073 written(&first);
8074 written(&second);
8075 let left = fs::read(&first).expect("the first file");
8076 let right = fs::read(&second).expect("the second file");
8077 assert_eq!(left.len(), right.len(), "two writes of the same rows differ in length");
8078 assert!(left == right, "two writes of the same rows differ in their bytes");
8079
8080 let reader = Reader::open(&first).expect("valid directory");
8083 assert_eq!(reader.table().rows(), 70 * 64);
8084 let read = reader.read(0, &[0, 1]).expect("the first part back");
8085 assert_eq!(read.value_at(0, 0), Value::Varchar("v0".to_owned()));
8086 assert_eq!(read.value_at(0, 1), Value::BigInt(0));
8087 fs::remove_file(first).expect("remove scratch file");
8088 fs::remove_file(second).expect("remove scratch file");
8089 }
8090
8091 fn three_tables(path: &PathBuf) {
8093 let writer = Writer::create(
8094 path,
8095 "region",
8096 vec![
8097 Field::new("r_key", LogicalType::Integer),
8098 Field::new("r_name", LogicalType::Varchar),
8099 ],
8100 )
8101 .expect("new file");
8102 let mut writer = writer;
8103 writer
8104 .append(
8105 &Chunk::new(vec![
8106 Vector::from_values(
8107 LogicalType::Integer,
8108 &[Value::Integer(0), Value::Integer(1)],
8109 )
8110 .expect("keys"),
8111 Vector::from_values(
8112 LogicalType::Varchar,
8113 &[Value::Varchar("AFRICA".to_owned()), Value::Varchar("ASIA".to_owned())],
8114 )
8115 .expect("names"),
8116 ])
8117 .expect("two columns"),
8118 )
8119 .expect("a part");
8120 let mut writer = writer
8121 .next("empty", vec![Field::new("nothing", LogicalType::BigInt)])
8122 .expect("a second table");
8123 writer
8124 .append(
8125 &Chunk::new(vec![
8126 Vector::from_values(LogicalType::BigInt, &[Value::BigInt(7)]).expect("a row"),
8127 ])
8128 .expect("one column"),
8129 )
8130 .expect("a part");
8131 let mut writer =
8132 writer.next("wide", vec![Field::new("n", LogicalType::BigInt)]).expect("a third table");
8133 for part in 0..70_i64 {
8134 let values = (0..64).map(|row| Value::BigInt(part * 64 + row)).collect::<Vec<_>>();
8135 writer
8136 .append(
8137 &Chunk::new(vec![
8138 Vector::from_values(LogicalType::BigInt, &values).expect("a column"),
8139 ])
8140 .expect("one column"),
8141 )
8142 .expect("a part");
8143 }
8144 writer.finish().expect("commit");
8145 }
8146
8147 #[test]
8148 fn three_tables_in_one_file_read_back_by_name() {
8149 let file = path("three-tables");
8150 three_tables(&file);
8151 let catalog = Catalog::open(&file).expect("a committed catalog");
8152 assert_eq!(catalog.names().collect::<Vec<_>>(), ["region", "empty", "wide"]);
8153
8154 let region = catalog.table("region").expect("the first table");
8155 assert_eq!(region.table().rows(), 2);
8156 assert_eq!(
8157 region.read(0, &[1]).expect("names").value_at(1, 0),
8158 Value::Varchar("ASIA".to_owned())
8159 );
8160
8161 let wide = catalog.table("wide").expect("the third table");
8162 assert_eq!(wide.table().rows(), 70 * 64);
8163 assert_eq!(wide.read(0, &[0]).expect("the first part").value_at(0, 0), Value::BigInt(0));
8164
8165 let empty = catalog.table("empty").expect("the second table");
8168 assert_eq!(empty.table().rows(), 1);
8169 assert_eq!(empty.read(0, &[0]).expect("the row").value_at(0, 0), Value::BigInt(7));
8170
8171 fs::remove_file(file).expect("remove scratch file");
8172 }
8173
8174 #[test]
8175 fn a_name_the_file_does_not_hold_is_an_error_rather_than_the_first_table() {
8176 let file = path("three-tables-missing");
8177 three_tables(&file);
8178 let catalog = Catalog::open(&file).expect("a committed catalog");
8179 let error = catalog.table("nation").expect_err("no such table");
8180 assert!(error.message().contains("nation"), "{}", error.message());
8181 fs::remove_file(file).expect("remove scratch file");
8182 }
8183
8184 #[test]
8185 fn a_file_of_three_tables_will_not_open_as_one() {
8186 let file = path("three-tables-unnamed");
8187 three_tables(&file);
8188 let error = Reader::open(&file).expect_err("more than one table");
8189 assert!(error.message().contains("more than one table"), "{}", error.message());
8190 fs::remove_file(file).expect("remove scratch file");
8191 }
8192
8193 #[test]
8195 fn decimals_of_every_storage_width_round_trip() {
8196 let file = path("decimals");
8197 let widths = [(4_u8, 2_u8), (9, 2), (18, 4), (38, 6)];
8198 let fields = widths
8199 .iter()
8200 .enumerate()
8201 .map(|(index, (width, scale))| {
8202 Field::new(
8203 format!("d{index}"),
8204 LogicalType::decimal(*width, *scale).expect("a decimal type"),
8205 )
8206 })
8207 .collect::<Vec<_>>();
8208 let mut writer = Writer::create(&file, "money", fields).expect("new file");
8209 let rows: [i128; 3] = [-1234, 0, 999];
8210 let columns = widths
8211 .iter()
8212 .map(|(width, scale)| {
8213 let values = rows
8214 .iter()
8215 .map(|unscaled| Value::Decimal {
8216 unscaled: *unscaled,
8217 width: *width,
8218 scale: *scale,
8219 })
8220 .collect::<Vec<_>>();
8221 Vector::from_values(
8222 LogicalType::decimal(*width, *scale).expect("a decimal type"),
8223 &values,
8224 )
8225 .expect("a decimal column")
8226 })
8227 .collect::<Vec<_>>();
8228 writer.append(&Chunk::new(columns).expect("four columns")).expect("a part");
8229 writer.finish().expect("commit");
8230
8231 let reader = Reader::open(&file).expect("a committed file");
8232 for (index, (width, scale)) in widths.iter().enumerate() {
8233 assert_eq!(
8234 reader.table().fields()[index].ty,
8235 LogicalType::decimal(*width, *scale).expect("a decimal type"),
8236 "column {index} came back as another type"
8237 );
8238 let column = reader.read(0, &[index]).expect("the column");
8239 for (row, unscaled) in rows.iter().enumerate() {
8240 assert_eq!(
8241 column.value_at(row, 0),
8242 Value::Decimal { unscaled: *unscaled, width: *width, scale: *scale },
8243 "column {index} row {row}"
8244 );
8245 }
8246 }
8247 fs::remove_file(file).expect("remove scratch file");
8248 }
8249
8250 #[test]
8251 fn two_tables_of_one_name_are_refused_before_anything_is_committed() {
8252 let file = path("two-of-a-name");
8253 let writer = Writer::create(&file, "t", vec![Field::new("a", LogicalType::BigInt)])
8254 .expect("new file");
8255 let error = writer
8256 .next("t", vec![Field::new("a", LogicalType::BigInt)])
8257 .expect_err("the same name twice");
8258 assert!(error.message().contains("same name"), "{}", error.message());
8259 fs::remove_file(file).expect("remove scratch file");
8260 }
8261
8262 #[test]
8263 fn opening_the_catalog_reads_no_table_directory() {
8264 let file = path("catalog-only");
8265 three_tables(&file);
8266 let catalog = Catalog::open(&file).expect("a committed catalog");
8267 assert_eq!(catalog.opening.reads, 2, "opening the catalog read more than the slot");
8270 assert_eq!(catalog.names().len(), 3);
8271 fs::remove_file(file).expect("remove scratch file");
8272 }
8273
8274 #[test]
8285 fn the_checksum_answers_what_it_has_always_answered() {
8286 let bytes: Vec<u8> =
8287 (0..1000_u32).map(|at| (at.wrapping_mul(31).wrapping_add(7) % 251) as u8).collect();
8288 for (length, expected) in [
8289 (0, 0xef46_db37_51d8_e999),
8290 (1, 0xa96c_7f0c_e858_bbb7),
8291 (3, 0x56e6_9576_32a4_87f9),
8292 (4, 0xc60d_15b1_e3ff_8f04),
8293 (5, 0x8088_1585_8624_dd4e),
8294 (7, 0xafbe_fc3d_6c6f_9a8e),
8295 (8, 0x3da5_c7aa_2696_83e0),
8296 (9, 0x465e_c429_b13c_3892),
8297 (15, 0xdee8_9d8a_065a_6233),
8298 (16, 0x1330_489a_7767_9c80),
8299 (31, 0x3391_303d_485e_846e),
8300 (32, 0x40b7_aff7_5d45_bbc8),
8301 (33, 0x4997_cae4_951c_17a5),
8302 (39, 0x5807_28fd_5c14_5739),
8303 (40, 0xf95c_f6f5_c08a_3d3b),
8304 (63, 0x2944_b4da_fc69_b206),
8305 (64, 0xbb76_f6ef_19bd_5a1b),
8306 (65, 0x814e_0c65_4a9f_d640),
8307 (127, 0x00de_aab1_31cf_f89b),
8308 (1000, 0x9e33_00c1_cde3_c58d),
8309 ] {
8310 assert_eq!(checksum(&bytes[..length]), expected, "the checksum of {length} bytes");
8311 }
8312 assert_eq!(checksum(b"the quick brown fox jumps over the lazy dog"), 0xed71_4233_c5a9_a792);
8313 }
8314 #[test]
8321 fn a_declared_order_comes_back_out_of_the_file() {
8322 let path = path("clustered");
8323 let shipped = vec![
8324 Field::new("key", LogicalType::BigInt),
8325 Field::new("line", LogicalType::Integer),
8326 Field::new("shipdate", LogicalType::Date),
8327 ];
8328 let plain = vec![Field::new("a", LogicalType::Integer)];
8329 let stage_zero =
8330 Clustering::new(vec![2, 0, 1], Width::Month, shipped.len()).expect("valid");
8331
8332 let mut writer = Writer::create(&path, "lineitem", shipped)
8333 .expect("new file")
8334 .declare(stage_zero.clone())
8335 .expect("the columns are the table's");
8336 let column = |ty: LogicalType, values: &[Value]| {
8337 Vector::from_values(ty, values).expect("the values match the type")
8338 };
8339 writer
8340 .append(
8341 &Chunk::new(vec![
8342 column(
8343 LogicalType::BigInt,
8344 &[Value::BigInt(0), Value::BigInt(1), Value::BigInt(2), Value::BigInt(3)],
8345 ),
8346 column(
8347 LogicalType::Integer,
8348 &[
8349 Value::Integer(1),
8350 Value::Integer(1),
8351 Value::Integer(1),
8352 Value::Integer(1),
8353 ],
8354 ),
8355 column(
8356 LogicalType::Date,
8357 &[Value::Date(0), Value::Date(1), Value::Date(2), Value::Date(3)],
8358 ),
8359 ])
8360 .expect("three columns"),
8361 )
8362 .expect("four rows");
8363 let mut writer = writer.next("nation", plain).expect("a second table");
8364 writer
8365 .append(
8366 &Chunk::new(vec![column(LogicalType::Integer, &[Value::Integer(7)])])
8367 .expect("one column"),
8368 )
8369 .expect("one row");
8370 writer.finish().expect("commit");
8371
8372 let catalog = Catalog::open(&path).expect("reopen");
8373 let lineitem = catalog.table("lineitem").expect("the clustered table");
8374 assert_eq!(lineitem.table().clustering(), Some(&stage_zero));
8375 let nation = catalog.table("nation").expect("the plain table");
8376 assert_eq!(nation.table().clustering(), None, "nobody declared one here");
8377
8378 assert_eq!(lineitem.table().rows(), 4);
8381 assert_eq!(nation.table().rows(), 1);
8382 fs::remove_file(&path).ok();
8383 }
8384
8385 #[test]
8387 fn a_declaration_off_the_end_of_the_table_never_reaches_the_file() {
8388 let path = path("clustered-bad");
8389 let writer = Writer::create(&path, "items", vec![Field::new("a", LogicalType::Integer)])
8390 .expect("new file");
8391 let wrong = Clustering::new(vec![3], Width::Exact, 4).expect("valid against four columns");
8392 assert!(writer.declare(wrong).is_err(), "the table has one column, not four");
8393 fs::remove_file(&path).ok();
8394 }
8395}