1#![forbid(unsafe_code)]
29
30use std::cmp::Ordering;
31use std::collections::{HashMap, VecDeque};
32use std::fs::{File, OpenOptions};
33use std::io::{Read, Seek, SeekFrom};
34use std::mem::{size_of, size_of_val};
35use std::path::Path;
36use std::sync::atomic::{AtomicUsize, Ordering as Atomic};
37use std::sync::{Arc, Mutex, OnceLock};
38
39use rudb_common::bounds::{Bound, Op};
40use rudb_common::{Error, Field, LogicalType, Result, Value};
41use rudb_encoding::{bitpack, chooser, integer, string};
42use rudb_storage::sieve::Sieve;
43use rudb_storage::{Probe, Range, Zone};
44use rudb_vector::string::StringColumn;
45use rudb_vector::validity::Validity;
46use rudb_vector::{Buffer, Chunk, Data, Packed, TextSource, Vector};
47
48const MAGIC: &[u8; 8] = b"RUDBNV10";
49const DIRECTORY: &[u8; 8] = b"RUDBDI10";
50const FORMAT: u32 = 19;
51const HEADER: u64 = 80;
52const SLOT_BYTES: usize = 28;
53const MAX_PAGE: usize = 256 * 1024 * 1024;
54const MAX_DIRECTORY: usize = 128 * 1024 * 1024;
55const FREQUENCIES: &[u8; 8] = b"RUDBFQ2\0";
56const FREQUENCY_CANDIDATES: usize = 32_768;
57const FREQUENCY_ENTRIES: usize = 512;
58const FREQUENCY_BUILD_RANK: usize = 10;
59const FREQUENCY_ORDINALS: usize = 65_536;
60const MAX_FREQUENCY_WORKERS: usize = 32;
67
68const MAX_ENCODE_WORKERS: usize = 32;
75
76const SIEVE_BUDGET: usize = 8 * 1024;
84
85fn io(error: std::io::Error) -> Error {
86 Error::io(error.to_string())
87}
88
89fn invalid(message: &str) -> Error {
90 Error::invalid_input(format!("invalid rudb native file: {message}"))
91}
92
93fn sum(counts: impl Iterator<Item = u64>) -> u64 {
95 counts.fold(0, u64::saturating_add)
96}
97
98fn span_bytes(spans: &[Span], at: usize) -> u64 {
100 spans.get(at).map_or(0, |span| u64::from(span.length))
101}
102
103fn page_bytes(pages: &[Option<Page>], at: usize) -> u64 {
105 pages.get(at).and_then(Option::as_ref).map_or(0, Page::bytes)
106}
107
108fn checksum(bytes: &[u8]) -> u64 {
109 const P1: u64 = 11_400_714_785_074_694_791;
110 const P2: u64 = 14_029_467_366_897_019_727;
111 const P3: u64 = 1_609_587_929_392_839_161;
112 const P4: u64 = 9_650_029_242_287_828_579;
113 const P5: u64 = 2_870_177_450_012_600_261;
114 let round = |state: u64, word: u64| {
115 state.wrapping_add(word.wrapping_mul(P2)).rotate_left(31).wrapping_mul(P1)
116 };
117 let merge = |state: u64, lane: u64| (state ^ round(0, lane)).wrapping_mul(P1).wrapping_add(P4);
118 let word =
119 |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("eight checksum bytes"));
120
121 let mut at = 0;
122 let mut hash = if bytes.len() >= 32 {
123 let mut one = P1.wrapping_add(P2);
124 let mut two = P2;
125 let mut three = 0;
126 let mut four = 0_u64.wrapping_sub(P1);
127 while at + 32 <= bytes.len() {
128 one = round(one, word(at));
129 two = round(two, word(at + 8));
130 three = round(three, word(at + 16));
131 four = round(four, word(at + 24));
132 at += 32;
133 }
134 let combined = one
135 .rotate_left(1)
136 .wrapping_add(two.rotate_left(7))
137 .wrapping_add(three.rotate_left(12))
138 .wrapping_add(four.rotate_left(18));
139 merge(merge(merge(merge(combined, one), two), three), four)
140 } else {
141 P5
142 };
143 hash = hash.wrapping_add(bytes.len() as u64);
144 while at + 8 <= bytes.len() {
145 hash ^= round(0, word(at));
146 hash = hash.rotate_left(27).wrapping_mul(P1).wrapping_add(P4);
147 at += 8;
148 }
149 if at + 4 <= bytes.len() {
150 let tail = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four checksum bytes"));
151 hash ^= u64::from(tail).wrapping_mul(P1);
152 hash = hash.rotate_left(23).wrapping_mul(P2).wrapping_add(P3);
153 at += 4;
154 }
155 while at < bytes.len() {
156 hash ^= u64::from(bytes[at]).wrapping_mul(P5);
157 hash = hash.rotate_left(11).wrapping_mul(P1);
158 at += 1;
159 }
160 hash ^= hash >> 33;
161 hash = hash.wrapping_mul(P2);
162 hash ^= hash >> 29;
163 hash = hash.wrapping_mul(P3);
164 hash ^ (hash >> 32)
165}
166
167#[derive(Debug, Clone, Copy)]
168struct Slot {
169 offset: u64,
170 length: u32,
171 generation: u64,
172 hash: u64,
173}
174
175impl Slot {
176 fn bytes(self) -> [u8; SLOT_BYTES] {
177 let mut result = [0; SLOT_BYTES];
178 result[..8].copy_from_slice(&self.offset.to_le_bytes());
179 result[8..12].copy_from_slice(&self.length.to_le_bytes());
180 result[12..20].copy_from_slice(&self.generation.to_le_bytes());
181 result[20..28].copy_from_slice(&self.hash.to_le_bytes());
182 result
183 }
184
185 fn read(bytes: &[u8]) -> Self {
186 Self {
187 offset: u64::from_le_bytes(bytes[..8].try_into().expect("eight bytes")),
188 length: u32::from_le_bytes(bytes[8..12].try_into().expect("four bytes")),
189 generation: u64::from_le_bytes(bytes[12..20].try_into().expect("eight bytes")),
190 hash: u64::from_le_bytes(bytes[20..28].try_into().expect("eight bytes")),
191 }
192 }
193}
194
195#[derive(Debug, Clone, Copy)]
196struct Page {
197 offset: u64,
198 length: u32,
199 hash: u64,
200}
201
202impl Page {
203 fn bytes(&self) -> u64 {
205 u64::from(self.length)
206 }
207}
208
209#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
210enum FrequencyValue {
211 Null,
212 Integer(i128),
213 Code(u32),
214}
215
216#[derive(Debug, Clone)]
217struct FrequencyEntry {
218 value: FrequencyValue,
219 count: u64,
220}
221
222#[derive(Debug, Clone)]
227struct FrequencySummary {
228 entries: Vec<FrequencyEntry>,
229 omitted_max: u64,
230 ordinals: Vec<u64>,
231}
232
233#[derive(Debug, Clone, PartialEq, Eq)]
235pub struct FrequencyOccurrences {
236 pub omitted_max: u64,
238 pub ordinals: Vec<u64>,
240}
241
242#[derive(Debug, Clone, Copy, Default)]
249struct Span {
250 offset: u64,
251 length: u32,
252}
253
254#[derive(Debug, Clone)]
256pub struct Stripe {
257 rows: usize,
258 parts: Vec<u32>,
261 index: Span,
265 pages: Vec<Span>,
266 memberships: Vec<Option<Page>>,
267 sieves: Vec<Option<Page>>,
270 zone: Zone,
271}
272
273impl Stripe {
274 #[must_use]
276 pub fn rows(&self) -> usize {
277 self.rows
278 }
279
280 #[must_use]
282 pub fn parts(&self) -> usize {
283 self.parts.len()
284 }
285}
286
287#[derive(Debug, Clone)]
289pub struct Table {
290 name: String,
291 fields: Vec<Field>,
292 stripes: Vec<Stripe>,
293 rows: usize,
294 dictionaries: Vec<Option<Page>>,
295 frequencies: Vec<Option<FrequencySummary>>,
296 distincts: Vec<Option<u64>>,
306}
307
308impl Table {
309 #[must_use]
311 pub fn name(&self) -> &str {
312 &self.name
313 }
314
315 #[must_use]
317 pub fn fields(&self) -> &[Field] {
318 &self.fields
319 }
320
321 #[must_use]
323 pub fn rows(&self) -> usize {
324 self.rows
325 }
326
327 #[must_use]
329 pub fn stripes(&self) -> &[Stripe] {
330 &self.stripes
331 }
332}
333
334#[derive(Debug, Clone)]
336pub struct ColumnLayout {
337 pub name: String,
339 pub kind: String,
341 pub pages: u64,
343 pub memberships: u64,
345 pub sieves: u64,
347 pub dictionary: u64,
349}
350
351impl ColumnLayout {
352 #[must_use]
354 pub fn total(&self) -> u64 {
355 self.pages
356 .saturating_add(self.memberships)
357 .saturating_add(self.sieves)
358 .saturating_add(self.dictionary)
359 }
360}
361
362#[derive(Debug, Clone)]
373pub struct Layout {
374 pub file: u64,
376 pub rows: usize,
378 pub stripes: usize,
380 pub parts: usize,
382 pub columns: Vec<ColumnLayout>,
384 pub indexes: u64,
387 pub directory: u64,
389 pub header: u64,
391}
392
393impl Layout {
394 #[must_use]
396 pub fn columns_total(&self) -> u64 {
397 self.columns.iter().map(ColumnLayout::total).fold(0, u64::saturating_add)
398 }
399
400 #[must_use]
406 pub fn unaccounted(&self) -> u64 {
407 self.file
408 .saturating_sub(self.columns_total())
409 .saturating_sub(self.indexes)
410 .saturating_sub(self.directory)
411 .saturating_sub(self.header)
412 }
413}
414
415#[derive(Debug)]
417struct GlobalDictionary {
418 primary: HashMap<u64, u32>,
419 collisions: HashMap<u64, Vec<u32>>,
420 offsets: Vec<u32>,
421 payload: Vec<u8>,
422 counts: Vec<u64>,
423 nulls: u64,
424}
425
426impl GlobalDictionary {
427 fn new() -> Self {
428 Self {
429 primary: HashMap::new(),
430 collisions: HashMap::new(),
431 offsets: vec![0],
432 payload: Vec::new(),
433 counts: Vec::new(),
434 nulls: 0,
435 }
436 }
437
438 fn bytes(&self, code: u32) -> Option<&[u8]> {
439 let start = *self.offsets.get(code as usize)? as usize;
440 let end = *self.offsets.get(code as usize + 1)? as usize;
441 self.payload.get(start..end)
442 }
443
444 fn code(&mut self, text: &str) -> Result<u32> {
445 let hash = checksum(text.as_bytes());
446 if let Some(&code) = self.primary.get(&hash) {
447 if self.bytes(code) == Some(text.as_bytes()) {
448 return Ok(code);
449 }
450 if let Some(codes) = self.collisions.get(&hash) {
451 if let Some(code) =
452 codes.iter().copied().find(|&code| self.bytes(code) == Some(text.as_bytes()))
453 {
454 return Ok(code);
455 }
456 }
457 let code = self.insert(text)?;
458 self.collisions.entry(hash).or_default().push(code);
459 return Ok(code);
460 }
461 let code = self.insert(text)?;
462 self.primary.insert(hash, code);
463 Ok(code)
464 }
465
466 fn insert(&mut self, text: &str) -> Result<u32> {
467 let code = u32::try_from(self.offsets.len() - 1)
468 .map_err(|_| invalid("global dictionary has too many values"))?;
469 self.payload.extend_from_slice(text.as_bytes());
470 self.offsets.push(
471 u32::try_from(self.payload.len())
472 .map_err(|_| invalid("global dictionary payload exceeds 4 GiB"))?,
473 );
474 self.counts.push(0);
475 Ok(code)
476 }
477
478 fn ranked(&self) -> Vec<(u64, u32)> {
498 let count = self.offsets.len() - 1;
499 let mut ranked = (0..count)
500 .map(|code| {
501 let code = code as u32;
502 (head(self.bytes(code).unwrap_or_default()), code)
503 })
504 .collect::<Vec<_>>();
505 ranked.sort_unstable_by(|left, right| {
506 left.0.cmp(&right.0).then_with(|| self.bytes(left.1).cmp(&self.bytes(right.1)))
507 });
508 ranked
509 }
510
511 fn observe(&mut self, code: u32, null: bool) -> Result<()> {
512 if null {
513 self.nulls = self.nulls.saturating_add(1);
514 return Ok(());
515 }
516 let count = self
517 .counts
518 .get_mut(code as usize)
519 .ok_or_else(|| invalid("global dictionary count code is out of range"))?;
520 *count = count.saturating_add(1);
521 Ok(())
522 }
523}
524
525#[derive(Debug)]
527pub struct Writer {
528 file: File,
529 at: u64,
537 table: Table,
538 generation: u64,
539 order: Vec<((u64, u64), (u64, u64))>,
542 next_order: u64,
543 dictionaries: Vec<Option<GlobalDictionary>>,
544 pending: Vec<PendingChunk>,
545}
546
547#[derive(Debug)]
555struct PendingChunk {
556 order: (u64, u64),
557 chunk: Chunk,
558}
559
560#[derive(Debug)]
566struct ColumnStripe {
567 pages: Vec<Vec<u8>>,
568 codes: Vec<Option<Vec<u32>>>,
569 sieves: Vec<Option<Sieve>>,
570 ranges: Vec<Range>,
571}
572
573fn weight(ty: &LogicalType) -> usize {
581 match ty {
582 LogicalType::Varchar | LogicalType::Blob => 64,
583 LogicalType::BigInt
584 | LogicalType::UBigInt
585 | LogicalType::Timestamp
586 | LogicalType::Double
587 | LogicalType::Decimal { .. } => 8,
588 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date | LogicalType::Float => 4,
589 LogicalType::SmallInt | LogicalType::USmallInt => 2,
590 _ => 1,
591 }
592}
593
594pub const STRIPE_PARTS: usize = 64;
601
602const INDEX_ENTRY: usize = size_of::<u32>() + size_of::<u64>();
604
605fn index_section(parts: usize) -> Result<usize> {
607 parts
608 .checked_mul(INDEX_ENTRY)
609 .and_then(|bytes| bytes.checked_add(size_of::<u64>()))
610 .ok_or_else(|| invalid("index page length overflow"))
611}
612
613impl Writer {
614 pub fn create(
620 path: impl AsRef<Path>,
621 name: impl Into<String>,
622 fields: Vec<Field>,
623 ) -> Result<Self> {
624 for field in &fields {
625 type_tag(&field.ty)?;
626 }
627 let file =
628 OpenOptions::new().write(true).read(true).create_new(true).open(path).map_err(io)?;
629 let mut header = [0; HEADER as usize];
630 header[..8].copy_from_slice(MAGIC);
631 header[8..12].copy_from_slice(&FORMAT.to_le_bytes());
632 write_at(&file, 0, &header)?;
633 Ok(Self {
634 file,
635 at: HEADER,
636 dictionaries: fields
637 .iter()
638 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
639 .collect(),
640 table: Table {
641 name: name.into(),
642 dictionaries: vec![None; fields.len()],
643 distincts: vec![None; fields.len()],
644 fields,
645 stripes: Vec::new(),
646 rows: 0,
647 frequencies: Vec::new(),
648 },
649 generation: 1,
650 order: Vec::new(),
651 next_order: 0,
652 pending: Vec::with_capacity(STRIPE_PARTS),
653 })
654 }
655
656 fn put(&mut self, bytes: &[u8]) -> Result<()> {
661 write_at(&self.file, self.at, bytes)?;
662 self.at = self
663 .at
664 .checked_add(bytes.len() as u64)
665 .ok_or_else(|| invalid("native file length overflow"))?;
666 Ok(())
667 }
668
669 pub fn append(&mut self, chunk: &Chunk) -> Result<()> {
675 let order = (self.next_order, 0);
676 self.next_order = self.next_order.saturating_add(1);
677 self.append_at(order, chunk)
678 }
679
680 pub fn append_at(&mut self, order: (u64, u64), chunk: &Chunk) -> Result<()> {
691 if chunk.is_empty() {
692 return Ok(());
693 }
694 self.admit(chunk)?;
695 if self.pending.last().is_some_and(|last| last.order > order) {
696 self.flush_pending()?;
697 }
698 self.pending.push(PendingChunk { order, chunk: chunk.clone() });
703 if self.pending.len() == STRIPE_PARTS {
704 self.flush_pending()?;
705 }
706 Ok(())
707 }
708
709 pub fn append_stripe(&mut self, parts: Vec<((u64, u64), Chunk)>) -> Result<()> {
725 if parts.len() > STRIPE_PARTS {
726 return Err(invalid("a stripe was handed more parts than it holds"));
727 }
728 self.flush_pending()?;
731 for (order, chunk) in parts {
732 if chunk.is_empty() {
733 continue;
734 }
735 self.admit(&chunk)?;
736 self.pending.push(PendingChunk { order, chunk });
737 }
738 self.flush_pending()
739 }
740
741 fn admit(&mut self, chunk: &Chunk) -> Result<()> {
743 if chunk.width() != self.table.fields.len() {
744 return Err(invalid("chunk width differs from table schema"));
745 }
746 for (index, field) in self.table.fields.iter().enumerate() {
747 if chunk.column(index)?.logical_type() != &field.ty {
748 return Err(invalid("chunk type differs from table schema"));
749 }
750 }
751 self.table.rows = self
752 .table
753 .rows
754 .checked_add(chunk.len())
755 .ok_or_else(|| invalid("row count overflow"))?;
756 Ok(())
757 }
758
759 fn encode_column(
767 index: usize,
768 held: &[PendingChunk],
769 mut dictionary: Option<&mut GlobalDictionary>,
770 ) -> Result<ColumnStripe> {
771 let mut stripe = ColumnStripe {
772 pages: Vec::with_capacity(held.len()),
773 codes: Vec::with_capacity(held.len()),
774 sieves: Vec::with_capacity(held.len()),
775 ranges: Vec::with_capacity(held.len()),
776 };
777 for pending in held {
778 let column = pending.chunk.column(index)?;
779 let (bytes, unique) = encode(column, dictionary.as_deref_mut())?;
780 if bytes.len() > MAX_PAGE {
781 return Err(invalid("column page exceeds the configured bound"));
782 }
783 let range = Range::of(column);
786 let sieve = match dictionary {
799 Some(_) => None,
800 None => Sieve::of(column, &range, SIEVE_BUDGET)
801 .filter(|sieve| sieve.len() < bytes.len()),
802 };
803 stripe.pages.push(bytes);
804 stripe.codes.push(unique);
805 stripe.sieves.push(sieve);
806 stripe.ranges.push(range);
807 }
808 Ok(stripe)
809 }
810
811 fn encode_columns(&mut self, held: &[PendingChunk]) -> Result<Vec<ColumnStripe>> {
820 let width = self.table.fields.len();
821 let workers = std::thread::available_parallelism()
822 .map_or(1, usize::from)
823 .min(MAX_ENCODE_WORKERS)
824 .min(width);
825 if workers <= 1 || held.len() <= 1 {
826 return self
827 .dictionaries
828 .iter_mut()
829 .enumerate()
830 .map(|(index, dictionary)| Self::encode_column(index, held, dictionary.as_mut()))
831 .collect();
832 }
833 let mut jobs: Vec<(usize, Option<GlobalDictionary>)> =
836 std::mem::take(&mut self.dictionaries).into_iter().enumerate().collect();
837 jobs.sort_by_key(|(index, _)| weight(&self.table.fields[*index].ty));
839 let queue = Mutex::new(jobs);
840 let pieces = std::thread::scope(|scope| {
841 (0..workers)
842 .map(|_| {
843 scope.spawn(|| {
844 let mut mine = Vec::new();
845 loop {
846 let taken = queue
847 .lock()
848 .map_err(|_| Error::internal("a native encode worker panicked"))?
849 .pop();
850 let Some((index, mut dictionary)) = taken else { break };
851 let encoded = Self::encode_column(index, held, dictionary.as_mut())?;
852 mine.push((index, dictionary, encoded));
853 }
854 Ok(mine)
855 })
856 })
857 .collect::<Vec<_>>()
858 .into_iter()
859 .map(|handle| {
860 handle.join().map_err(|_| Error::internal("a native encode worker panicked"))?
861 })
862 .collect::<Result<Vec<_>>>()
863 })?;
864 let mut dictionaries: Vec<Option<GlobalDictionary>> = (0..width).map(|_| None).collect();
865 let mut encoded: Vec<Option<ColumnStripe>> = (0..width).map(|_| None).collect();
866 for piece in pieces {
867 for (index, dictionary, stripe) in piece {
868 dictionaries[index] = dictionary;
869 encoded[index] = Some(stripe);
870 }
871 }
872 self.dictionaries = dictionaries;
873 encoded
874 .into_iter()
875 .map(|stripe| stripe.ok_or_else(|| Error::internal("a column was never encoded")))
876 .collect()
877 }
878
879 fn flush_pending(&mut self) -> Result<()> {
881 if self.pending.is_empty() {
882 return Ok(());
883 }
884 let width = self.table.fields.len();
885 let mut held = std::mem::take(&mut self.pending);
888 let parts = held.len();
889 let encoded = self.encode_columns(&held)?;
890 let mut pages = Vec::with_capacity(width);
891 let mut memberships = vec![None; width];
892 let mut ranges = Vec::with_capacity(width);
893 let mut index = Vec::with_capacity(width.saturating_mul(index_section(parts)?));
894 for stripe in &encoded {
895 let offset = self.at;
896 let section = index.len();
897 let mut length = 0_usize;
898 for bytes in &stripe.pages {
899 write_at(&self.file, self.at + length as u64, bytes)?;
900 put_u32(
901 &mut index,
902 u32::try_from(bytes.len()).map_err(|_| invalid("part length overflow"))?,
903 );
904 put_u64(&mut index, checksum(bytes));
905 length = length
906 .checked_add(bytes.len())
907 .ok_or_else(|| invalid("column page length overflow"))?;
908 }
909 let hash = checksum(&index[section..]);
910 put_u64(&mut index, hash);
911 if length > MAX_PAGE {
912 return Err(invalid("column page exceeds the configured bound"));
913 }
914 self.at = self
915 .at
916 .checked_add(length as u64)
917 .ok_or_else(|| invalid("native file length overflow"))?;
918 pages.push(Span {
919 offset,
920 length: u32::try_from(length).map_err(|_| invalid("page length overflow"))?,
921 });
922 ranges.push(merged_range(stripe.ranges.iter().cloned()));
923 }
924 for (membership, stripe) in memberships.iter_mut().zip(&encoded) {
925 if stripe.codes.iter().all(Option::is_none) {
926 continue;
927 }
928 let lists = stripe
929 .codes
930 .iter()
931 .map(|codes| codes.clone().unwrap_or_default())
932 .collect::<Vec<_>>();
933 let bytes = encode_membership(&merged_codes(lists));
934 let offset = self.at;
935 self.put(&bytes)?;
936 *membership = Some(Page {
937 offset,
938 length: u32::try_from(bytes.len())
939 .map_err(|_| invalid("membership page length overflow"))?,
940 hash: checksum(&bytes),
941 });
942 }
943 let mut sieves = vec![None; width];
944 for (page, stripe) in sieves.iter_mut().zip(&encoded) {
945 if stripe.sieves.iter().all(Option::is_none) {
946 continue;
947 }
948 let bytes = encode_sieves(stripe.sieves.iter())?;
949 let offset = self.at;
950 self.put(&bytes)?;
951 *page = Some(Page {
952 offset,
953 length: u32::try_from(bytes.len())
954 .map_err(|_| invalid("sieve page length overflow"))?,
955 hash: checksum(&bytes),
956 });
957 }
958 let offset = self.at;
959 self.put(&index)?;
960 let index = Span {
961 offset,
962 length: u32::try_from(index.len())
963 .map_err(|_| invalid("index page length overflow"))?,
964 };
965 let mut rows = 0_usize;
966 let mut lengths = Vec::with_capacity(parts);
967 let mut span = None;
968 for pending in held.drain(..) {
969 let part = pending.chunk.len();
970 rows = rows.checked_add(part).ok_or_else(|| invalid("row count overflow"))?;
971 lengths.push(u32::try_from(part).map_err(|_| invalid("part row count overflow"))?);
972 span = Some(
973 span.map_or((pending.order, pending.order), |(first, _)| (first, pending.order)),
974 );
975 }
976 self.order.push(span.ok_or_else(|| invalid("a stripe was flushed with no parts"))?);
977 self.table.stripes.push(Stripe {
978 rows,
979 parts: lengths,
980 index,
981 pages,
982 memberships,
983 sieves,
984 zone: Zone::from_ranges(ranges),
985 });
986 self.pending = held;
988 Ok(())
989 }
990
991 fn numeric_frequency(&self, column: usize) -> Result<Option<FrequencySummary>> {
995 let ty = &self.table.fields[column].ty;
996 if !matches!(
997 ty,
998 LogicalType::TinyInt
999 | LogicalType::SmallInt
1000 | LogicalType::Integer
1001 | LogicalType::BigInt
1002 | LogicalType::UTinyInt
1003 | LogicalType::USmallInt
1004 | LogicalType::UInteger
1005 | LogicalType::UBigInt
1006 | LogicalType::Date
1007 | LogicalType::Timestamp
1008 ) {
1009 return Ok(None);
1010 }
1011 let mut candidates: HashMap<FrequencyValue, u32> = HashMap::new();
1012 let mut decrements = 0_u64;
1013 self.visit_numeric(column, |_, value| {
1014 if let Some(count) = candidates.get_mut(&value) {
1015 *count = count.saturating_add(1);
1016 } else if candidates.len() < FREQUENCY_CANDIDATES {
1017 candidates.insert(value, 1);
1018 } else {
1019 candidates.retain(|_, count| {
1020 *count -= 1;
1021 *count != 0
1022 });
1023 decrements = decrements.saturating_add(1);
1024 }
1025 })?;
1026 let (exact, ordinals) = if decrements == 0 {
1027 (
1028 candidates
1029 .into_iter()
1030 .map(|(value, count)| (value, u64::from(count)))
1031 .collect::<HashMap<_, _>>(),
1032 Vec::new(),
1033 )
1034 } else {
1035 let mut lower = candidates.values().copied().collect::<Vec<_>>();
1036 lower.sort_unstable_by(|left, right| right.cmp(left));
1037 if lower.len() < FREQUENCY_BUILD_RANK
1038 || u64::from(lower[FREQUENCY_BUILD_RANK - 1]) <= decrements
1039 {
1040 return Ok(None);
1041 }
1042 let mut exact =
1043 candidates.into_keys().map(|value| (value, 0_u64)).collect::<HashMap<_, _>>();
1044 let mut ordinals = Vec::new();
1045 let mut exceeded = false;
1046 self.visit_numeric(column, |ordinal, value| {
1047 if let Some(count) = exact.get_mut(&value) {
1048 *count = count.saturating_add(1);
1049 if !exceeded {
1050 if ordinals.len() < FREQUENCY_ORDINALS {
1051 ordinals.push(ordinal);
1052 } else {
1053 ordinals.clear();
1054 exceeded = true;
1055 }
1056 }
1057 }
1058 })?;
1059 (exact, ordinals)
1060 };
1061 let mut entries = exact
1062 .into_iter()
1063 .map(|(value, count)| FrequencyEntry { value, count })
1064 .collect::<Vec<_>>();
1065 entries.sort_unstable_by(|left, right| {
1066 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
1067 });
1068 let omitted_max =
1069 entries.get(FREQUENCY_ENTRIES).map_or(decrements, |entry| decrements.max(entry.count));
1070 entries.truncate(FREQUENCY_ENTRIES);
1071 Ok(Some(FrequencySummary { entries, omitted_max, ordinals }))
1072 }
1073
1074 fn visit_numeric(
1075 &self,
1076 column: usize,
1077 mut visit: impl FnMut(u64, FrequencyValue),
1078 ) -> Result<()> {
1079 let ty = &self.table.fields[column].ty;
1080 let mut start = 0_u64;
1081 for stripe in &self.table.stripes {
1082 let spans = read_index(&self.file, stripe, column)?;
1083 let page = stripe.pages[column];
1084 let mut bytes = vec![0; page.length as usize];
1085 read_at(&self.file, page.offset, &mut bytes)?;
1086 for (span, &rows) in spans.iter().zip(&stripe.parts) {
1087 let part = part_bytes(&bytes, *span)?;
1088 if checksum(part) != span.hash {
1089 return Err(invalid("column page checksum differs while building frequencies"));
1090 }
1091 let rows = rows as usize;
1092 let vector = decode(ty, rows, part, None)?;
1093 for row in 0..rows {
1095 let value = if vector.is_null_at(row) {
1096 FrequencyValue::Null
1097 } else {
1098 let widened = match vector.signed_at(row) {
1102 Some(value) => Some(value),
1103 None => match vector.value_at(row) {
1104 Value::UTinyInt(value) => Some(i128::from(value)),
1105 Value::USmallInt(value) => Some(i128::from(value)),
1106 Value::UInteger(value) => Some(i128::from(value)),
1107 Value::UBigInt(value) => Some(i128::from(value)),
1108 _ => None,
1109 },
1110 };
1111 FrequencyValue::Integer(widened.ok_or_else(|| {
1112 invalid("numeric frequency page did not contain an integer value")
1113 })?)
1114 };
1115 visit(start.saturating_add(row as u64), value);
1116 }
1117 start = start.saturating_add(rows as u64);
1118 }
1119 }
1120 Ok(())
1121 }
1122
1123 fn numeric_frequencies(&self) -> Result<Vec<Option<FrequencySummary>>> {
1131 let mut columns = self
1132 .table
1133 .fields
1134 .iter()
1135 .enumerate()
1136 .filter_map(|(column, field)| {
1137 matches!(
1138 field.ty,
1139 LogicalType::TinyInt
1140 | LogicalType::SmallInt
1141 | LogicalType::Integer
1142 | LogicalType::BigInt
1143 | LogicalType::UTinyInt
1144 | LogicalType::USmallInt
1145 | LogicalType::UInteger
1146 | LogicalType::UBigInt
1147 | LogicalType::Date
1148 | LogicalType::Timestamp
1149 )
1150 .then_some(column)
1151 })
1152 .collect::<Vec<_>>();
1153 let workers = std::thread::available_parallelism()
1154 .map_or(1, usize::from)
1155 .min(MAX_FREQUENCY_WORKERS)
1156 .min(columns.len());
1157 if workers <= 1 {
1158 let mut frequencies = vec![None; self.table.fields.len()];
1159 for column in columns {
1160 frequencies[column] = self.numeric_frequency(column)?;
1161 }
1162 return Ok(frequencies);
1163 }
1164 columns.sort_by_key(|&column| weight(&self.table.fields[column].ty));
1167 let queue = Mutex::new(columns);
1168 let pieces = std::thread::scope(|scope| {
1169 (0..workers)
1170 .map(|_| {
1171 scope.spawn(|| {
1172 let mut mine = Vec::new();
1173 loop {
1174 let taken = queue
1175 .lock()
1176 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1177 .pop();
1178 let Some(column) = taken else { break };
1179 mine.push((column, self.numeric_frequency(column)?));
1180 }
1181 Ok(mine)
1182 })
1183 })
1184 .collect::<Vec<_>>()
1185 .into_iter()
1186 .map(|handle| {
1187 handle
1188 .join()
1189 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1190 })
1191 .collect::<Result<Vec<_>>>()
1192 })?;
1193 let mut frequencies = vec![None; self.table.fields.len()];
1194 for piece in pieces {
1195 for (column, summary) in piece {
1196 frequencies[column] = summary;
1197 }
1198 }
1199 Ok(frequencies)
1200 }
1201
1202 pub fn finish(mut self) -> Result<Table> {
1208 self.flush_pending()?;
1209 let mut stripes = std::mem::take(&mut self.order)
1210 .into_iter()
1211 .zip(std::mem::take(&mut self.table.stripes))
1212 .collect::<Vec<_>>();
1213 stripes.sort_by_key(|(order, _)| order.0);
1214 let mut previous: Option<(u64, u64)> = None;
1215 for ((first, last), _) in &stripes {
1216 if previous.is_some_and(|previous| previous >= *first) {
1217 return Err(invalid("chunks did not arrive in source order"));
1218 }
1219 previous = Some(*last);
1220 }
1221 self.table.stripes = stripes.into_iter().map(|(_, stripe)| stripe).collect();
1222 self.table.frequencies = self.numeric_frequencies()?;
1223 let dictionaries = std::mem::take(&mut self.dictionaries);
1224 let orders = rankings(&dictionaries)?;
1225 for (index, (dictionary, order)) in dictionaries.into_iter().zip(orders).enumerate() {
1226 let Some(dictionary) = dictionary else { continue };
1227 self.table.distincts[index] =
1231 Some(dictionary.counts.iter().filter(|count| **count != 0).count() as u64);
1232 self.table.frequencies[index] = Some(code_frequency(&dictionary));
1233 let encoded = encode_global_dictionary(dictionary, &order)?;
1234 let offset = self.at;
1235 self.put(&encoded.index)?;
1236 self.put(&encoded.ranks)?;
1237 for block in &encoded.payload {
1238 self.put(block)?;
1239 }
1240 let payload_len =
1241 encoded.payload.iter().try_fold(0_usize, |len, block| len.checked_add(block.len()));
1242 let length = payload_len
1243 .and_then(|len| len.checked_add(encoded.index.len()))
1244 .and_then(|len| len.checked_add(encoded.ranks.len()))
1245 .ok_or_else(|| invalid("dictionary page length overflow"))?;
1246 self.table.dictionaries[index] = Some(Page {
1247 offset,
1248 length: u32::try_from(length)
1249 .map_err(|_| invalid("dictionary page length overflow"))?,
1250 hash: checksum(&encoded.index),
1251 });
1252 }
1253 let directory = encode_directory(&self.table)?;
1254 if directory.len() > MAX_DIRECTORY {
1255 return Err(invalid("directory exceeds the configured bound"));
1256 }
1257 let offset = self.at;
1258 self.put(&directory)?;
1259 self.file.sync_all().map_err(io)?;
1260 let slot = Slot {
1261 offset,
1262 length: u32::try_from(directory.len())
1263 .map_err(|_| invalid("directory length overflow"))?,
1264 generation: self.generation,
1265 hash: checksum(&directory),
1266 };
1267 write_at(&self.file, 16, &slot.bytes())?;
1270 self.file.sync_all().map_err(io)?;
1271 Ok(self.table)
1272 }
1273}
1274
1275#[derive(Debug, Clone)]
1277pub struct Reader {
1278 file: Arc<File>,
1279 table: Arc<Table>,
1280 dictionaries: Arc<Vec<OnceLock<Arc<Vector>>>>,
1281 loading: Arc<Vec<Mutex<()>>>,
1290 opened: Arc<AtomicUsize>,
1294 sieves: Arc<Vec<Vec<SieveSlot>>>,
1298 places: Arc<Vec<Place>>,
1300 cache: Arc<Vec<Mutex<Cached>>>,
1301 pages: Arc<AtomicUsize>,
1304 indexes: Arc<AtomicUsize>,
1307 kept: Arc<AtomicUsize>,
1310 size: u64,
1312 directory: u64,
1314 opening: Opening,
1316}
1317
1318#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1330pub struct Opening {
1331 pub reads: u32,
1334 pub bytes: u64,
1336}
1337
1338#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1340pub struct Reads {
1341 pub opening: Opening,
1343 pub pages: usize,
1345 pub indexes: usize,
1347 pub dictionaries: usize,
1350}
1351
1352#[derive(Debug, Clone, Copy)]
1354struct Place {
1355 stripe: u32,
1356 part: u32,
1357 rows: u32,
1358}
1359
1360#[derive(Debug, Clone, Copy)]
1362struct PartSpan {
1363 start: usize,
1364 length: usize,
1365 hash: u64,
1366}
1367
1368#[derive(Debug, Clone)]
1374struct CachedColumn {
1375 stripe: usize,
1376 index: Arc<Vec<PartSpan>>,
1377 page: Option<Arc<Vec<u8>>>,
1378}
1379
1380#[derive(Debug, Default)]
1400struct Cached {
1401 pages: Vec<Option<Arc<Vec<u8>>>>,
1402 order: VecDeque<usize>,
1403 loading: Vec<usize>,
1404 index: Vec<Option<Arc<Vec<PartSpan>>>>,
1405}
1406
1407const CACHED_STRIPES_PER_COLUMN: usize = 4;
1419
1420type SieveSlot = OnceLock<Arc<Vec<Option<Sieve>>>>;
1422
1423#[derive(Debug)]
1424struct NativeText {
1425 file: Arc<File>,
1426 values: usize,
1428 offsets: Vec<u8>,
1437 offset_bits: usize,
1440 ranks: usize,
1442 rank_at: u64,
1446 rank_ends: Vec<u64>,
1450 rank_hashes: Vec<u64>,
1451 rank_blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1452 code_bits: usize,
1455 code_ranks: OnceLock<Option<Vec<u32>>>,
1462 payload: u64,
1463 ends: Vec<u64>,
1466 hashes: Vec<u64>,
1467 blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1469}
1470
1471const TEXT_PAYLOAD_VALUES: usize = 1024;
1487
1488const TEXT_OFFSET_RUN: usize = 512;
1495
1496const DICTIONARY_HEADER: usize = 16;
1499
1500const TEXT_RANK_BLOCK: usize = 512;
1511
1512const RANK_BLOCK_HEADER: usize = size_of::<u64>() + 1;
1526
1527impl NativeText {
1528 fn payload_block(&self, block: usize) -> Result<Option<&[u8]>> {
1535 let Some(slot) = self.blocks.get(block) else { return Ok(None) };
1536 let bytes = slot
1537 .get_or_init(|| {
1538 let start = if block == 0 { 0 } else { self.ends[block - 1] };
1539 let end = self.ends[block];
1540 let len = end
1541 .checked_sub(start)
1542 .ok_or_else(|| invalid("global dictionary block ends before it starts"))?;
1543 let mut stored = vec![
1544 0;
1545 usize::try_from(len).map_err(|_| invalid(
1546 "global dictionary block does not fit in memory"
1547 ))?
1548 ];
1549 read_at(&self.file, self.payload + start, &mut stored)?;
1550 if checksum(&stored) != self.hashes[block] {
1551 return Err(invalid("global dictionary payload checksum differs"));
1552 }
1553 let first = block * TEXT_PAYLOAD_VALUES;
1554 let last = (first + TEXT_PAYLOAD_VALUES).min(self.values);
1555 let want = self.end_within(last - 1)? as usize;
1556 let values = string::decode_flat(&stored)?;
1557 if values.len() != last - first {
1558 return Err(invalid("global dictionary block holds the wrong value count"));
1559 }
1560 let bytes = values.into_bytes();
1561 if bytes.len() != want {
1562 return Err(invalid("global dictionary block decodes to the wrong length"));
1563 }
1564 Ok(bytes)
1565 })
1566 .as_ref()
1567 .map_err(Clone::clone)?;
1568 Ok(Some(bytes.as_slice()))
1569 }
1570
1571 fn end_within(&self, index: usize) -> Result<u32> {
1573 let run = index / TEXT_OFFSET_RUN;
1574 let bytes = self
1575 .offsets
1576 .get(run * TEXT_OFFSET_RUN / 8 * self.offset_bits..)
1577 .ok_or_else(|| invalid("global dictionary offsets are short"))?;
1578 let end = bitpack::tail_at(bytes, self.offset_bits, index % TEXT_OFFSET_RUN)
1579 .map_err(|_| invalid("global dictionary offsets are short"))?;
1580 u32::try_from(end).map_err(|_| invalid("global dictionary offset is past the payload"))
1581 }
1582
1583 fn start_within(&self, index: usize) -> Result<u32> {
1586 if index % TEXT_PAYLOAD_VALUES == 0 { Ok(0) } else { self.end_within(index - 1) }
1587 }
1588
1589 fn span_within(&self, index: usize) -> Result<(u32, u32)> {
1591 let end = self.end_within(index)?;
1592 let start = self.start_within(index)?;
1593 if start > end {
1594 return Err(invalid("global dictionary value ends before it starts"));
1595 }
1596 Ok((start, end))
1597 }
1598
1599 fn rank_parts(&self, rank: usize) -> Result<(&[u8], usize)> {
1606 let slot = self
1607 .rank_blocks
1608 .get(rank / TEXT_RANK_BLOCK)
1609 .ok_or_else(|| invalid("global dictionary rank is past the order"))?;
1610 let block = slot
1611 .get_or_init(|| {
1612 let which = rank / TEXT_RANK_BLOCK;
1613 let start = if which == 0 { 0 } else { self.rank_ends[which - 1] };
1614 let end = self.rank_ends[which];
1615 let mut bytes = vec![0; (end - start) as usize];
1616 read_at(&self.file, self.rank_at + start, &mut bytes)?;
1617 if checksum(&bytes)
1618 != *self
1619 .rank_hashes
1620 .get(rank / TEXT_RANK_BLOCK)
1621 .ok_or_else(|| invalid("global dictionary rank block has no checksum"))?
1622 {
1623 return Err(invalid("global dictionary rank checksum differs"));
1624 }
1625 Ok(bytes)
1626 })
1627 .as_ref()
1628 .map_err(Clone::clone)?;
1629 Ok((block.as_slice(), rank % TEXT_RANK_BLOCK))
1630 }
1631
1632 fn head_at(&self, rank: usize) -> Result<u64> {
1634 let (block, within) = self.rank_parts(rank)?;
1635 let (base, width, packed) = rank_heads(block)?;
1636 let above = bitpack::tail_at(packed, width, within)
1637 .map_err(|_| invalid("global dictionary rank block is short of heads"))?;
1638 Ok(base.wrapping_add(above))
1639 }
1640
1641 fn rank_codes<'block>(&self, block: &'block [u8], count: usize) -> Result<&'block [u8]> {
1643 let (_, width, packed) = rank_heads(block)?;
1644 packed
1645 .get(bitpack::tail_len(count, width)..)
1646 .ok_or_else(|| invalid("global dictionary rank block is short of codes"))
1647 }
1648
1649 fn rank_block_len(&self, rank: usize) -> usize {
1651 let first = rank / TEXT_RANK_BLOCK * TEXT_RANK_BLOCK;
1652 TEXT_RANK_BLOCK.min(self.ranks - first)
1653 }
1654}
1655
1656fn rank_heads(block: &[u8]) -> Result<(u64, usize, &[u8])> {
1658 let header = block
1659 .get(..RANK_BLOCK_HEADER)
1660 .ok_or_else(|| invalid("global dictionary rank block is short"))?;
1661 let base = u64::from_le_bytes(header[..8].try_into().expect("eight bytes"));
1662 let width = header[8] as usize;
1663 if width > 64 {
1664 return Err(invalid("global dictionary rank block packs heads past a word"));
1665 }
1666 Ok((base, width, &block[RANK_BLOCK_HEADER..]))
1667}
1668
1669fn offset_width(offsets: &[u32]) -> usize {
1676 let values = offsets.len() - 1;
1677 let mut span = 0;
1678 for first in (0..values).step_by(TEXT_PAYLOAD_VALUES) {
1679 let last = (first + TEXT_PAYLOAD_VALUES).min(values);
1680 span = span.max(offsets[last] - offsets[first]);
1681 }
1682 (u32::BITS - span.leading_zeros()) as usize
1683}
1684
1685fn offset_bytes(values: usize, bits: usize) -> usize {
1688 let full = values / TEXT_OFFSET_RUN;
1689 let rest = values % TEXT_OFFSET_RUN;
1690 full * TEXT_OFFSET_RUN / 8 * bits + bitpack::tail_len(rest, bits)
1691}
1692
1693fn encode_offsets(offsets: &[u32], bits: usize, out: &mut Vec<u8>) -> Result<()> {
1695 let values = offsets.len() - 1;
1696 let mut run = Vec::with_capacity(TEXT_OFFSET_RUN);
1697 for first in (0..values).step_by(TEXT_OFFSET_RUN) {
1698 let last = (first + TEXT_OFFSET_RUN).min(values);
1699 let base = offsets[first / TEXT_PAYLOAD_VALUES * TEXT_PAYLOAD_VALUES];
1700 run.clear();
1701 run.extend((first..last).map(|value| u64::from(offsets[value + 1] - base)));
1702 bitpack::pack_tail(&run, bits, out)
1703 .map_err(|_| invalid("global dictionary offsets do not pack"))?;
1704 }
1705 Ok(())
1706}
1707
1708fn code_width(values: usize) -> usize {
1710 match u64::try_from(values).unwrap_or(u64::MAX) {
1711 0 | 1 => 0,
1712 last => (u64::BITS - (last - 1).leading_zeros()) as usize,
1713 }
1714}
1715
1716impl TextSource for NativeText {
1717 fn len(&self) -> usize {
1718 self.values
1719 }
1720
1721 fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
1722 if index >= self.values {
1723 return Ok(None);
1724 }
1725 let (start, end) = self.span_within(index)?;
1726 if start == end {
1727 return Ok(Some(&[]));
1728 }
1729 let block = index / TEXT_PAYLOAD_VALUES;
1732 let Some(bytes) = self.payload_block(block)? else { return Ok(None) };
1733 Ok(bytes.get(start as usize..end as usize))
1734 }
1735
1736 fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
1737 if index >= self.values {
1738 return Ok(None);
1739 }
1740 let (start, end) = self.span_within(index)?;
1741 Ok(Some((end - start) as usize))
1742 }
1743
1744 fn ranks(&self) -> Option<usize> {
1745 (self.ranks > 0).then_some(self.ranks)
1746 }
1747
1748 fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
1749 let settled = self.head_at(rank)?.cmp(&head(wanted));
1753 if settled != Ordering::Equal {
1754 return Ok(settled);
1755 }
1756 let code = self.code_at_rank(rank)?;
1757 let bytes = self
1758 .bytes_at(code as usize)?
1759 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
1760 Ok(bytes.cmp(wanted))
1761 }
1762
1763 fn code_at_rank(&self, rank: usize) -> Result<u32> {
1764 let (block, within) = self.rank_parts(rank)?;
1765 let codes = self.rank_codes(block, self.rank_block_len(rank))?;
1766 let code = bitpack::tail_at(codes, self.code_bits, within)
1767 .map_err(|_| invalid("global dictionary rank block is short of codes"))?;
1768 let code = u32::try_from(code)
1769 .map_err(|_| invalid("global dictionary order names a code it does not have"))?;
1770 if code as usize >= self.len() {
1771 return Err(invalid("global dictionary order names a code it does not have"));
1772 }
1773 Ok(code)
1774 }
1775
1776 fn code_ranks(&self) -> Option<&[u32]> {
1777 if self.ranks == 0 || self.ranks != self.len() {
1781 return None;
1782 }
1783 self.code_ranks
1784 .get_or_init(|| {
1785 let mut ranks = vec![u32::MAX; self.ranks];
1786 for first in (0..self.ranks).step_by(TEXT_RANK_BLOCK) {
1789 let (block, _) = self.rank_parts(first).ok()?;
1790 let count = self.rank_block_len(first);
1791 let codes = self.rank_codes(block, count).ok()?;
1792 for (within, code) in bitpack::unpack_tail(codes, self.code_bits, count)
1793 .ok()?
1794 .into_iter()
1795 .enumerate()
1796 {
1797 let code = usize::try_from(code).ok()?;
1798 *ranks.get_mut(code)? = u32::try_from(first + within).ok()?;
1799 }
1800 }
1801 if ranks.contains(&u32::MAX) {
1802 return None;
1803 }
1804 Some(ranks)
1805 })
1806 .as_deref()
1807 }
1808
1809 fn footprint(&self) -> usize {
1810 self.offsets.capacity()
1811 + self
1812 .code_ranks
1813 .get()
1814 .and_then(Option::as_ref)
1815 .map_or(0, |ranks| ranks.capacity() * size_of::<u32>())
1816 + self.rank_hashes.capacity() * size_of::<u64>()
1817 + self.rank_ends.capacity() * size_of::<u64>()
1818 + self.rank_blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1819 + self
1820 .rank_blocks
1821 .iter()
1822 .filter_map(OnceLock::get)
1823 .filter_map(|result| result.as_ref().ok())
1824 .map(Vec::capacity)
1825 .sum::<usize>()
1826 + self.blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1827 + self.hashes.capacity() * size_of::<u64>()
1828 + self.ends.capacity() * size_of::<u64>()
1829 + self
1830 .blocks
1831 .iter()
1832 .filter_map(OnceLock::get)
1833 .filter_map(|result| result.as_ref().ok())
1834 .map(Vec::capacity)
1835 .sum::<usize>()
1836 }
1837}
1838
1839fn places(table: &Table) -> Result<Vec<Place>> {
1841 let mut places = Vec::with_capacity(table.stripes.len().saturating_mul(STRIPE_PARTS));
1842 for (at, stripe) in table.stripes.iter().enumerate() {
1843 let index = u32::try_from(at).map_err(|_| invalid("too many stripes"))?;
1844 for (part, &rows) in stripe.parts.iter().enumerate() {
1845 places.push(Place {
1846 stripe: index,
1847 part: u32::try_from(part).map_err(|_| invalid("too many parts in a stripe"))?,
1848 rows,
1849 });
1850 }
1851 }
1852 Ok(places)
1853}
1854
1855fn read_index(file: &File, stripe: &Stripe, column: usize) -> Result<Vec<PartSpan>> {
1860 let parts = stripe.parts.len();
1861 let section = index_section(parts)?;
1862 let at = column.checked_mul(section).ok_or_else(|| invalid("index page offset overflow"))?;
1863 let end = at.checked_add(section).ok_or_else(|| invalid("index page offset overflow"))?;
1864 if end > stripe.index.length as usize {
1865 return Err(invalid("index page is shorter than its columns"));
1866 }
1867 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1868 let mut bytes = vec![0; section];
1869 let offset = stripe
1870 .index
1871 .offset
1872 .checked_add(at as u64)
1873 .ok_or_else(|| invalid("index page offset overflow"))?;
1874 read_at(file, offset, &mut bytes)?;
1875 let entries = section - size_of::<u64>();
1876 let stored = u64::from_le_bytes(bytes[entries..].try_into().expect("eight bytes"));
1877 if checksum(&bytes[..entries]) != stored {
1878 return Err(invalid(&format!(
1881 "index page section checksum differs, column {column} of {parts} parts at {offset}, \
1882 wanted {stored:016x} and got {:016x}",
1883 checksum(&bytes[..entries]),
1884 )));
1885 }
1886 let mut spans = Vec::with_capacity(parts);
1887 let mut start = 0_usize;
1888 for part in 0..parts {
1889 let at = part * INDEX_ENTRY;
1890 let length = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four bytes")) as usize;
1891 let hash = u64::from_le_bytes(bytes[at + 4..at + 12].try_into().expect("eight bytes"));
1892 spans.push(PartSpan { start, length, hash });
1893 start = start.checked_add(length).ok_or_else(|| invalid("column page length overflow"))?;
1894 }
1895 if start != page.length as usize {
1896 return Err(invalid("column page length differs from its index"));
1897 }
1898 Ok(spans)
1899}
1900
1901fn part_bytes(page: &[u8], span: PartSpan) -> Result<&[u8]> {
1903 let end = span.start.checked_add(span.length).ok_or_else(|| invalid("part range overflow"))?;
1904 page.get(span.start..end).ok_or_else(|| invalid("part exceeds its column page"))
1905}
1906
1907fn remember(cached: &mut Cached, held: &CachedColumn, kept: usize) {
1912 if let Some(slot) = cached.index.get_mut(held.stripe) {
1913 if slot.is_none() {
1914 *slot = Some(Arc::clone(&held.index));
1915 }
1916 }
1917 let Some(page) = held.page.clone() else { return };
1918 let Some(slot) = cached.pages.get_mut(held.stripe) else { return };
1919 if slot.is_none() {
1920 cached.order.push_back(held.stripe);
1921 }
1922 *slot = Some(page);
1923 while cached.order.len() > kept.max(1) {
1924 let Some(oldest) = cached.order.pop_front() else { break };
1925 if let Some(slot) = cached.pages.get_mut(oldest) {
1926 *slot = None;
1927 }
1928 }
1929}
1930
1931impl Reader {
1932 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1938 let mut file = File::open(path).map_err(io)?;
1939 let size = file.metadata().map_err(io)?.len();
1940 if size < HEADER {
1941 return Err(invalid("file is shorter than its header"));
1942 }
1943 let mut header = [0; HEADER as usize];
1944 file.read_exact(&mut header).map_err(io)?;
1945 let mut opening = Opening { reads: 1, bytes: HEADER };
1946 let version = u32::from_le_bytes([header[8], header[9], header[10], header[11]]);
1947 if &header[..8] != MAGIC {
1952 return Err(invalid("the header does not begin with a rudb native magic"));
1953 }
1954 if version != FORMAT {
1955 return Err(invalid(&format!(
1956 "the file is format {version} and this build reads format {FORMAT}, so it has to \
1957 be written again"
1958 )));
1959 }
1960 let mut selected = None;
1961 for start in [16, 16 + SLOT_BYTES] {
1962 let slot = Slot::read(&header[start..start + SLOT_BYTES]);
1963 if slot.generation == 0 || slot.length == 0 || slot.length as usize > MAX_DIRECTORY {
1964 continue;
1965 }
1966 let Some(end) = slot.offset.checked_add(u64::from(slot.length)) else { continue };
1967 if slot.offset < HEADER || end > size {
1968 continue;
1969 }
1970 let mut bytes = vec![0; slot.length as usize];
1971 file.seek(SeekFrom::Start(slot.offset)).map_err(io)?;
1972 file.read_exact(&mut bytes).map_err(io)?;
1973 opening.reads += 1;
1974 opening.bytes += u64::from(slot.length);
1975 if checksum(&bytes) == slot.hash
1976 && selected
1977 .as_ref()
1978 .is_none_or(|(old, _): &(Slot, Vec<u8>)| old.generation < slot.generation)
1979 {
1980 selected = Some((slot, bytes));
1981 }
1982 }
1983 let (slot, bytes) =
1984 selected.ok_or_else(|| invalid("no committed directory slot is valid"))?;
1985 let table = decode_directory(&bytes, size)?;
1986 let places = places(&table)?;
1987 let dictionaries = (0..table.fields.len()).map(|_| OnceLock::new()).collect();
1988 let table_fields = table.fields.len();
1989 let stripes = table.stripes.len();
1990 let cache = (0..table.fields.len())
1991 .map(|_| {
1992 Mutex::new(Cached {
1993 pages: (0..stripes).map(|_| None).collect(),
1994 index: (0..stripes).map(|_| None).collect(),
1995 ..Cached::default()
1996 })
1997 })
1998 .collect::<Vec<_>>();
1999 let sieves = (0..table.fields.len())
2000 .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
2001 .collect();
2002 Ok(Self {
2003 file: Arc::new(file),
2004 table: Arc::new(table),
2005 dictionaries: Arc::new(dictionaries),
2006 loading: Arc::new((0..table_fields).map(|_| Mutex::new(())).collect()),
2007 opened: Arc::new(AtomicUsize::new(0)),
2008 sieves: Arc::new(sieves),
2009 places: Arc::new(places),
2010 cache: Arc::new(cache),
2011 pages: Arc::new(AtomicUsize::new(0)),
2012 indexes: Arc::new(AtomicUsize::new(0)),
2013 kept: Arc::new(AtomicUsize::new(CACHED_STRIPES_PER_COLUMN)),
2014 size,
2015 directory: u64::from(slot.length),
2016 opening,
2017 })
2018 }
2019
2020 #[must_use]
2027 pub fn reads(&self) -> Reads {
2028 Reads {
2029 opening: self.opening,
2030 pages: self.pages.load(Atomic::Relaxed),
2031 indexes: self.indexes.load(Atomic::Relaxed),
2032 dictionaries: self.opened.load(Atomic::Relaxed),
2033 }
2034 }
2035
2036 #[must_use]
2041 pub fn layout(&self) -> Layout {
2042 let table = &self.table;
2043 let stripes = table.stripes.as_slice();
2044 let columns = table
2045 .fields
2046 .iter()
2047 .enumerate()
2048 .map(|(at, field)| ColumnLayout {
2049 name: field.name.clone(),
2050 kind: field.ty.to_string(),
2051 pages: sum(stripes.iter().map(|stripe| span_bytes(&stripe.pages, at))),
2052 memberships: sum(stripes.iter().map(|stripe| page_bytes(&stripe.memberships, at))),
2053 sieves: sum(stripes.iter().map(|stripe| page_bytes(&stripe.sieves, at))),
2054 dictionary: page_bytes(&table.dictionaries, at),
2055 })
2056 .collect();
2057 Layout {
2058 file: self.size,
2059 rows: table.rows,
2060 stripes: stripes.len(),
2061 parts: self.places.len(),
2062 columns,
2063 indexes: sum(stripes.iter().map(|stripe| u64::from(stripe.index.length))),
2064 directory: self.directory,
2065 header: HEADER,
2066 }
2067 }
2068
2069 #[must_use]
2071 pub fn parts(&self) -> usize {
2072 self.places.len()
2073 }
2074
2075 #[must_use]
2082 pub fn stripe_parts(&self) -> Vec<std::ops::Range<usize>> {
2083 let mut runs = Vec::with_capacity(self.table.stripes.len());
2084 let mut start = 0;
2085 for stripe in &self.table.stripes {
2086 let end = start + stripe.parts.len();
2087 runs.push(start..end);
2088 start = end;
2089 }
2090 runs
2091 }
2092
2093 pub fn keep_stripes(&self, stripes: usize) {
2100 self.kept.fetch_max(stripes, Atomic::Relaxed);
2101 }
2102
2103 #[must_use]
2105 pub fn part_rows(&self, at: usize) -> usize {
2106 self.places.get(at).map_or(0, |place| place.rows as usize)
2107 }
2108
2109 #[must_use]
2111 pub fn table(&self) -> &Table {
2112 &self.table
2113 }
2114
2115 pub fn top_frequencies(&self, column: usize, top: usize) -> Result<Option<Vec<(Value, u64)>>> {
2124 let field = self
2125 .table
2126 .fields
2127 .get(column)
2128 .ok_or_else(|| invalid("frequency column index out of range"))?;
2129 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2130 return Ok(None);
2131 };
2132 if top == 0 || summary.entries.len() < top {
2133 return Ok(None);
2134 }
2135 let boundary = summary.entries[top - 1].count;
2136 if boundary <= summary.omitted_max {
2137 return Ok(None);
2138 }
2139 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
2140 }
2141
2142 pub fn exact_frequencies(&self, column: usize) -> Result<Option<Vec<(Value, u64)>>> {
2162 let field = self
2163 .table
2164 .fields
2165 .get(column)
2166 .ok_or_else(|| invalid("frequency column index out of range"))?;
2167 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2168 return Ok(None);
2169 };
2170 if summary.omitted_max > 0 {
2171 return Ok(None);
2172 }
2173 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
2174 }
2175
2176 fn decode_frequencies(
2178 &self,
2179 column: usize,
2180 ty: &LogicalType,
2181 entries: &[FrequencyEntry],
2182 ) -> Result<Vec<(Value, u64)>> {
2183 let dictionary = if *ty == LogicalType::Varchar { self.dictionary(column)? } else { None };
2184 let mut out = Vec::with_capacity(entries.len());
2185 for entry in entries {
2186 let value = match entry.value {
2187 FrequencyValue::Null => Value::Null,
2188 FrequencyValue::Integer(value) => match *ty {
2189 LogicalType::TinyInt => Value::TinyInt(
2190 i8::try_from(value)
2191 .map_err(|_| invalid("frequency TINYINT is out of range"))?,
2192 ),
2193 LogicalType::UTinyInt => Value::UTinyInt(
2194 u8::try_from(value)
2195 .map_err(|_| invalid("frequency UTINYINT is out of range"))?,
2196 ),
2197 LogicalType::USmallInt => Value::USmallInt(
2198 u16::try_from(value)
2199 .map_err(|_| invalid("frequency USMALLINT is out of range"))?,
2200 ),
2201 LogicalType::UInteger => Value::UInteger(
2202 u32::try_from(value)
2203 .map_err(|_| invalid("frequency UINTEGER is out of range"))?,
2204 ),
2205 LogicalType::UBigInt => Value::UBigInt(
2206 u64::try_from(value)
2207 .map_err(|_| invalid("frequency UBIGINT is out of range"))?,
2208 ),
2209 LogicalType::SmallInt => Value::SmallInt(
2210 i16::try_from(value)
2211 .map_err(|_| invalid("frequency SMALLINT is out of range"))?,
2212 ),
2213 LogicalType::Integer => Value::Integer(
2214 i32::try_from(value)
2215 .map_err(|_| invalid("frequency INTEGER is out of range"))?,
2216 ),
2217 LogicalType::BigInt => Value::BigInt(
2218 i64::try_from(value)
2219 .map_err(|_| invalid("frequency BIGINT is out of range"))?,
2220 ),
2221 LogicalType::Date => Value::Date(
2222 i32::try_from(value)
2223 .map_err(|_| invalid("frequency DATE is out of range"))?,
2224 ),
2225 LogicalType::Timestamp => Value::Timestamp(
2226 i64::try_from(value)
2227 .map_err(|_| invalid("frequency TIMESTAMP is out of range"))?,
2228 ),
2229 _ => return Err(invalid("integer frequency belongs to another type")),
2230 },
2231 FrequencyValue::Code(code) => dictionary
2232 .as_ref()
2233 .ok_or_else(|| invalid("frequency code has no dictionary"))?
2234 .try_value_at(code as usize)?,
2235 };
2236 out.push((value, entry.count));
2237 }
2238 Ok(out)
2239 }
2240
2241 pub fn frequency_occurrences(&self, column: usize) -> Result<Option<FrequencyOccurrences>> {
2251 self.table
2252 .fields
2253 .get(column)
2254 .ok_or_else(|| invalid("frequency column index out of range"))?;
2255 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2256 return Ok(None);
2257 };
2258 if summary.ordinals.is_empty() {
2259 return Ok(None);
2260 }
2261 Ok(Some(FrequencyOccurrences {
2262 omitted_max: summary.omitted_max,
2263 ordinals: summary.ordinals.clone(),
2264 }))
2265 }
2266
2267 pub fn distinct_values(&self, column: usize) -> Result<Option<u64>> {
2291 self.table
2292 .distincts
2293 .get(column)
2294 .copied()
2295 .ok_or_else(|| invalid("distinct column index out of range"))
2296 }
2297
2298 pub fn null_count(&self, column: usize) -> Result<u64> {
2309 if column >= self.table.fields.len() {
2310 return Err(invalid("null count column index out of range"));
2311 }
2312 let mut nulls = 0_u64;
2313 for stripe in &self.table.stripes {
2314 let range = stripe
2315 .zone
2316 .column(column)
2317 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2318 nulls = nulls
2319 .checked_add(range.nulls as u64)
2320 .ok_or_else(|| invalid("null count overflow"))?;
2321 }
2322 Ok(nulls)
2323 }
2324
2325 pub fn text_extremes(&self, column: usize) -> Result<Option<(Value, Value)>> {
2340 if self.null_count(column)? > 0 {
2341 return Ok(None);
2342 }
2343 let Some(dictionary) = self.dictionary(column)? else { return Ok(None) };
2344 let Some(ranks) = dictionary.ranks() else { return Ok(None) };
2345 if ranks == 0 {
2346 return Ok(None);
2347 }
2348 let low = text_at_rank(&dictionary, 0)?;
2349 let high = text_at_rank(&dictionary, ranks - 1)?;
2350 Ok(Some((low, high)))
2351 }
2352
2353 pub fn exact_extremes(&self, column: usize) -> Result<Option<(Bound, Bound)>> {
2376 if column >= self.table.fields.len() {
2377 return Err(invalid("extremes column index out of range"));
2378 }
2379 let mut low: Option<Bound> = None;
2380 let mut high: Option<Bound> = None;
2381 for stripe in &self.table.stripes {
2382 let range = stripe
2383 .zone
2384 .column(column)
2385 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2386 if !range.exact {
2387 return Ok(None);
2388 }
2389 let (Some(small), Some(large)) = (range.low.as_ref(), range.high.as_ref()) else {
2394 if stripe.rows > range.nulls {
2395 return Ok(None);
2396 }
2397 continue;
2398 };
2399 low = Some(low.map_or_else(|| small.clone(), |held| held.smaller(small.clone())));
2400 high = Some(high.map_or_else(|| large.clone(), |held| held.larger(large.clone())));
2401 }
2402 Ok(low.zip(high))
2403 }
2404
2405 pub fn exact_sum(&self, column: usize) -> Result<Option<(i128, u64)>> {
2418 if column >= self.table.fields.len() {
2419 return Err(invalid("sum column index out of range"));
2420 }
2421 let mut total = 0_i128;
2422 let mut rows = 0_u64;
2423 for stripe in &self.table.stripes {
2424 let range = stripe
2425 .zone
2426 .column(column)
2427 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2428 let Some(part) = range.sum else { return Ok(None) };
2429 let Some(sum) = total.checked_add(part) else { return Ok(None) };
2430 total = sum;
2431 rows = rows.saturating_add(stripe.rows as u64 - range.nulls as u64);
2432 }
2433 Ok(Some((total, rows)))
2434 }
2435
2436 fn dictionary(&self, column: usize) -> Result<Option<Arc<Vector>>> {
2445 let Some(page) = self.table.dictionaries[column] else { return Ok(None) };
2446 if let Some(dictionary) = self.dictionaries[column].get() {
2447 return Ok(Some(Arc::clone(dictionary)));
2448 }
2449 let _queued = self.loading[column].lock().map_err(|_| invalid("a poisoned dictionary"))?;
2450 if let Some(dictionary) = self.dictionaries[column].get() {
2451 return Ok(Some(Arc::clone(dictionary)));
2452 }
2453 self.opened.fetch_add(1, Atomic::Relaxed);
2454 let dictionary = Arc::new(open_global_dictionary(
2455 Arc::clone(&self.file),
2456 page,
2457 &self.table.fields[column].ty,
2458 )?);
2459 let _ = self.dictionaries[column].set(Arc::clone(&dictionary));
2460 Ok(Some(dictionary))
2461 }
2462
2463 pub fn read(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2472 self.read_impl(part, columns, true)
2473 }
2474
2475 pub fn read_sparse(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2485 self.read_impl(part, columns, false)
2486 }
2487
2488 pub fn skips_codes(&self, part: usize, column: usize, candidates: &[u32]) -> Result<bool> {
2495 if candidates.is_empty() {
2496 return Ok(true);
2497 }
2498 if candidates.windows(2).any(|pair| pair[0] >= pair[1]) {
2499 return Err(Error::internal("native code candidates are not sorted and unique"));
2500 }
2501 let stripe = self.stripe_of(part)?;
2502 let Some(page) = stripe.memberships.get(column).copied().flatten() else {
2503 return Ok(false);
2504 };
2505 let mut bytes = vec![0; page.length as usize];
2506 read_at(&self.file, page.offset, &mut bytes)?;
2507 if checksum(&bytes) != page.hash {
2508 return Err(invalid("membership page checksum differs"));
2509 }
2510 let codes = decode_membership(&bytes)?;
2511 let mut left = 0;
2512 let mut right = 0;
2513 while left < codes.len() && right < candidates.len() {
2514 match codes[left].cmp(&candidates[right]) {
2515 Ordering::Less => left += 1,
2516 Ordering::Greater => right += 1,
2517 Ordering::Equal => return Ok(false),
2518 }
2519 }
2520 Ok(true)
2521 }
2522
2523 fn stripe_of(&self, part: usize) -> Result<&Stripe> {
2524 let place = self.places.get(part).ok_or_else(|| invalid("part index out of range"))?;
2525 self.table
2526 .stripes
2527 .get(place.stripe as usize)
2528 .ok_or_else(|| invalid("stripe index out of range"))
2529 }
2530
2531 fn held(&self, at: usize, stripe: &Stripe, column: usize, whole: bool) -> Result<CachedColumn> {
2548 let cache = self.cache.get(column).ok_or_else(|| invalid("column index out of range"))?;
2549 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2550 let known = cached.index.get(at).and_then(Clone::clone);
2551 let page = cached.pages.get(at).and_then(Clone::clone);
2552 if let Some(index) = known.clone() {
2553 if !whole || page.is_some() {
2554 return Ok(CachedColumn { stripe: at, index, page });
2555 }
2556 }
2557 if cached.loading.contains(&at) {
2558 drop(cached);
2559 if let Some(index) = known {
2563 return Ok(CachedColumn { stripe: at, index, page: None });
2564 }
2565 let held = self.page_of(stripe, column, at, false, None)?;
2566 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2567 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2568 return Ok(held);
2569 }
2570 cached.loading.push(at);
2571 drop(cached);
2572
2573 let read = self.page_of(stripe, column, at, whole, known);
2574
2575 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2579 if let Some(position) = cached.loading.iter().position(|loading| *loading == at) {
2580 cached.loading.remove(position);
2581 }
2582 let held = read?;
2583 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2584 Ok(held)
2585 }
2586
2587 fn page_of(
2593 &self,
2594 stripe: &Stripe,
2595 column: usize,
2596 at: usize,
2597 whole: bool,
2598 known: Option<Arc<Vec<PartSpan>>>,
2599 ) -> Result<CachedColumn> {
2600 let index = match known {
2601 Some(index) => index,
2602 None => {
2603 self.indexes.fetch_add(1, Atomic::Relaxed);
2604 Arc::new(read_index(&self.file, stripe, column)?)
2605 }
2606 };
2607 let page = if whole {
2608 self.pages.fetch_add(1, Atomic::Relaxed);
2609 let span = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2610 let mut bytes = vec![0; span.length as usize];
2611 read_at(&self.file, span.offset, &mut bytes)?;
2612 Some(Arc::new(bytes))
2613 } else {
2614 None
2615 };
2616 Ok(CachedColumn { stripe: at, index, page })
2617 }
2618
2619 fn read_impl(&self, at: usize, columns: &[usize], whole: bool) -> Result<Chunk> {
2620 let place = *self.places.get(at).ok_or_else(|| invalid("part index out of range"))?;
2621 let index = place.stripe as usize;
2622 let stripe =
2623 self.table.stripes.get(index).ok_or_else(|| invalid("stripe index out of range"))?;
2624 let rows = place.rows as usize;
2625 let mut picked = Vec::with_capacity(columns.len());
2626 for &column in columns {
2627 let field = self
2628 .table
2629 .fields
2630 .get(column)
2631 .ok_or_else(|| invalid("column index out of range"))?;
2632 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2633 let held = self.held(index, stripe, column, whole)?;
2634 let span = *held
2635 .index
2636 .get(place.part as usize)
2637 .ok_or_else(|| invalid("part index out of range"))?;
2638 let owned;
2639 let bytes = match &held.page {
2640 Some(held) => part_bytes(held, span)?,
2641 None => {
2642 let offset = page
2643 .offset
2644 .checked_add(span.start as u64)
2645 .ok_or_else(|| invalid("part range overflow"))?;
2646 let mut bytes = vec![0; span.length];
2647 read_at(&self.file, offset, &mut bytes)?;
2648 owned = bytes;
2649 &owned
2650 }
2651 };
2652 if checksum(bytes) != span.hash {
2653 return Err(invalid(&format!(
2654 "column page checksum differs, column {column} part {} at {}+{} of {} bytes, \
2655 wanted {:016x} and got {:016x}",
2656 place.part,
2657 page.offset,
2658 span.start,
2659 span.length,
2660 span.hash,
2661 checksum(bytes),
2662 )));
2663 }
2664 let dictionary = self.dictionary(column)?;
2665 picked.push(decode(&field.ty, rows, bytes, dictionary)?);
2666 }
2667 Chunk::with_rows(picked, rows)
2668 }
2669
2670 #[must_use]
2680 pub fn skips(&self, part: usize, probes: &[Probe]) -> bool {
2681 let Some(place) = self.places.get(part).copied() else { return false };
2682 let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
2683 if stripe.zone.skips(probes) {
2684 return true;
2685 }
2686 probes.iter().any(|probe| self.sifted(place, probe))
2687 }
2688
2689 #[must_use]
2700 pub fn stripe_skips(&self, stripe: usize, probes: &[Probe]) -> bool {
2701 self.table.stripes.get(stripe).is_some_and(|held| held.zone.skips(probes))
2702 }
2703
2704 fn sifted(&self, place: Place, probe: &Probe) -> bool {
2710 if probe.op != Op::Equal {
2711 return false;
2712 }
2713 match self.stripe_sieves(place.stripe as usize, probe.column) {
2714 Some(sieves) => sieves
2715 .get(place.part as usize)
2716 .and_then(Option::as_ref)
2717 .is_some_and(|sieve| sieve.excludes(&probe.value)),
2718 None => false,
2719 }
2720 }
2721
2722 fn stripe_sieves(&self, stripe: usize, column: usize) -> Option<&[Option<Sieve>]> {
2729 let slot = self.sieves.get(column)?.get(stripe)?;
2730 if let Some(held) = slot.get() {
2731 return Some(held);
2732 }
2733 let page = self.table.stripes.get(stripe)?.sieves.get(column).copied().flatten()?;
2734 let mut bytes = vec![0; page.length as usize];
2735 read_at(&self.file, page.offset, &mut bytes).ok()?;
2736 if checksum(&bytes) != page.hash {
2737 return None;
2738 }
2739 let sieves = Arc::new(decode_sieves(&bytes).ok()?);
2740 let _ = slot.set(sieves);
2741 slot.get().map(|held| held.as_slice())
2742 }
2743}
2744
2745fn text_at_rank(dictionary: &Vector, rank: usize) -> Result<Value> {
2747 let code = dictionary.code_at_rank(rank)? as usize;
2748 let text = dictionary
2749 .try_text_at(code)?
2750 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
2751 Ok(Value::Varchar(text.into()))
2752}
2753
2754#[cfg(unix)]
2759fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2760 use std::os::unix::fs::FileExt;
2761 while !bytes.is_empty() {
2762 let written = file.write_at(bytes, offset).map_err(io)?;
2763 if written == 0 {
2764 return Err(invalid("a write to the native file wrote nothing"));
2765 }
2766 offset += written as u64;
2767 bytes = &bytes[written..];
2768 }
2769 Ok(())
2770}
2771
2772#[cfg(windows)]
2774fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2775 use std::os::windows::fs::FileExt;
2776 while !bytes.is_empty() {
2777 let written = file.seek_write(bytes, offset).map_err(io)?;
2778 if written == 0 {
2779 return Err(invalid("a write to the native file wrote nothing"));
2780 }
2781 offset += written as u64;
2782 bytes = &bytes[written..];
2783 }
2784 Ok(())
2785}
2786
2787#[cfg(not(any(unix, windows)))]
2789fn write_at(file: &File, offset: u64, bytes: &[u8]) -> Result<()> {
2790 use std::io::Write;
2791 let mut file = file.try_clone().map_err(io)?;
2792 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2793 file.write_all(bytes).map_err(io)
2794}
2795
2796#[cfg(unix)]
2806fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2807 use std::os::unix::fs::FileExt;
2808 while !bytes.is_empty() {
2809 let read = file.read_at(bytes, offset).map_err(io)?;
2810 if read == 0 {
2811 return Err(invalid("column page ends before its declared length"));
2812 }
2813 offset += read as u64;
2814 bytes = &mut bytes[read..];
2815 }
2816 Ok(())
2817}
2818
2819#[cfg(windows)]
2825fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2826 use std::os::windows::fs::FileExt;
2827 while !bytes.is_empty() {
2828 let read = file.seek_read(bytes, offset).map_err(io)?;
2829 if read == 0 {
2830 return Err(invalid("column page ends before its declared length"));
2831 }
2832 offset += read as u64;
2833 bytes = &mut bytes[read..];
2834 }
2835 Ok(())
2836}
2837
2838#[cfg(not(any(unix, windows)))]
2843fn read_at(file: &File, offset: u64, bytes: &mut [u8]) -> Result<()> {
2844 let mut file = file.try_clone().map_err(io)?;
2845 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2846 file.read_exact(bytes).map_err(io)
2847}
2848
2849fn type_tag(ty: &LogicalType) -> Result<u8> {
2850 match ty {
2851 LogicalType::SmallInt => Ok(1),
2852 LogicalType::Integer => Ok(2),
2853 LogicalType::BigInt => Ok(3),
2854 LogicalType::Varchar => Ok(4),
2855 LogicalType::Date => Ok(5),
2856 LogicalType::Timestamp => Ok(6),
2857 LogicalType::Boolean => Ok(7),
2858 LogicalType::TinyInt => Ok(8),
2859 LogicalType::UTinyInt => Ok(9),
2860 LogicalType::USmallInt => Ok(10),
2861 LogicalType::UInteger => Ok(11),
2862 LogicalType::UBigInt => Ok(12),
2863 _ => Err(Error::not_implemented(format!("native storage for {ty}"))),
2864 }
2865}
2866
2867fn tag_type(tag: u8) -> Result<LogicalType> {
2868 match tag {
2869 1 => Ok(LogicalType::SmallInt),
2870 2 => Ok(LogicalType::Integer),
2871 3 => Ok(LogicalType::BigInt),
2872 4 => Ok(LogicalType::Varchar),
2873 5 => Ok(LogicalType::Date),
2874 6 => Ok(LogicalType::Timestamp),
2875 7 => Ok(LogicalType::Boolean),
2876 8 => Ok(LogicalType::TinyInt),
2877 9 => Ok(LogicalType::UTinyInt),
2878 10 => Ok(LogicalType::USmallInt),
2879 11 => Ok(LogicalType::UInteger),
2880 12 => Ok(LogicalType::UBigInt),
2881 _ => Err(invalid("column type tag is unknown")),
2882 }
2883}
2884
2885fn put_u16(out: &mut Vec<u8>, value: u16) {
2886 out.extend_from_slice(&value.to_le_bytes());
2887}
2888fn put_u32(out: &mut Vec<u8>, value: u32) {
2889 out.extend_from_slice(&value.to_le_bytes());
2890}
2891fn put_u64(out: &mut Vec<u8>, value: u64) {
2892 out.extend_from_slice(&value.to_le_bytes());
2893}
2894fn put_var_u64(out: &mut Vec<u8>, mut value: u64) {
2895 while value >= 0x80 {
2896 out.push((value as u8 & 0x7f) | 0x80);
2897 value >>= 7;
2898 }
2899 out.push(value as u8);
2900}
2901
2902fn frequency_order(left: FrequencyValue, right: FrequencyValue) -> Ordering {
2903 match (left, right) {
2904 (FrequencyValue::Null, FrequencyValue::Null) => Ordering::Equal,
2905 (FrequencyValue::Null, _) => Ordering::Less,
2906 (_, FrequencyValue::Null) => Ordering::Greater,
2907 (FrequencyValue::Integer(left), FrequencyValue::Integer(right)) => left.cmp(&right),
2908 (FrequencyValue::Code(left), FrequencyValue::Code(right)) => left.cmp(&right),
2909 (FrequencyValue::Integer(_), FrequencyValue::Code(_)) => Ordering::Less,
2910 (FrequencyValue::Code(_), FrequencyValue::Integer(_)) => Ordering::Greater,
2911 }
2912}
2913
2914fn code_frequency(dictionary: &GlobalDictionary) -> FrequencySummary {
2915 let mut entries = dictionary
2916 .counts
2917 .iter()
2918 .enumerate()
2919 .filter(|(_, count)| **count != 0)
2920 .map(|(code, &count)| FrequencyEntry { value: FrequencyValue::Code(code as u32), count })
2921 .collect::<Vec<_>>();
2922 if dictionary.nulls != 0 {
2923 entries.push(FrequencyEntry { value: FrequencyValue::Null, count: dictionary.nulls });
2924 }
2925 entries.sort_unstable_by(|left, right| {
2926 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
2927 });
2928 let omitted_max = entries.get(FREQUENCY_ENTRIES).map_or(0, |entry| entry.count);
2929 entries.truncate(FREQUENCY_ENTRIES);
2930 FrequencySummary { entries, omitted_max, ordinals: Vec::new() }
2931}
2932
2933fn encode_directory(table: &Table) -> Result<Vec<u8>> {
2934 let mut out = DIRECTORY.to_vec();
2935 let name = table.name.as_bytes();
2936 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
2937 out.extend_from_slice(name);
2938 put_u16(&mut out, u16::try_from(table.fields.len()).map_err(|_| invalid("too many columns"))?);
2939 for field in &table.fields {
2940 let name = field.name.as_bytes();
2941 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?);
2942 out.extend_from_slice(name);
2943 out.push(type_tag(&field.ty)?);
2944 out.push(u8::from(field.not_null));
2945 }
2946 for dictionary in &table.dictionaries {
2947 match dictionary {
2948 None => out.push(0),
2949 Some(page) => {
2950 out.push(1);
2951 put_u64(&mut out, page.offset);
2952 put_u32(&mut out, page.length);
2953 put_u64(&mut out, page.hash);
2954 }
2955 }
2956 }
2957 for distinct in &table.distincts {
2958 match distinct {
2959 None => out.push(0),
2960 Some(count) => {
2961 out.push(1);
2962 put_u64(&mut out, *count);
2963 }
2964 }
2965 }
2966 put_u64(&mut out, u64::try_from(table.rows).map_err(|_| invalid("row count overflow"))?);
2967 put_u32(&mut out, u32::try_from(table.stripes.len()).map_err(|_| invalid("too many stripes"))?);
2968 for stripe in &table.stripes {
2969 put_u32(
2970 &mut out,
2971 u32::try_from(stripe.parts.len()).map_err(|_| invalid("too many parts in a stripe"))?,
2972 );
2973 for &rows in &stripe.parts {
2974 put_u32(&mut out, rows);
2975 }
2976 put_u64(&mut out, stripe.index.offset);
2977 put_u32(&mut out, stripe.index.length);
2978 for page in &stripe.pages {
2979 put_u64(&mut out, page.offset);
2980 put_u32(&mut out, page.length);
2981 }
2982 for (field, membership) in table.fields.iter().zip(&stripe.memberships) {
2983 if field.ty != LogicalType::Varchar {
2984 continue;
2985 }
2986 let page =
2987 membership.ok_or_else(|| invalid("string page has no code membership index"))?;
2988 put_u64(&mut out, page.offset);
2989 put_u32(&mut out, page.length);
2990 put_u64(&mut out, page.hash);
2991 }
2992 for sieve in &stripe.sieves {
2993 match sieve {
2994 None => out.push(0),
2995 Some(page) => {
2996 out.push(1);
2997 put_u64(&mut out, page.offset);
2998 put_u32(&mut out, page.length);
2999 put_u64(&mut out, page.hash);
3000 }
3001 }
3002 }
3003 for range in stripe.zone.columns() {
3004 put_bound(&mut out, range.low.as_ref())?;
3005 put_bound(&mut out, range.high.as_ref())?;
3006 put_u32(
3007 &mut out,
3008 u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?,
3009 );
3010 out.push(u8::from(range.exact));
3011 match range.sum {
3012 None => out.push(0),
3013 Some(total) => {
3014 out.push(1);
3015 out.extend_from_slice(&total.to_le_bytes());
3016 }
3017 }
3018 }
3019 }
3020 out.extend_from_slice(FREQUENCIES);
3021 put_u16(
3022 &mut out,
3023 u16::try_from(table.frequencies.len())
3024 .map_err(|_| invalid("too many frequency columns"))?,
3025 );
3026 for summary in &table.frequencies {
3027 let Some(summary) = summary else {
3028 out.push(0);
3029 continue;
3030 };
3031 out.push(1);
3032 put_u64(&mut out, summary.omitted_max);
3033 put_u32(
3034 &mut out,
3035 u32::try_from(summary.entries.len())
3036 .map_err(|_| invalid("too many frequency entries"))?,
3037 );
3038 for entry in &summary.entries {
3039 match entry.value {
3040 FrequencyValue::Null => out.push(0),
3041 FrequencyValue::Integer(value) => {
3042 out.push(1);
3043 out.extend_from_slice(&value.to_le_bytes());
3044 }
3045 FrequencyValue::Code(value) => {
3046 out.push(2);
3047 put_u32(&mut out, value);
3048 }
3049 }
3050 put_u64(&mut out, entry.count);
3051 }
3052 put_u32(
3053 &mut out,
3054 u32::try_from(summary.ordinals.len())
3055 .map_err(|_| invalid("too many frequency ordinals"))?,
3056 );
3057 let mut previous = 0_u64;
3058 for (at, &ordinal) in summary.ordinals.iter().enumerate() {
3059 let delta = if at == 0 {
3060 ordinal
3061 } else {
3062 ordinal
3063 .checked_sub(previous)
3064 .ok_or_else(|| invalid("frequency ordinals are not ordered"))?
3065 };
3066 if at != 0 && delta == 0 {
3067 return Err(invalid("frequency ordinals are not unique"));
3068 }
3069 put_var_u64(&mut out, delta);
3070 previous = ordinal;
3071 }
3072 }
3073 Ok(out)
3074}
3075
3076struct Cursor<'a> {
3077 bytes: &'a [u8],
3078 at: usize,
3079}
3080impl<'a> Cursor<'a> {
3081 fn take(&mut self, len: usize) -> Result<&'a [u8]> {
3082 let end = self.at.checked_add(len).ok_or_else(|| invalid("directory offset overflow"))?;
3083 let bytes =
3084 self.bytes.get(self.at..end).ok_or_else(|| invalid("directory is truncated"))?;
3085 self.at = end;
3086 Ok(bytes)
3087 }
3088 fn u8(&mut self) -> Result<u8> {
3089 Ok(self.take(1)?[0])
3090 }
3091 fn u16(&mut self) -> Result<u16> {
3092 Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("two bytes")))
3093 }
3094 fn u32(&mut self) -> Result<u32> {
3095 Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("four bytes")))
3096 }
3097 fn u64(&mut self) -> Result<u64> {
3098 Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("eight bytes")))
3099 }
3100 fn var_u64(&mut self) -> Result<u64> {
3101 let mut value = 0_u64;
3102 for shift in (0..=63).step_by(7) {
3103 let byte = self.u8()?;
3104 let part = u64::from(byte & 0x7f);
3105 if shift == 63 && part > 1 {
3106 return Err(invalid("frequency ordinal varint overflows"));
3107 }
3108 value |= part << shift;
3109 if byte & 0x80 == 0 {
3110 return Ok(value);
3111 }
3112 }
3113 Err(invalid("frequency ordinal varint is too long"))
3114 }
3115 fn bound(&mut self) -> Result<Option<Bound>> {
3116 Ok(match self.u8()? {
3117 0 => None,
3118 1 => Some(Bound::Int(i128::from_le_bytes(
3119 self.take(16)?.try_into().expect("sixteen bytes"),
3120 ))),
3121 2 => Some(Bound::Real(f64::from_le_bytes(
3122 self.take(8)?.try_into().expect("eight bytes"),
3123 ))),
3124 3 => {
3125 let length = self.u32()? as usize;
3126 Some(Bound::Bytes(self.take(length)?.to_vec()))
3127 }
3128 _ => return Err(invalid("bound tag differs")),
3129 })
3130 }
3131 fn text(&mut self) -> Result<String> {
3132 let len = self.u16()? as usize;
3133 String::from_utf8(self.take(len)?.to_vec()).map_err(|_| invalid("name is not UTF-8"))
3134 }
3135}
3136
3137fn decode_directory(bytes: &[u8], size: u64) -> Result<Table> {
3138 let mut cur = Cursor { bytes, at: 0 };
3139 if cur.take(8)? != DIRECTORY {
3140 return Err(invalid("directory magic differs"));
3141 }
3142 let name = cur.text()?;
3143 let width = cur.u16()? as usize;
3144 let mut fields = Vec::with_capacity(width);
3145 for _ in 0..width {
3146 let name = cur.text()?;
3147 let ty = tag_type(cur.u8()?)?;
3148 let not_null = match cur.u8()? {
3149 0 => false,
3150 1 => true,
3151 _ => return Err(invalid("nullability flag differs")),
3152 };
3153 fields.push(Field { name, ty, not_null });
3154 }
3155 let mut dictionaries = Vec::with_capacity(width);
3156 for _ in 0..width {
3157 dictionaries.push(match cur.u8()? {
3158 0 => None,
3159 1 => {
3160 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3161 let end = page
3162 .offset
3163 .checked_add(u64::from(page.length))
3164 .ok_or_else(|| invalid("dictionary page offset overflow"))?;
3165 if page.offset < HEADER || end > size {
3170 return Err(invalid("dictionary page range is outside the file"));
3171 }
3172 Some(page)
3173 }
3174 _ => return Err(invalid("dictionary page tag differs")),
3175 });
3176 }
3177 let mut distincts = Vec::with_capacity(width);
3178 for _ in 0..width {
3179 distincts.push(match cur.u8()? {
3180 0 => None,
3181 1 => Some(cur.u64()?),
3182 _ => return Err(invalid("distinct count tag differs")),
3183 });
3184 }
3185 let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
3186 let count = cur.u32()? as usize;
3187 let mut stripes = Vec::with_capacity(count);
3188 let mut total = 0_usize;
3189 for _ in 0..count {
3190 let count = cur.u32()? as usize;
3191 if count == 0 || count > STRIPE_PARTS {
3192 return Err(invalid("stripe part count is outside its bound"));
3193 }
3194 let mut parts = Vec::with_capacity(count);
3195 let mut stripe_rows = 0_usize;
3196 for _ in 0..count {
3197 let rows = cur.u32()?;
3198 if rows == 0 {
3199 return Err(invalid("empty part"));
3200 }
3201 parts.push(rows);
3202 stripe_rows = stripe_rows
3203 .checked_add(rows as usize)
3204 .ok_or_else(|| invalid("stripe row count overflow"))?;
3205 }
3206 total =
3207 total.checked_add(stripe_rows).ok_or_else(|| invalid("stripe row count overflow"))?;
3208 let index = Span { offset: cur.u64()?, length: cur.u32()? };
3209 let section = index_section(count)?;
3210 let wanted = section
3211 .checked_mul(width)
3212 .and_then(|bytes| u32::try_from(bytes).ok())
3213 .ok_or_else(|| invalid("index page length overflow"))?;
3214 let end = index
3215 .offset
3216 .checked_add(u64::from(index.length))
3217 .ok_or_else(|| invalid("index page offset overflow"))?;
3218 if index.offset < HEADER || end > size || index.length != wanted {
3219 return Err(invalid("index page range is outside the file"));
3220 }
3221 let mut pages = Vec::with_capacity(width);
3222 for _ in 0..width {
3223 let offset = cur.u64()?;
3224 let length = cur.u32()?;
3225 let end = offset
3226 .checked_add(u64::from(length))
3227 .ok_or_else(|| invalid("page offset overflow"))?;
3228 if offset < HEADER || end > size || length as usize > MAX_PAGE {
3229 return Err(invalid("page range is outside the file"));
3230 }
3231 pages.push(Span { offset, length });
3232 }
3233 let mut memberships = vec![None; width];
3234 for (column, field) in fields.iter().enumerate() {
3235 if field.ty != LogicalType::Varchar {
3236 continue;
3237 }
3238 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3239 let end = page
3240 .offset
3241 .checked_add(u64::from(page.length))
3242 .ok_or_else(|| invalid("membership page offset overflow"))?;
3243 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3244 return Err(invalid("membership page range is outside the file"));
3245 }
3246 memberships[column] = Some(page);
3247 }
3248 let mut sieves = vec![None; width];
3249 for sieve in sieves.iter_mut().take(width) {
3250 match cur.u8()? {
3251 0 => continue,
3252 1 => {}
3253 _ => return Err(invalid("a sieve page has an unknown tag")),
3254 }
3255 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3256 let end = page
3257 .offset
3258 .checked_add(u64::from(page.length))
3259 .ok_or_else(|| invalid("sieve page offset overflow"))?;
3260 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3261 return Err(invalid("sieve page range is outside the file"));
3262 }
3263 *sieve = Some(page);
3264 }
3265 let mut ranges = Vec::with_capacity(width);
3266 for _ in 0..width {
3267 let low = cur.bound()?;
3268 let high = cur.bound()?;
3269 let nulls = cur.u32()? as usize;
3270 if nulls > stripe_rows {
3271 return Err(invalid("null count exceeds stripe rows"));
3272 }
3273 let exact = cur.u8()? != 0;
3274 let sum = match cur.u8()? {
3275 0 => None,
3276 1 => Some(i128::from_le_bytes(
3277 cur.take(16)?.try_into().map_err(|_| invalid("a stripe sum is truncated"))?,
3278 )),
3279 _ => return Err(invalid("a stripe sum has an unknown tag")),
3280 };
3281 ranges.push(Range { low, high, nulls, exact, sum });
3282 }
3283 stripes.push(Stripe {
3284 rows: stripe_rows,
3285 parts,
3286 index,
3287 pages,
3288 memberships,
3289 sieves,
3290 zone: Zone::from_ranges(ranges),
3291 });
3292 }
3293 if total != rows {
3294 return Err(invalid("table row count differs from stripes"));
3295 }
3296 let frequencies = if cur.at == bytes.len() {
3297 vec![None; width]
3298 } else {
3299 if cur.take(8)? != FREQUENCIES {
3300 return Err(invalid("directory extension magic differs"));
3301 }
3302 if cur.u16()? as usize != width {
3303 return Err(invalid("frequency column count differs"));
3304 }
3305 let mut frequencies = Vec::with_capacity(width);
3306 for field in &fields {
3307 let summary = match cur.u8()? {
3308 0 => None,
3309 1 => {
3310 let omitted_max = cur.u64()?;
3311 let count = cur.u32()? as usize;
3312 if count > FREQUENCY_ENTRIES {
3313 return Err(invalid("frequency entry count exceeds its bound"));
3314 }
3315 let mut entries = Vec::with_capacity(count);
3316 for _ in 0..count {
3318 let value = match cur.u8()? {
3319 0 => FrequencyValue::Null,
3320 1 => FrequencyValue::Integer(i128::from_le_bytes(
3321 cur.take(16)?.try_into().expect("sixteen bytes"),
3322 )),
3323 2 => FrequencyValue::Code(cur.u32()?),
3324 _ => return Err(invalid("frequency value tag differs")),
3325 };
3326 let valid = matches!(
3327 (&field.ty, value),
3328 (_, FrequencyValue::Null)
3329 | (LogicalType::Varchar, FrequencyValue::Code(_))
3330 | (
3331 LogicalType::TinyInt
3332 | LogicalType::SmallInt
3333 | LogicalType::Integer
3334 | LogicalType::BigInt
3335 | LogicalType::UTinyInt
3336 | LogicalType::USmallInt
3337 | LogicalType::UInteger
3338 | LogicalType::UBigInt
3339 | LogicalType::Date
3340 | LogicalType::Timestamp,
3341 FrequencyValue::Integer(_),
3342 )
3343 );
3344 if !valid {
3345 return Err(invalid("frequency value does not match its column"));
3346 }
3347 let count = cur.u64()?;
3348 if count == 0 || count > rows as u64 {
3349 return Err(invalid("frequency count is outside the table"));
3350 }
3351 entries.push(FrequencyEntry { value, count });
3352 }
3353 if entries.windows(2).any(|pair| pair[0].count < pair[1].count) {
3354 return Err(invalid("frequency entries are not descending"));
3355 }
3356 let ordinals = {
3357 let ordinal_count = cur.u32()? as usize;
3358 if ordinal_count > FREQUENCY_ORDINALS || ordinal_count > rows {
3359 return Err(invalid("frequency ordinal count exceeds its bound"));
3360 }
3361 let mut ordinals = Vec::with_capacity(ordinal_count);
3362 let mut previous = 0_u64;
3363 for at in 0..ordinal_count {
3364 let delta = cur.var_u64()?;
3365 if at != 0 && delta == 0 {
3366 return Err(invalid("frequency ordinals are not increasing"));
3367 }
3368 let ordinal = if at == 0 {
3369 delta
3370 } else {
3371 previous
3372 .checked_add(delta)
3373 .ok_or_else(|| invalid("frequency ordinal overflows"))?
3374 };
3375 if ordinal >= rows as u64 {
3376 return Err(invalid("frequency ordinal is outside the table"));
3377 }
3378 ordinals.push(ordinal);
3379 previous = ordinal;
3380 }
3381 ordinals
3382 };
3383 Some(FrequencySummary { entries, omitted_max, ordinals })
3384 }
3385 _ => return Err(invalid("frequency summary tag differs")),
3386 };
3387 frequencies.push(summary);
3388 }
3389 frequencies
3390 };
3391 if cur.at != bytes.len() {
3392 return Err(invalid("directory has trailing bytes"));
3393 }
3394 Ok(Table { name, fields, stripes, rows, dictionaries, distincts, frequencies })
3395}
3396
3397fn put_bound(out: &mut Vec<u8>, bound: Option<&Bound>) -> Result<()> {
3398 match bound {
3399 None => out.push(0),
3400 Some(Bound::Int(value)) => {
3401 out.push(1);
3402 out.extend_from_slice(&value.to_le_bytes());
3403 }
3404 Some(Bound::Real(value)) => {
3405 out.push(2);
3406 out.extend_from_slice(&value.to_le_bytes());
3407 }
3408 Some(Bound::Bytes(value)) => {
3409 out.push(3);
3410 put_u32(out, u32::try_from(value.len()).map_err(|_| invalid("bound length overflow"))?);
3411 out.extend_from_slice(value);
3412 }
3413 }
3414 Ok(())
3415}
3416
3417#[derive(Debug)]
3434struct Codes;
3435
3436impl chooser::Chooser for Codes {
3437 fn name(&self) -> &'static str {
3438 "codes"
3439 }
3440
3441 fn narrow_strings(
3442 &self,
3443 _values: &[&[u8]],
3444 offered: &[string::Kind],
3445 _depth: u8,
3446 ) -> Vec<string::Kind> {
3447 offered.to_vec()
3450 }
3451
3452 fn narrow_integers(
3453 &self,
3454 _values: &[i64],
3455 offered: &[integer::Kind],
3456 depth: u8,
3457 ) -> Vec<integer::Kind> {
3458 let keep: &[integer::Kind] = if depth == 0 {
3459 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Rle]
3460 } else {
3461 &[integer::Kind::Constant, integer::Kind::Packed]
3462 };
3463 let narrowed: Vec<integer::Kind> =
3464 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3465 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3468 }
3469}
3470
3471#[derive(Debug)]
3483struct Fixed;
3484
3485impl chooser::Chooser for Fixed {
3486 fn name(&self) -> &'static str {
3487 "fixed"
3488 }
3489
3490 fn narrow_strings(
3491 &self,
3492 _values: &[&[u8]],
3493 offered: &[string::Kind],
3494 _depth: u8,
3495 ) -> Vec<string::Kind> {
3496 offered.to_vec()
3497 }
3498
3499 fn narrow_integers(
3500 &self,
3501 _values: &[i64],
3502 offered: &[integer::Kind],
3503 depth: u8,
3504 ) -> Vec<integer::Kind> {
3505 let keep: &[integer::Kind] = if depth == 0 {
3506 &[
3507 integer::Kind::Constant,
3508 integer::Kind::Packed,
3509 integer::Kind::Delta,
3510 integer::Kind::Rle,
3511 integer::Kind::Sparse,
3512 integer::Kind::Strided,
3513 ]
3514 } else {
3515 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Delta]
3516 };
3517 let narrowed: Vec<integer::Kind> =
3518 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3519 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3520 }
3521}
3522
3523fn widened(data: &Data) -> Option<Vec<i64>> {
3530 match data {
3531 Data::Int8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3532 Data::UInt8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3533 Data::Int16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3534 Data::UInt16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3535 Data::Int32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3536 Data::UInt32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3537 Data::Int64(values) => Some(values.to_vec()),
3538 _ => None,
3539 }
3540}
3541
3542fn narrowed(ty: &LogicalType, values: Vec<i64>) -> Result<Data> {
3547 fn fit<T: TryFrom<i64>>(values: &[i64]) -> Result<Vec<T>> {
3548 values
3549 .iter()
3550 .map(|value| T::try_from(*value).map_err(|_| invalid("page value is not of its type")))
3551 .collect()
3552 }
3553 Ok(match ty {
3554 LogicalType::TinyInt => Data::Int8(fit::<i8>(&values)?.into()),
3555 LogicalType::UTinyInt => Data::UInt8(fit::<u8>(&values)?.into()),
3556 LogicalType::SmallInt => Data::Int16(fit::<i16>(&values)?.into()),
3557 LogicalType::USmallInt => Data::UInt16(fit::<u16>(&values)?.into()),
3558 LogicalType::Integer | LogicalType::Date => Data::Int32(fit::<i32>(&values)?.into()),
3559 LogicalType::UInteger => Data::UInt32(fit::<u32>(&values)?.into()),
3560 LogicalType::BigInt | LogicalType::Timestamp => Data::Int64(values.into()),
3561 _ => return Err(invalid("cascade codec belongs to a page that is not integers")),
3562 })
3563}
3564
3565fn plain_width(ty: &LogicalType) -> Option<usize> {
3568 Some(match ty {
3569 LogicalType::TinyInt | LogicalType::UTinyInt => 1,
3570 LogicalType::SmallInt | LogicalType::USmallInt => 2,
3571 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date => 4,
3572 LogicalType::BigInt | LogicalType::Timestamp => 8,
3573 _ => return None,
3574 })
3575}
3576
3577fn cascaded(
3583 flat: &Vector,
3584 ty: &LogicalType,
3585 packed: Option<&Packed<'_>>,
3586) -> Result<Option<Vec<u8>>> {
3587 let (Some(width), Some(data)) = (plain_width(ty), flat.data()) else { return Ok(None) };
3588 let Some(values) = widened(data) else { return Ok(None) };
3589 let plain = values.len().saturating_mul(width);
3590 let best = match packed {
3591 Some(packed) => plain.min(21 + size_of_val(packed.words())),
3593 None => plain,
3594 };
3595 let out = integer::encode_with(&values, &Fixed)?;
3596 Ok((out.len() < best).then_some(out))
3597}
3598
3599fn encoded_codes(codes: &[u32]) -> Result<Option<Vec<u8>>> {
3611 let wide: Vec<i64> = codes.iter().map(|code| i64::from(*code)).collect();
3612 let coded = integer::encode_with(&wide, &Codes)?;
3613 let plain = codes.len().saturating_mul(size_of::<u32>());
3614 Ok((coded.len() < plain).then_some(coded))
3615}
3616
3617fn encode(
3618 vector: &Vector,
3619 global: Option<&mut GlobalDictionary>,
3620) -> Result<(Vec<u8>, Option<Vec<u32>>)> {
3621 let ty = vector.logical_type();
3622 let flat = vector.flatten()?;
3624 let mut out = Vec::new();
3625 let mut global_codes = None;
3626 if let Some(global) = global {
3627 let mut codes = Vec::with_capacity(flat.len());
3628 for row in 0..flat.len() {
3629 let text = flat.text_at(row).unwrap_or("");
3630 let code = global.code(text)?;
3631 global.observe(code, flat.is_null_at(row))?;
3632 codes.push(code);
3633 }
3634 global_codes = Some(codes);
3635 }
3636 let membership = global_codes.as_deref().map(unique_codes);
3637 let dictionary = if global_codes.is_none() && ty == &LogicalType::Varchar {
3638 string_dictionary(&flat)?
3639 } else {
3640 None
3641 };
3642 let packed_vector = if dictionary.is_none() && global_codes.is_none() {
3643 Some(flat.bit_packed()?)
3644 } else {
3645 None
3646 };
3647 let packed = packed_vector.as_ref().and_then(Vector::packed_parts);
3648 let coded = match global_codes.as_deref() {
3649 Some(codes) => encoded_codes(codes)?,
3650 None => None,
3651 };
3652 let cascade = if dictionary.is_none() && global_codes.is_none() {
3656 cascaded(&flat, ty, packed.as_ref())?
3657 } else {
3658 None
3659 };
3660 out.push(if coded.is_some() {
3661 4
3662 } else if cascade.is_some() {
3663 5
3664 } else if global_codes.is_some() {
3665 3
3666 } else if dictionary.is_some() {
3667 1
3668 } else if packed.is_some() {
3669 2
3670 } else {
3671 0
3672 });
3673 let nulls = flat.validity();
3674 let flag = match nulls {
3675 Validity::AllValid => 0,
3676 Validity::AllInvalid => 1,
3677 Validity::Mask(_) => 2,
3678 };
3679 out.push(flag);
3680 if flag == 2 {
3681 for group in (0..vector.len()).step_by(8) {
3682 let mut bits = 0_u8;
3683 for bit in 0..8 {
3684 if group + bit < vector.len() && !flat.is_null_at(group + bit) {
3685 bits |= 1 << bit;
3686 }
3687 }
3688 out.push(bits);
3689 }
3690 }
3691 if let Some(coded) = coded {
3692 out.extend_from_slice(&coded);
3693 return Ok((out, membership));
3694 }
3695 if let Some(cascade) = cascade {
3696 out.extend_from_slice(&cascade);
3697 return Ok((out, membership));
3698 }
3699 if let Some(codes) = global_codes {
3700 for code in codes {
3701 put_u32(&mut out, code);
3702 }
3703 return Ok((out, membership));
3704 }
3705 if let Some(dictionary) = dictionary {
3706 out.extend_from_slice(&dictionary);
3707 return Ok((out, membership));
3708 }
3709 if let Some(packed) = packed {
3710 if packed.offset() != 0 {
3711 return Err(invalid("writer received a sliced packed vector"));
3712 }
3713 out.push(u8::try_from(packed.width()).map_err(|_| invalid("packed width overflow"))?);
3714 out.extend_from_slice(&packed.base().to_le_bytes());
3715 put_u32(
3716 &mut out,
3717 u32::try_from(packed.words().len()).map_err(|_| invalid("too many packed words"))?,
3718 );
3719 for word in packed.words() {
3720 put_u64(&mut out, *word);
3721 }
3722 return Ok((out, membership));
3723 }
3724 let data = flat.data().ok_or_else(|| invalid("scalar column did not flatten"))?;
3725 match (ty, data) {
3726 (LogicalType::TinyInt, Data::Int8(values)) => {
3727 for value in &**values {
3728 out.extend_from_slice(&value.to_le_bytes());
3729 }
3730 }
3731 (LogicalType::UTinyInt, Data::UInt8(values)) => {
3732 for value in &**values {
3733 out.extend_from_slice(&value.to_le_bytes());
3734 }
3735 }
3736 (LogicalType::SmallInt, Data::Int16(values)) => {
3737 for value in &**values {
3738 out.extend_from_slice(&value.to_le_bytes());
3739 }
3740 }
3741 (LogicalType::USmallInt, Data::UInt16(values)) => {
3742 for value in &**values {
3743 out.extend_from_slice(&value.to_le_bytes());
3744 }
3745 }
3746 (LogicalType::UInteger, Data::UInt32(values)) => {
3747 for value in &**values {
3748 out.extend_from_slice(&value.to_le_bytes());
3749 }
3750 }
3751 (LogicalType::UBigInt, Data::UInt64(values)) => {
3752 for value in &**values {
3753 out.extend_from_slice(&value.to_le_bytes());
3754 }
3755 }
3756 (LogicalType::Integer | LogicalType::Date, Data::Int32(values)) => {
3757 for value in &**values {
3758 out.extend_from_slice(&value.to_le_bytes());
3759 }
3760 }
3761 (LogicalType::BigInt | LogicalType::Timestamp, Data::Int64(values)) => {
3762 for value in &**values {
3763 out.extend_from_slice(&value.to_le_bytes());
3764 }
3765 }
3766 (LogicalType::Boolean, Data::Bool(values)) => {
3767 for value in &**values {
3768 out.push(u8::from(*value));
3769 }
3770 }
3771 (LogicalType::Varchar, Data::Varlen(values)) => {
3772 let mut bytes = Vec::new();
3773 put_u32(&mut out, 0);
3774 for row in 0..vector.len() {
3775 let value = values.bytes(row).ok_or_else(|| invalid("string view is invalid"))?;
3776 bytes.extend_from_slice(value);
3777 put_u32(
3778 &mut out,
3779 u32::try_from(bytes.len())
3780 .map_err(|_| invalid("string payload exceeds 4GiB"))?,
3781 );
3782 }
3783 out.extend_from_slice(&bytes);
3784 }
3785 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
3786 }
3787 Ok((out, membership))
3788}
3789
3790fn put_varint(out: &mut Vec<u8>, mut value: u32) {
3791 while value >= 0x80 {
3792 out.push((value as u8 & 0x7f) | 0x80);
3793 value >>= 7;
3794 }
3795 out.push(value as u8);
3796}
3797
3798fn unique_codes(codes: &[u32]) -> Vec<u32> {
3800 let mut unique = codes.to_vec();
3801 unique.sort_unstable();
3802 unique.dedup();
3803 unique
3804}
3805
3806fn merged_codes(lists: Vec<Vec<u32>>) -> Vec<u32> {
3812 let mut lists = lists;
3813 while lists.len() > 1 {
3814 let mut next = Vec::with_capacity(lists.len().div_ceil(2));
3815 for pair in lists.chunks(2) {
3816 match pair {
3817 [left, right] => next.push(merged_pair(left, right)),
3818 [only] => next.push(only.clone()),
3819 _ => {}
3820 }
3821 }
3822 lists = next;
3823 }
3824 lists.pop().unwrap_or_default()
3825}
3826
3827fn merged_pair(left: &[u32], right: &[u32]) -> Vec<u32> {
3828 let mut out = Vec::with_capacity(left.len().saturating_add(right.len()));
3829 let mut at = 0;
3830 let mut to = 0;
3831 while at < left.len() && to < right.len() {
3832 match left[at].cmp(&right[to]) {
3833 Ordering::Less => {
3834 out.push(left[at]);
3835 at += 1;
3836 }
3837 Ordering::Greater => {
3838 out.push(right[to]);
3839 to += 1;
3840 }
3841 Ordering::Equal => {
3842 out.push(left[at]);
3843 at += 1;
3844 to += 1;
3845 }
3846 }
3847 }
3848 out.extend_from_slice(&left[at..]);
3849 out.extend_from_slice(&right[to..]);
3850 out
3851}
3852
3853fn merged_range(ranges: impl Iterator<Item = Range>) -> Range {
3858 let mut merged = Range::default();
3859 let mut first = true;
3860 for range in ranges {
3861 merged.nulls = merged.nulls.saturating_add(range.nulls);
3862 merged.sum = match (merged.sum.take(), range.sum) {
3866 (Some(held), Some(next)) if !first => held.checked_add(next),
3867 (_, next) if first => next,
3868 _ => None,
3869 };
3870 merged.exact = if first { range.exact } else { merged.exact && range.exact };
3871 if first {
3872 merged.low = range.low;
3873 merged.high = range.high;
3874 first = false;
3875 continue;
3876 }
3877 merged.low = match (merged.low.take(), range.low) {
3878 (Some(held), Some(next)) => Some(held.smaller(next)),
3879 _ => None,
3880 };
3881 merged.high = match (merged.high.take(), range.high) {
3882 (Some(held), Some(next)) => Some(held.larger(next)),
3883 _ => None,
3884 };
3885 }
3886 merged
3887}
3888
3889fn encode_sieves<'a>(sieves: impl Iterator<Item = &'a Option<Sieve>>) -> Result<Vec<u8>> {
3895 let held: Vec<&Option<Sieve>> = sieves.collect();
3896 let mut out = Vec::new();
3897 put_u32(
3898 &mut out,
3899 u32::try_from(held.len()).map_err(|_| invalid("too many parts in a stripe"))?,
3900 );
3901 for sieve in &held {
3902 let length = sieve.as_ref().map_or(0, Sieve::len);
3903 put_u32(&mut out, u32::try_from(length).map_err(|_| invalid("sieve length overflow"))?);
3904 }
3905 for sieve in held.into_iter().flatten() {
3907 out.extend_from_slice(&sieve.to_bytes());
3908 }
3909 Ok(out)
3910}
3911
3912fn decode_sieves(bytes: &[u8]) -> Result<Vec<Option<Sieve>>> {
3918 let parts = u32::from_le_bytes(
3919 bytes
3920 .get(..4)
3921 .ok_or_else(|| invalid("sieve page is truncated"))?
3922 .try_into()
3923 .map_err(|_| invalid("sieve page is truncated"))?,
3924 ) as usize;
3925 let mut lengths = Vec::with_capacity(parts);
3926 for part in 0..parts {
3927 let at = 4 + part * 4;
3928 let field = bytes.get(at..at + 4).ok_or_else(|| invalid("sieve page is truncated"))?;
3929 lengths.push(u32::from_le_bytes(
3930 field.try_into().map_err(|_| invalid("sieve page is truncated"))?,
3931 ) as usize);
3932 }
3933 let mut at = 4 + parts * 4;
3934 let mut out = Vec::with_capacity(parts);
3935 for length in lengths {
3936 if length == 0 {
3937 out.push(None);
3938 continue;
3939 }
3940 let end = at.checked_add(length).ok_or_else(|| invalid("sieve page is truncated"))?;
3941 let field = bytes.get(at..end).ok_or_else(|| invalid("sieve page is truncated"))?;
3942 out.push(Sieve::from_bytes(field));
3943 at = end;
3944 }
3945 if at != bytes.len() {
3946 return Err(invalid("sieve page has trailing bytes"));
3947 }
3948 Ok(out)
3949}
3950
3951fn encode_membership(unique: &[u32]) -> Vec<u8> {
3957 let mut out = Vec::with_capacity(unique.len().saturating_mul(2).saturating_add(5));
3958 put_varint(&mut out, u32::try_from(unique.len()).unwrap_or(u32::MAX));
3959 let mut previous = 0;
3960 for (at, &code) in unique.iter().enumerate() {
3961 put_varint(&mut out, if at == 0 { code } else { code - previous });
3962 previous = code;
3963 }
3964 out
3965}
3966
3967fn take_varint(bytes: &[u8], at: &mut usize) -> Result<u32> {
3968 let mut value = 0_u32;
3969 for shift in (0..35).step_by(7) {
3970 let byte = *bytes.get(*at).ok_or_else(|| invalid("membership varint is truncated"))?;
3971 *at += 1;
3972 let part = u32::from(byte & 0x7f);
3973 if shift == 28 && part > 0x0f {
3974 return Err(invalid("membership varint overflow"));
3975 }
3976 value = value
3977 .checked_add(
3978 part.checked_shl(shift).ok_or_else(|| invalid("membership varint overflow"))?,
3979 )
3980 .ok_or_else(|| invalid("membership varint overflow"))?;
3981 if byte & 0x80 == 0 {
3982 return Ok(value);
3983 }
3984 }
3985 Err(invalid("membership varint is too long"))
3986}
3987
3988fn decode_membership(bytes: &[u8]) -> Result<Vec<u32>> {
3989 let mut at = 0;
3990 let count = take_varint(bytes, &mut at)? as usize;
3991 let mut codes = Vec::with_capacity(count);
3992 let mut previous = 0_u32;
3993 for index in 0..count {
3994 let delta = take_varint(bytes, &mut at)?;
3995 let code = if index == 0 {
3996 delta
3997 } else {
3998 previous.checked_add(delta).ok_or_else(|| invalid("membership code overflow"))?
3999 };
4000 if index > 0 && code <= previous {
4001 return Err(invalid("membership codes are not increasing"));
4002 }
4003 codes.push(code);
4004 previous = code;
4005 }
4006 if at != bytes.len() {
4007 return Err(invalid("membership page has trailing bytes"));
4008 }
4009 Ok(codes)
4010}
4011
4012fn string_dictionary(vector: &Vector) -> Result<Option<Vec<u8>>> {
4013 let mut by_text = HashMap::new();
4014 let mut values = Vec::new();
4015 let mut codes = Vec::with_capacity(vector.len());
4016 let mut plain_bytes = 0_usize;
4017 for row in 0..vector.len() {
4018 let text = vector.text_at(row).unwrap_or("");
4019 plain_bytes = plain_bytes.saturating_add(text.len());
4020 let code = match by_text.get(text) {
4021 Some(&code) => code,
4022 None => {
4023 let code = u32::try_from(values.len())
4024 .map_err(|_| invalid("too many dictionary values"))?;
4025 by_text.insert(text, code);
4026 values.push(text);
4027 code
4028 }
4029 };
4030 codes.push(code);
4031 }
4032 let dictionary_bytes = values.iter().map(|value| value.len()).sum::<usize>();
4033 let encoded = 8_usize
4034 .saturating_add((values.len() + 1).saturating_mul(4))
4035 .saturating_add(dictionary_bytes)
4036 .saturating_add(codes.len().saturating_mul(4));
4037 let plain = (vector.len() + 1).saturating_mul(4).saturating_add(plain_bytes);
4038 if encoded >= plain {
4039 return Ok(None);
4040 }
4041 let mut out = Vec::with_capacity(encoded);
4042 put_u32(
4043 &mut out,
4044 u32::try_from(values.len()).map_err(|_| invalid("too many dictionary values"))?,
4045 );
4046 put_u32(
4047 &mut out,
4048 u32::try_from(dictionary_bytes).map_err(|_| invalid("dictionary payload exceeds 4GiB"))?,
4049 );
4050 let mut offset = 0_u32;
4051 put_u32(&mut out, offset);
4052 for value in &values {
4053 offset = offset
4054 .checked_add(
4055 u32::try_from(value.len()).map_err(|_| invalid("dictionary value is too long"))?,
4056 )
4057 .ok_or_else(|| invalid("dictionary payload exceeds 4GiB"))?;
4058 put_u32(&mut out, offset);
4059 }
4060 for value in values {
4061 out.extend_from_slice(value.as_bytes());
4062 }
4063 for code in codes {
4064 put_u32(&mut out, code);
4065 }
4066 Ok(Some(out))
4067}
4068
4069struct EncodedDictionary {
4070 index: Vec<u8>,
4071 ranks: Vec<u8>,
4072 payload: Vec<Vec<u8>>,
4075}
4076
4077fn head(bytes: &[u8]) -> u64 {
4079 let mut word = [0; 8];
4080 let take = bytes.len().min(8);
4081 word[..take].copy_from_slice(&bytes[..take]);
4082 u64::from_be_bytes(word)
4083}
4084
4085fn rankings(dictionaries: &[Option<GlobalDictionary>]) -> Result<Vec<Vec<(u64, u32)>>> {
4093 let present =
4094 dictionaries.iter().enumerate().filter(|(_, held)| held.is_some()).map(|(at, _)| at);
4095 let present = present.collect::<Vec<_>>();
4096 let mut orders = vec![Vec::new(); dictionaries.len()];
4097 let workers = std::thread::available_parallelism()
4098 .map_or(1, usize::from)
4099 .min(MAX_FREQUENCY_WORKERS)
4100 .min(present.len());
4101 if workers <= 1 {
4102 for at in present {
4103 if let Some(dictionary) = &dictionaries[at] {
4104 orders[at] = dictionary.ranked();
4105 }
4106 }
4107 return Ok(orders);
4108 }
4109 let width = present.len().div_ceil(workers);
4110 let pieces = std::thread::scope(|scope| {
4111 present
4112 .chunks(width)
4113 .map(|columns| {
4114 scope.spawn(|| {
4115 columns
4116 .iter()
4117 .filter_map(|&at| dictionaries[at].as_ref().map(|held| (at, held.ranked())))
4118 .collect::<Vec<_>>()
4119 })
4120 })
4121 .collect::<Vec<_>>()
4122 .into_iter()
4123 .map(|handle| {
4124 handle.join().map_err(|_| Error::internal("a dictionary sort worker panicked"))
4125 })
4126 .collect::<Result<Vec<_>>>()
4127 })?;
4128 for piece in pieces {
4129 for (at, order) in piece {
4130 orders[at] = order;
4131 }
4132 }
4133 Ok(orders)
4134}
4135
4136fn encode_global_dictionary(
4137 dictionary: GlobalDictionary,
4138 order: &[(u64, u32)],
4139) -> Result<EncodedDictionary> {
4140 let values = dictionary.offsets.len() - 1;
4141 if order.len() != values {
4142 return Err(invalid("global dictionary order does not cover its values"));
4143 }
4144 let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
4145 let payload = encode_payload(&dictionary)?;
4146 if payload.len() != blocks {
4147 return Err(invalid("global dictionary payload is not the blocks it says it is"));
4148 }
4149 let (ranks, rank_ends) = encode_ranks(order, code_width(values))?;
4150 let rank_blocks = values.div_ceil(TEXT_RANK_BLOCK);
4151 let offset_bits = offset_width(&dictionary.offsets);
4152 let mut index = Vec::with_capacity(
4153 DICTIONARY_HEADER + offset_bytes(values, offset_bits) + (blocks + rank_blocks) * 16,
4154 );
4155 put_u32(
4156 &mut index,
4157 u32::try_from(values).map_err(|_| invalid("global dictionary has too many values"))?,
4158 );
4159 put_u32(&mut index, TEXT_PAYLOAD_VALUES as u32);
4160 put_u32(
4161 &mut index,
4162 u32::try_from(blocks).map_err(|_| invalid("global dictionary has too many blocks"))?,
4163 );
4164 put_u32(&mut index, offset_bits as u32);
4165 encode_offsets(&dictionary.offsets, offset_bits, &mut index)?;
4166 let mut at = 0_u64;
4170 for block in &payload {
4171 at = at
4172 .checked_add(block.len() as u64)
4173 .ok_or_else(|| invalid("global dictionary payload overflow"))?;
4174 put_u64(&mut index, at);
4175 }
4176 for block in &payload {
4177 put_u64(&mut index, checksum(block));
4178 }
4179 if rank_ends.len() != rank_blocks {
4182 return Err(invalid("global dictionary order is not the blocks it says it is"));
4183 }
4184 for end in &rank_ends {
4185 put_u64(&mut index, *end);
4186 }
4187 let mut at = 0_usize;
4188 for end in &rank_ends {
4189 let end = usize::try_from(*end).map_err(|_| invalid("global dictionary order overflow"))?;
4190 put_u64(&mut index, checksum(&ranks[at..end]));
4191 at = end;
4192 }
4193 Ok(EncodedDictionary { index, ranks, payload })
4194}
4195
4196const PAYLOAD_SAMPLE_BLOCKS: usize = 8;
4203
4204fn payload_shapes() -> Vec<chooser::Settled> {
4230 let integers = vec![integer::Kind::Packed];
4231 [
4232 vec![string::Kind::Front, string::Kind::Lz],
4233 vec![string::Kind::Lz, string::Kind::Fsst],
4234 vec![string::Kind::Lz, string::Kind::Plain],
4235 vec![string::Kind::Fsst],
4236 vec![string::Kind::Plain],
4237 ]
4238 .into_iter()
4239 .map(|strings| chooser::Settled::new(strings, integers.clone()))
4240 .collect()
4241}
4242
4243fn encode_payload(dictionary: &GlobalDictionary) -> Result<Vec<Vec<u8>>> {
4249 let values = dictionary.offsets.len() - 1;
4250 let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
4251 let run = |block: usize| {
4252 let first = block * TEXT_PAYLOAD_VALUES;
4253 let last = (first + TEXT_PAYLOAD_VALUES).min(values);
4254 (first..last)
4255 .map(|value| {
4256 let from = dictionary.offsets[value] as usize;
4257 let to = dictionary.offsets[value + 1] as usize;
4258 &dictionary.payload[from..to]
4259 })
4260 .collect::<Vec<_>>()
4261 };
4262 let shape = (blocks > PAYLOAD_SAMPLE_BLOCKS).then(|| settle_shape(&run, blocks)).transpose()?;
4265 let one = |block: usize| match &shape {
4266 Some(shape) => string::encode_with(&run(block), shape),
4267 None => string::encode(&run(block)),
4268 };
4269 let workers = std::thread::available_parallelism()
4270 .map_or(1, usize::from)
4271 .min(MAX_FREQUENCY_WORKERS)
4272 .min(blocks);
4273 if workers <= 1 {
4274 return (0..blocks).map(one).collect();
4275 }
4276 let next = AtomicUsize::new(0);
4277 let pieces = std::thread::scope(|scope| {
4278 (0..workers)
4279 .map(|_| {
4280 scope.spawn(|| {
4281 let mut mine = Vec::new();
4282 loop {
4283 let block = next.fetch_add(1, Atomic::Relaxed);
4284 if block >= blocks {
4285 break;
4286 }
4287 mine.push((block, one(block)?));
4288 }
4289 Ok(mine)
4290 })
4291 })
4292 .collect::<Vec<_>>()
4293 .into_iter()
4294 .map(|handle| {
4295 handle.join().map_err(|_| Error::internal("a dictionary encode worker panicked"))?
4296 })
4297 .collect::<Result<Vec<_>>>()
4298 })?;
4299 let mut payload = vec![Vec::new(); blocks];
4300 for piece in pieces {
4301 for (block, bytes) in piece {
4302 payload[block] = bytes;
4303 }
4304 }
4305 Ok(payload)
4306}
4307
4308fn settle_shape<'a>(
4316 run: &dyn Fn(usize) -> Vec<&'a [u8]>,
4317 blocks: usize,
4318) -> Result<chooser::Settled> {
4319 let last = blocks - 1;
4320 let sample = (0..PAYLOAD_SAMPLE_BLOCKS)
4321 .map(|region| run(region * last / (PAYLOAD_SAMPLE_BLOCKS - 1)))
4322 .collect::<Vec<_>>();
4323 let mut best: Option<(chooser::Settled, usize)> = None;
4324 for shape in payload_shapes() {
4325 let mut size = 0;
4326 for block in &sample {
4327 size += string::encode_with(block, &shape)?.len();
4328 }
4329 if best.as_ref().is_none_or(|(_, smallest)| size < *smallest) {
4330 best = Some((shape, size));
4331 }
4332 }
4333 best.map(|(shape, _)| shape)
4334 .ok_or_else(|| invalid("no shape applies to a global dictionary payload"))
4335}
4336
4337fn encode_ranks(order: &[(u64, u32)], code_bits: usize) -> Result<(Vec<u8>, Vec<u64>)> {
4344 let mut out = Vec::with_capacity(order.len() * 4);
4345 let mut ends = Vec::with_capacity(order.len().div_ceil(TEXT_RANK_BLOCK));
4346 let mut heads = Vec::with_capacity(TEXT_RANK_BLOCK);
4347 let mut codes = Vec::with_capacity(TEXT_RANK_BLOCK);
4348 for block in order.chunks(TEXT_RANK_BLOCK) {
4349 let base = block.first().map_or(0, |&(head, _)| head);
4352 let span = block.last().map_or(0, |&(head, _)| head.wrapping_sub(base));
4353 let width = (u64::BITS - span.leading_zeros()) as usize;
4354 heads.clear();
4355 codes.clear();
4356 for &(head, code) in block {
4357 heads.push(head.wrapping_sub(base));
4358 codes.push(u64::from(code));
4359 }
4360 put_u64(&mut out, base);
4361 out.push(width as u8);
4362 bitpack::pack_tail(&heads, width, &mut out)
4363 .map_err(|_| invalid("global dictionary heads do not pack"))?;
4364 bitpack::pack_tail(&codes, code_bits, &mut out)
4365 .map_err(|_| invalid("global dictionary codes do not pack"))?;
4366 ends.push(out.len() as u64);
4367 }
4368 Ok((out, ends))
4369}
4370
4371fn open_global_dictionary(file: Arc<File>, page: Page, ty: &LogicalType) -> Result<Vector> {
4372 if ty != &LogicalType::Varchar {
4373 return Err(invalid("global dictionary belongs to a non-string column"));
4374 }
4375 let mut header = [0; DICTIONARY_HEADER];
4376 read_at(&file, page.offset, &mut header)?;
4377 let count = u32::from_le_bytes(header[0..4].try_into().expect("four bytes")) as usize;
4378 let per_block = u32::from_le_bytes(header[4..8].try_into().expect("four bytes")) as usize;
4379 let blocks = u32::from_le_bytes(header[8..12].try_into().expect("four bytes")) as usize;
4380 let offset_bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
4381 if per_block != TEXT_PAYLOAD_VALUES {
4382 return Err(invalid("global dictionary block width differs"));
4383 }
4384 if blocks != count.div_ceil(TEXT_PAYLOAD_VALUES) {
4385 return Err(invalid("global dictionary block count differs from its value count"));
4386 }
4387 if offset_bits > u32::BITS as usize {
4388 return Err(invalid("global dictionary packs offsets past a payload"));
4389 }
4390 let offset_len = offset_bytes(count, offset_bits);
4391 let ranks = count;
4396 let rank_blocks = ranks.div_ceil(TEXT_RANK_BLOCK);
4397 let hash_len = blocks
4400 .checked_add(rank_blocks)
4401 .and_then(|words| words.checked_mul(16))
4402 .ok_or_else(|| invalid("global dictionary block count overflow"))?;
4403 let index_len = DICTIONARY_HEADER
4404 .checked_add(offset_len)
4405 .and_then(|len| len.checked_add(hash_len))
4406 .ok_or_else(|| invalid("global dictionary header overflow"))?;
4407 if index_len > page.length as usize {
4408 return Err(invalid("global dictionary offset index exceeds its page"));
4409 }
4410 let mut index = vec![0; index_len];
4411 index[..DICTIONARY_HEADER].copy_from_slice(&header);
4412 read_at(&file, page.offset + DICTIONARY_HEADER as u64, &mut index[DICTIONARY_HEADER..])?;
4413 if checksum(&index) != page.hash {
4414 return Err(invalid("global dictionary index checksum differs"));
4415 }
4416 let offsets = index[DICTIONARY_HEADER..DICTIONARY_HEADER + offset_len].to_vec();
4417 let mut words = index[DICTIONARY_HEADER + offset_len..]
4418 .chunks_exact(8)
4419 .map(|part| u64::from_le_bytes(part.try_into().expect("eight bytes")))
4420 .collect::<Vec<_>>();
4421 let mut hashes = words.split_off(blocks);
4422 let mut rank_ends = hashes.split_off(blocks);
4423 let rank_hashes = rank_ends.split_off(rank_blocks);
4424 let ends = words;
4425 if rank_ends.windows(2).any(|pair| pair[0] >= pair[1]) {
4428 return Err(invalid("global dictionary order blocks do not rise"));
4429 }
4430 let rank_len = usize::try_from(rank_ends.last().copied().unwrap_or_default())
4431 .map_err(|_| invalid("global dictionary rank overflow"))?;
4432 let body_len = index_len
4433 .checked_add(rank_len)
4434 .ok_or_else(|| invalid("global dictionary header overflow"))?;
4435 if body_len > page.length as usize {
4436 return Err(invalid("global dictionary order exceeds its page"));
4437 }
4438 let stored_len = page.length as usize - body_len;
4441 if ends.last().copied().unwrap_or_default() as usize != stored_len
4442 || ends.windows(2).any(|pair| pair[0] > pair[1])
4443 {
4444 return Err(invalid("global dictionary blocks do not bound the payload"));
4445 }
4446 Vector::external_text(
4447 LogicalType::Varchar,
4448 Arc::new(NativeText {
4449 file,
4450 values: count,
4451 offsets,
4452 offset_bits,
4453 ranks,
4454 rank_at: page.offset + index_len as u64,
4455 rank_ends,
4456 rank_hashes,
4457 rank_blocks: (0..rank_blocks).map(|_| OnceLock::new()).collect(),
4458 code_bits: code_width(count),
4459 code_ranks: OnceLock::new(),
4460 payload: page.offset + body_len as u64,
4461 ends,
4462 hashes,
4463 blocks: (0..blocks).map(|_| OnceLock::new()).collect(),
4464 }),
4465 )
4466}
4467
4468fn decode(
4469 ty: &LogicalType,
4470 rows: usize,
4471 bytes: &[u8],
4472 global: Option<Arc<Vector>>,
4473) -> Result<Vector> {
4474 let mut cur = Cursor { bytes, at: 0 };
4475 let codec = cur.u8()?;
4476 let flag = cur.u8()?;
4477 let validity = match flag {
4478 0 => Validity::AllValid,
4479 1 => Validity::AllInvalid,
4480 2 => {
4481 let mask = cur.take(rows.div_ceil(8))?;
4482 Validity::from_iter(rows, |row| mask[row / 8] >> (row % 8) & 1 == 1)
4483 }
4484 _ => return Err(invalid("page validity tag differs")),
4485 };
4486 if codec == 1 {
4487 if ty != &LogicalType::Varchar {
4488 return Err(invalid("dictionary codec belongs to a non-string page"));
4489 }
4490 let count = cur.u32()? as usize;
4491 let payload_len = cur.u32()? as usize;
4492 let offset_bytes = cur.take(
4493 (count + 1)
4494 .checked_mul(4)
4495 .ok_or_else(|| invalid("dictionary offset count overflow"))?,
4496 )?;
4497 let offsets = offset_bytes
4498 .chunks_exact(4)
4499 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4500 .collect::<Vec<_>>();
4501 let payload = cur.take(payload_len)?.to_vec();
4502 if offsets.first() != Some(&0)
4503 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4504 || offsets.windows(2).any(|pair| pair[0] > pair[1])
4505 {
4506 return Err(invalid("dictionary offsets do not bound the payload"));
4507 }
4508 let mut strings = StringColumn::over(Buffer::from_vec(payload));
4509 for pair in offsets.windows(2) {
4510 strings.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4511 }
4512 let mut codes = Vec::with_capacity(rows);
4513 for _ in 0..rows {
4514 codes.push(cur.u32()?);
4515 }
4516 if codes.iter().any(|code| *code as usize >= count) {
4517 return Err(invalid("dictionary code is out of range"));
4518 }
4519 if cur.at != bytes.len() {
4520 return Err(invalid("dictionary page has trailing bytes"));
4521 }
4522 let dictionary = Vector::flat(LogicalType::Varchar, Data::Varlen(strings))?;
4523 return Ok(Vector::dictionary(codes, dictionary)?.with_validity(validity));
4524 }
4525 if codec == 3 || codec == 4 {
4526 let dictionary = global.ok_or_else(|| invalid("global code page has no dictionary"))?;
4527 let codes = if codec == 4 {
4528 let wide = integer::decode(&bytes[cur.at..])?;
4531 if wide.len() != rows {
4532 return Err(invalid("encoded code page holds the wrong number of rows"));
4533 }
4534 wide.into_iter()
4535 .map(|code| u32::try_from(code).map_err(|_| invalid("code is not a code")))
4536 .collect::<Result<Vec<u32>>>()?
4537 } else {
4538 let mut codes = Vec::with_capacity(rows);
4539 for _ in 0..rows {
4540 codes.push(cur.u32()?);
4541 }
4542 if cur.at != bytes.len() {
4543 return Err(invalid("global code page has trailing bytes"));
4544 }
4545 codes
4546 };
4547 let highest = codes.iter().copied().max();
4548 return Ok(Vector::stable_dictionary_validated(codes, dictionary, highest)?
4549 .with_validity(validity));
4550 }
4551 if codec == 5 {
4552 let values = integer::decode(&bytes[cur.at..])?;
4554 if values.len() != rows {
4555 return Err(invalid("cascade page holds the wrong number of rows"));
4556 }
4557 let data = narrowed(ty, values)?;
4558 return Ok(Vector::flat(ty.clone(), data)?.with_validity(validity));
4559 }
4560 if codec == 2 {
4561 let width = u32::from(cur.u8()?);
4562 let base = i128::from_le_bytes(cur.take(16)?.try_into().expect("sixteen bytes"));
4563 let count = cur.u32()? as usize;
4564 let mut words = Vec::with_capacity(count);
4565 for _ in 0..count {
4566 words.push(cur.u64()?);
4567 }
4568 if cur.at != bytes.len() {
4569 return Err(invalid("packed page has trailing bytes"));
4570 }
4571 return Ok(Vector::packed(ty.clone(), words, width, base, rows)?.with_validity(validity));
4572 }
4573 if codec != 0 {
4574 return Err(invalid("page codec is unknown"));
4575 }
4576 let data = match ty {
4577 LogicalType::TinyInt => {
4578 let values = cur.take(rows)?;
4579 Data::Int8(values.iter().map(|item| *item as i8).collect::<Vec<_>>().into())
4580 }
4581 LogicalType::UTinyInt => Data::UInt8(cur.take(rows)?.to_vec().into()),
4582 LogicalType::SmallInt => {
4583 let values =
4584 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4585 Data::Int16(
4586 values
4587 .chunks_exact(2)
4588 .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
4589 .collect::<Vec<_>>()
4590 .into(),
4591 )
4592 }
4593 LogicalType::USmallInt => {
4594 let values =
4595 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4596 Data::UInt16(
4597 values
4598 .chunks_exact(2)
4599 .map(|item| u16::from_le_bytes(item.try_into().expect("two bytes")))
4600 .collect::<Vec<_>>()
4601 .into(),
4602 )
4603 }
4604 LogicalType::UInteger => {
4605 let values =
4606 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4607 Data::UInt32(
4608 values
4609 .chunks_exact(4)
4610 .map(|item| u32::from_le_bytes(item.try_into().expect("four bytes")))
4611 .collect::<Vec<_>>()
4612 .into(),
4613 )
4614 }
4615 LogicalType::UBigInt => {
4616 let values =
4617 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4618 Data::UInt64(
4619 values
4620 .chunks_exact(8)
4621 .map(|item| u64::from_le_bytes(item.try_into().expect("eight bytes")))
4622 .collect::<Vec<_>>()
4623 .into(),
4624 )
4625 }
4626 LogicalType::Integer | LogicalType::Date => {
4627 let values =
4628 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4629 Data::Int32(
4630 values
4631 .chunks_exact(4)
4632 .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
4633 .collect::<Vec<_>>()
4634 .into(),
4635 )
4636 }
4637 LogicalType::BigInt | LogicalType::Timestamp => {
4638 let values =
4639 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4640 Data::Int64(
4641 values
4642 .chunks_exact(8)
4643 .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
4644 .collect::<Vec<_>>()
4645 .into(),
4646 )
4647 }
4648 LogicalType::Boolean => {
4649 let values = cur.take(rows)?;
4650 if values.iter().any(|value| *value > 1) {
4651 return Err(invalid("boolean page has another value"));
4652 }
4653 Data::Bool(values.iter().map(|value| *value == 1).collect::<Vec<_>>().into())
4654 }
4655 LogicalType::Varchar => {
4656 let offset_bytes = cur
4657 .take((rows + 1).checked_mul(4).ok_or_else(|| invalid("offset count overflow"))?)?;
4658 let offsets = offset_bytes
4659 .chunks_exact(4)
4660 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4661 .collect::<Vec<_>>();
4662 let payload = cur.take(bytes.len() - cur.at)?.to_vec();
4663 if offsets.first() != Some(&0)
4664 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4665 || offsets.windows(2).any(|pair| pair[0] > pair[1])
4666 {
4667 return Err(invalid("string offsets do not bound the payload"));
4668 }
4669 let mut values = StringColumn::over(Buffer::from_vec(payload));
4670 for pair in offsets.windows(2) {
4671 values.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4672 }
4673 Data::Varlen(values)
4674 }
4675 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
4676 };
4677 if cur.at != bytes.len() {
4678 return Err(invalid("page has trailing bytes"));
4679 }
4680 Ok(Vector::flat(ty.clone(), data)?.with_validity(validity))
4681}
4682
4683#[cfg(test)]
4684mod tests {
4685 use std::fs;
4686 use std::io::{Seek, SeekFrom, Write};
4687 use std::path::PathBuf;
4688 use std::time::{SystemTime, UNIX_EPOCH};
4689
4690 use rudb_common::Value;
4691 use rudb_common::bounds::Op;
4692
4693 use super::*;
4694
4695 #[test]
4696 fn checksum_matches_fixed_vectors() {
4697 assert_eq!(checksum(b""), 0xef46_db37_51d8_e999);
4698 assert_eq!(checksum(b"a"), 0xd24e_c4f1_a98c_6e5b);
4699 assert_eq!(checksum(b"abc"), 0x44bc_2cf5_ad77_0999);
4700 }
4701
4702 fn path(label: &str) -> PathBuf {
4703 let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
4704 std::env::temp_dir().join(format!("rudb-native-{label}-{}-{stamp}.rdb", std::process::id()))
4705 }
4706
4707 #[test]
4709 fn a_read_at_an_offset_ignores_where_another_thread_left_the_cursor() {
4710 const SPANS: usize = 64;
4711 const SPAN: usize = 512;
4712 let path = path("positional");
4713 let content: Vec<u8> =
4714 (0..SPANS).flat_map(|span| std::iter::repeat_n(span as u8, SPAN)).collect();
4715 fs::write(&path, &content).expect("the file is written");
4716 let file = Arc::new(File::open(&path).expect("the file opens"));
4717 std::thread::scope(|scope| {
4718 for _ in 0..8 {
4719 let file = Arc::clone(&file);
4720 scope.spawn(move || {
4721 for _ in 0..64 {
4722 for span in 0..SPANS {
4723 let mut bytes = [0_u8; SPAN];
4724 read_at(&file, (span * SPAN) as u64, &mut bytes)
4725 .expect("the span reads");
4726 assert!(
4727 bytes.iter().all(|byte| *byte == span as u8),
4728 "span {span} came back as {}",
4729 bytes[0],
4730 );
4731 }
4732 }
4733 });
4734 }
4735 });
4736 let mut past = [0_u8; SPAN];
4737 let end = (SPANS * SPAN) as u64;
4738 let error = read_at(&file, end, &mut past).expect_err("a read past the end is refused");
4739 assert!(error.message().contains("ends before its declared length"), "{error}");
4740 drop(file);
4741 let _ = fs::remove_file(&path);
4742 }
4743
4744 #[test]
4750 fn a_writer_puts_a_page_where_it_said_it_did_wherever_the_cursor_has_got_to() {
4751 let path = path("cursor");
4752 let mut writer = Writer::create(
4753 &path,
4754 "items",
4755 vec![
4756 Field::required("id", LogicalType::Integer),
4757 Field::new("text", LogicalType::Varchar),
4758 ],
4759 )
4760 .expect("new file");
4761 writer.append(&sample()).expect("first part");
4762 writer.file.seek(SeekFrom::Start(0)).expect("the cursor goes back to the header");
4763 writer.append(&sample()).expect("second part");
4764 writer.file.seek(SeekFrom::Start(1)).expect("and somewhere useless again");
4765 writer.finish().expect("commit");
4766 let reader = Reader::open(&path).expect("reopen from disk");
4767 assert_eq!(reader.table().rows(), 6);
4768 let ids = reader.read(0, &[0]).expect("the integer page reads back");
4769 assert_eq!(ids.value_at(0, 0), Value::Integer(4));
4770 assert_eq!(ids.value_at(2, 0), Value::Integer(-2));
4771 let text = reader.read(1, &[1]).expect("the text page reads back");
4772 assert_eq!(text.value_at(1, 0), Value::Null);
4773 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4774 let end = reader.table().stripes().iter().flat_map(|stripe| {
4777 stripe
4778 .pages
4779 .iter()
4780 .map(|page| page.offset + u64::from(page.length))
4781 .chain(std::iter::once(stripe.index.offset + u64::from(stripe.index.length)))
4782 });
4783 let last = end.fold(HEADER, u64::max);
4784 let directory = fs::metadata(&path).expect("the file is there").len();
4785 assert!(last <= directory, "a page runs to {last} in a file of {directory} bytes");
4786 fs::remove_file(path).expect("remove scratch file");
4787 }
4788
4789 fn dictionary_index_len(header: &[u8; DICTIONARY_HEADER]) -> u64 {
4795 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
4796 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
4797 let bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
4798 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
4799 DICTIONARY_HEADER as u64
4800 + offset_bytes(count as usize, bits) as u64
4801 + (blocks + rank_blocks) * 16
4802 }
4803
4804 fn last_rank_end(file: &File, offset: u64, header: &[u8; DICTIONARY_HEADER]) -> u64 {
4806 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
4807 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
4808 let bits = u32::from_le_bytes(header[12..16].try_into().expect("four bytes")) as usize;
4809 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
4810 let at = offset
4811 + DICTIONARY_HEADER as u64
4812 + offset_bytes(count as usize, bits) as u64
4813 + blocks * 16
4814 + (rank_blocks - 1) * 8;
4815 let mut end = [0; 8];
4816 read_at(file, at, &mut end).expect("the last rank block end");
4817 u64::from_le_bytes(end)
4818 }
4819
4820 fn sample() -> Chunk {
4821 Chunk::new(vec![
4822 Vector::from_values(
4823 LogicalType::Integer,
4824 &[Value::Integer(4), Value::Integer(9), Value::Integer(-2)],
4825 )
4826 .expect("integers"),
4827 Vector::from_values(
4828 LogicalType::Varchar,
4829 &[
4830 Value::Varchar("alpha".into()),
4831 Value::Null,
4832 Value::Varchar("long text after a slash".into()),
4833 ],
4834 )
4835 .expect("strings"),
4836 ])
4837 .expect("matching rows")
4838 }
4839
4840 fn sample_ids() -> Chunk {
4841 Chunk::new(vec![
4842 Vector::flat(LogicalType::Integer, Data::Int32(vec![7, 8, 9].into()))
4843 .expect("integers"),
4844 ])
4845 .expect("one column")
4846 }
4847
4848 #[test]
4849 fn committed_file_reopens_and_reads_only_requested_columns() {
4850 let path = path("reopen");
4851 let mut writer = Writer::create(
4852 &path,
4853 "items",
4854 vec![
4855 Field::required("id", LogicalType::Integer),
4856 Field::new("text", LogicalType::Varchar),
4857 ],
4858 )
4859 .expect("new file");
4860 writer.append(&sample()).expect("first part");
4861 writer.append(&sample()).expect("second part");
4862 writer.finish().expect("commit");
4863 let reader = Reader::open(&path).expect("reopen from disk");
4864 assert_eq!(reader.table().rows(), 6);
4865 assert_eq!(reader.table().stripes().len(), 1);
4868 assert_eq!(reader.parts(), 2);
4869 assert_eq!(reader.part_rows(0), 3);
4870 assert_eq!(reader.part_rows(1), 3);
4871 let text = reader.read(1, &[1]).expect("only text page");
4872 assert_eq!(text.width(), 1);
4873 assert_eq!(text.value_at(1, 0), Value::Null);
4874 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4875 let sparse = reader.read_sparse(1, &[1]).expect("one part without its whole page");
4876 assert_eq!(sparse.width(), 1);
4877 assert_eq!(sparse.value_at(1, 0), Value::Null);
4878 assert_eq!(sparse.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4879 assert!(!reader.skips_codes(0, 1, &[0]).expect("alpha is in the stripe"));
4880 assert!(!reader.skips_codes(0, 1, &[2]).expect("long text is in the stripe"));
4881 assert!(reader.skips_codes(0, 1, &[3]).expect("unknown code is absent"));
4882 let count = reader.read(0, &[]).expect("no page is needed for count");
4883 assert_eq!(count.len(), 3);
4884 assert!(reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }]));
4885 assert!(!reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(0) }]));
4886 let integers = reader.top_frequencies(0, 1).expect("valid integer synopsis").expect("kept");
4887 assert_eq!(
4888 integers,
4889 vec![(Value::Integer(-2), 2), (Value::Integer(4), 2), (Value::Integer(9), 2),]
4890 );
4891 let strings = reader.top_frequencies(1, 1).expect("valid string synopsis").expect("kept");
4892 assert_eq!(strings.len(), 3);
4893 assert!(strings.contains(&(Value::Null, 2)));
4894 assert!(strings.contains(&(Value::Varchar("alpha".into()), 2)));
4895 assert!(strings.contains(&(Value::Varchar("long text after a slash".into()), 2)));
4896 fs::remove_file(path).expect("remove scratch file");
4897 }
4898
4899 #[test]
4907 fn runs_handed_over_out_of_order_still_read_back_in_source_order() {
4908 let path = path("interleaved-runs");
4909 let mut writer =
4910 Writer::create(&path, "interleaved", vec![Field::new("v", LogicalType::BigInt)])
4911 .expect("new file");
4912 for morsel in [2_u64, 0, 3, 1] {
4913 let parts = (0..4_u64)
4914 .map(|chunk| {
4915 let first = i64::try_from(morsel * 32 + chunk * 8).expect("small");
4916 let values =
4917 (0..8_i64).map(|row| Value::BigInt(first + row)).collect::<Vec<_>>();
4918 let column =
4919 Vector::from_values(LogicalType::BigInt, &values).expect("a column");
4920 ((morsel, chunk), Chunk::new(vec![column]).expect("one column"))
4921 })
4922 .collect::<Vec<_>>();
4923 writer.append_stripe(parts).expect("a stripe");
4924 }
4925 writer.finish().expect("commit");
4926
4927 let reader = Reader::open(&path).expect("valid directory");
4928 assert_eq!(reader.table().stripes().len(), 4, "a run is a stripe of its own");
4929 assert_eq!(reader.table().rows(), 128);
4930 for part in 0..16_usize {
4931 let read = reader.read(part, &[0]).expect("a part back");
4932 for row in 0..8_usize {
4933 let want = i64::try_from(part * 8 + row).expect("small");
4934 assert_eq!(read.value_at(row, 0), Value::BigInt(want), "part {part} row {row}");
4935 }
4936 }
4937 fs::remove_file(path).expect("remove scratch file");
4938 }
4939
4940 #[test]
4943 fn runs_that_overlap_each_other_are_refused_at_commit() {
4944 let path = path("overlapping-runs");
4945 let mut writer =
4946 Writer::create(&path, "overlapping", vec![Field::new("v", LogicalType::BigInt)])
4947 .expect("new file");
4948 let one = |order: (u64, u64)| {
4949 let column =
4950 Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)]).expect("a column");
4951 (order, Chunk::new(vec![column]).expect("one column"))
4952 };
4953 writer.append_stripe(vec![one((0, 0)), one((0, 2))]).expect("a stripe");
4956 writer.append_stripe(vec![one((0, 1))]).expect("a stripe");
4957 let error = writer.finish().expect_err("the runs overlap");
4958 assert!(error.message().contains("source order"), "{error}");
4959 fs::remove_file(path).expect("remove scratch file");
4960 }
4961
4962 #[test]
4965 fn a_run_longer_than_a_stripe_is_refused() {
4966 let path = path("overlong-run");
4967 let mut writer =
4968 Writer::create(&path, "overlong", vec![Field::new("v", LogicalType::BigInt)])
4969 .expect("new file");
4970 let parts = (0..=STRIPE_PARTS)
4971 .map(|at| {
4972 let column = Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)])
4973 .expect("a column");
4974 let chunk = Chunk::new(vec![column]).expect("one column");
4975 ((0, u64::try_from(at).expect("small")), chunk)
4976 })
4977 .collect::<Vec<_>>();
4978 let error = writer.append_stripe(parts).expect_err("one part too many");
4979 assert!(error.message().contains("more parts than it holds"), "{error}");
4980 fs::remove_file(path).expect("remove scratch file");
4981 }
4982
4983 #[test]
4989 fn parts_past_the_stripe_bound_start_a_new_stripe() {
4990 let path = path("stripe-bound");
4991 let mut writer = Writer::create(
4992 &path,
4993 "items",
4994 vec![
4995 Field::required("id", LogicalType::Integer),
4996 Field::new("text", LogicalType::Varchar),
4997 ],
4998 )
4999 .expect("new file");
5000 let parts = STRIPE_PARTS * 2 + 3;
5001 for part in 0..parts {
5002 let id = part as i32;
5003 let chunk = Chunk::new(vec![
5004 Vector::from_values(
5005 LogicalType::Integer,
5006 &[Value::Integer(id), Value::Integer(-id)],
5007 )
5008 .expect("integers"),
5009 Vector::from_values(
5010 LogicalType::Varchar,
5011 &[Value::Varchar(format!("value {part}")), Value::Null],
5012 )
5013 .expect("strings"),
5014 ])
5015 .expect("matching rows");
5016 writer.append(&chunk).expect("one part");
5017 }
5018 writer.finish().expect("commit");
5019
5020 let reader = Reader::open(&path).expect("reopen from disk");
5021 assert_eq!(reader.parts(), parts);
5022 assert_eq!(reader.table().rows(), parts * 2);
5023 assert_eq!(reader.table().stripes().len(), parts.div_ceil(STRIPE_PARTS));
5024 assert_eq!(reader.table().stripes()[0].parts(), STRIPE_PARTS);
5025 assert_eq!(reader.table().stripes()[0].rows(), STRIPE_PARTS * 2);
5026 assert_eq!(reader.table().stripes()[2].parts(), 3);
5027 for part in (0..parts).rev() {
5030 let dense = reader.read(part, &[0, 1]).expect("a whole page read");
5031 let sparse = reader.read_sparse(part, &[0, 1]).expect("one part read");
5032 for chunk in [&dense, &sparse] {
5033 assert_eq!(chunk.len(), 2, "part {part} has its own row count");
5034 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5035 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
5036 assert_eq!(chunk.value_at(0, 1), Value::Varchar(format!("value {part}")));
5037 assert_eq!(chunk.value_at(1, 1), Value::Null);
5038 }
5039 }
5040 let above = [Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }];
5043 assert!(reader.skips(0, &above), "the first stripe stops at 63");
5044 assert!(!reader.skips(STRIPE_PARTS * 2, &above), "the third stripe reaches 130");
5045 fs::remove_file(path).expect("remove scratch file");
5046 }
5047
5048 fn scattered(n: i64) -> i64 {
5050 n.wrapping_mul(-7_046_029_254_386_353_131)
5051 }
5052
5053 #[test]
5059 fn a_part_is_skipped_when_its_sieve_does_not_hold_the_constant() {
5060 let path = path("sieve-skip");
5061 let mut writer =
5062 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
5063 .expect("new file");
5064 let parts = STRIPE_PARTS + 3;
5065 let per_part = 128;
5069 for part in 0..parts {
5070 let held: Vec<Value> = (0..per_part)
5071 .map(|row| Value::BigInt(scattered((part * per_part + row) as i64)))
5072 .collect();
5073 let chunk =
5074 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
5075 .expect("one column");
5076 writer.append(&chunk).expect("one part");
5077 }
5078 writer.finish().expect("commit");
5079
5080 let reader = Reader::open(&path).expect("reopen from disk");
5081 let probe = |value: i64| Probe {
5082 column: 0,
5083 op: Op::Equal,
5084 value: Bound::Int(i128::from(scattered(value))),
5085 };
5086 for wanted in [0_i64, (per_part + 1) as i64, (parts * per_part - 1) as i64] {
5087 let tests = [probe(wanted)];
5088 let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &tests)).collect();
5089 let home = wanted as usize / per_part;
5090 assert!(kept.contains(&home), "the part holding {wanted} is read");
5091 assert!(kept.len() <= 2, "{wanted} keeps {kept:?}, which is more than one stray part");
5095 }
5096 let absent = [probe((parts * per_part) as i64 + 1)];
5097 let kept = (0..parts).filter(|&part| !reader.skips(part, &absent)).count();
5098 assert!(kept <= 1, "{kept} parts of {parts} kept a value no part holds");
5099 let tests = [probe(0)];
5102 assert!(
5103 reader.table().stripes().iter().all(|stripe| !stripe.zone.skips(&tests)),
5104 "the bounds rule out no stripe at all"
5105 );
5106 fs::remove_file(path).expect("remove scratch file");
5107 }
5108
5109 #[test]
5119 fn a_sieve_larger_than_the_part_it_indexes_is_not_written() {
5120 let path = path("sieve-pays");
5121 let fields = vec![
5122 Field::required("spread", LogicalType::BigInt),
5123 Field::required("repeated", LogicalType::BigInt),
5124 ];
5125 let mut writer = Writer::create(&path, "hits", fields).expect("new file");
5126 let parts = 3;
5127 let per_part = 1024;
5128 for part in 0..parts {
5129 let base = (part * per_part) as i64;
5130 let spread: Vec<Value> =
5131 (0..per_part).map(|row| Value::BigInt(scattered(base + row as i64))).collect();
5132 let repeated: Vec<Value> =
5133 (0..per_part).map(|row| Value::BigInt(scattered((row / 256) as i64))).collect();
5134 let chunk = Chunk::new(vec![
5135 Vector::from_values(LogicalType::BigInt, &spread).expect("numbers"),
5136 Vector::from_values(LogicalType::BigInt, &repeated).expect("numbers"),
5137 ])
5138 .expect("two columns");
5139 writer.append(&chunk).expect("one part");
5140 }
5141 writer.finish().expect("commit");
5142
5143 let reader = Reader::open(&path).expect("reopen from disk");
5144 let layout = reader.layout();
5145 let spread = &layout.columns[0];
5146 let repeated = &layout.columns[1];
5147 assert!(spread.sieves > 0, "a column whose parts are worth a filter keeps one");
5148 assert_eq!(
5149 repeated.sieves, 0,
5150 "a column whose filter costs more than its parts keeps none"
5151 );
5152 for column in &layout.columns {
5155 assert!(
5156 column.sieves < column.pages,
5157 "{} spends {} on sieves over {} of data",
5158 column.name,
5159 column.sieves,
5160 column.pages
5161 );
5162 }
5163 let absent = [Probe {
5165 column: 0,
5166 op: Op::Equal,
5167 value: Bound::Int(i128::from(scattered((parts * per_part) as i64 + 1))),
5168 }];
5169 assert!((0..parts).all(|part| reader.skips(part, &absent)), "no part holds it");
5170 fs::remove_file(path).expect("remove scratch file");
5171 }
5172
5173 #[test]
5179 fn a_damaged_sieve_page_is_read_through_rather_than_refused() {
5180 let path = path("sieve-damaged");
5181 let mut writer =
5182 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
5183 .expect("new file");
5184 let rows = 128;
5185 let held: Vec<Value> = (0..rows).map(|row| Value::BigInt(scattered(row))).collect();
5186 let chunk =
5187 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
5188 .expect("one column");
5189 writer.append(&chunk).expect("one part");
5190 writer.finish().expect("commit");
5191
5192 let page =
5193 Reader::open(&path).expect("reopen").table.stripes[0].sieves[0].expect("a sieve page");
5194 let mut file = OpenOptions::new().write(true).open(&path).expect("open the sieve page");
5195 file.seek(SeekFrom::Start(page.offset + u64::from(page.length) - 1)).expect("seek");
5196 file.write_all(&[0xff]).expect("damage one byte");
5197 drop(file);
5198
5199 let reader = Reader::open(&path).expect("reopen the damaged file");
5200 let absent =
5201 [Probe { column: 0, op: Op::Equal, value: Bound::Int(i128::from(scattered(99))) }];
5202 assert!(!reader.skips(0, &absent), "a sieve that cannot be read skips nothing");
5203 assert_eq!(
5204 reader.read(0, &[0]).expect("the rows are untouched").len(),
5205 usize::try_from(rows).expect("a small count")
5206 );
5207 fs::remove_file(path).expect("remove scratch file");
5208 }
5209
5210 #[test]
5221 fn workers_that_want_the_same_stripe_read_it_once() {
5222 let path = path("single-flight");
5223 let mut writer =
5224 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5225 .expect("new file");
5226 for part in 0..STRIPE_PARTS {
5227 let id = part as i32;
5228 let chunk = Chunk::new(vec![
5229 Vector::from_values(
5230 LogicalType::Integer,
5231 &[Value::Integer(id), Value::Integer(-id)],
5232 )
5233 .expect("integers"),
5234 ])
5235 .expect("matching rows");
5236 writer.append(&chunk).expect("one part");
5237 }
5238 writer.finish().expect("commit");
5239
5240 let reader = Reader::open(&path).expect("reopen from disk");
5241 assert_eq!(reader.table().stripes().len(), 1, "one stripe is the point of the test");
5242 let barrier = std::sync::Barrier::new(8);
5243 std::thread::scope(|scope| {
5244 for worker in 0..8 {
5245 let reader = &reader;
5246 let barrier = &barrier;
5247 scope.spawn(move || {
5248 barrier.wait();
5249 for part in (worker..STRIPE_PARTS).step_by(8) {
5250 let chunk = reader.read(part, &[0]).expect("a whole page read");
5251 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5252 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
5253 }
5254 });
5255 }
5256 });
5257 assert_eq!(reader.pages.load(Atomic::Relaxed), 1, "one stripe, one page read, whoever won");
5258 fs::remove_file(path).expect("remove scratch file");
5259 }
5260
5261 #[test]
5274 fn opening_costs_the_same_over_a_thousand_times_the_rows() {
5275 let opened = |label: &str, rows_per_part: i32| {
5276 let path = path(label);
5277 let mut writer =
5278 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5279 .expect("new file");
5280 for part in 0..STRIPE_PARTS * 3 {
5281 let values = (0..rows_per_part)
5285 .map(|row| {
5286 Value::Integer((part as i32 * rows_per_part + row).wrapping_mul(2_654_435))
5287 })
5288 .collect::<Vec<_>>();
5289 let chunk = Chunk::new(vec![
5290 Vector::from_values(LogicalType::Integer, &values).expect("integers"),
5291 ])
5292 .expect("matching rows");
5293 writer.append(&chunk).expect("one part");
5294 }
5295 writer.finish().expect("commit");
5296 let reader = Reader::open(&path).expect("reopen from disk");
5297 let size = fs::metadata(&path).expect("the file is there").len();
5298 let out = (reader.reads(), reader.table().stripes().len(), size);
5299 fs::remove_file(path).expect("remove scratch file");
5300 out
5301 };
5302
5303 let (thin, thin_stripes, thin_size) = opened("open-thin", 1);
5304 let (fat, fat_stripes, fat_size) = opened("open-fat", 1000);
5305 assert_eq!(
5306 thin_stripes, fat_stripes,
5307 "the same stripe count is what makes this a fair ask"
5308 );
5309 assert!(
5310 fat_size > thin_size * 50,
5311 "the fat file has to actually be larger, and it is {fat_size} against {thin_size}"
5312 );
5313
5314 assert_eq!(thin.opening.reads, fat.opening.reads, "the same reads either way");
5315 assert_eq!(thin.pages, 0, "opening read a page");
5316 assert_eq!(fat.pages, 0, "opening read a page");
5317 assert_eq!(thin.indexes, 0, "opening read an index");
5318 assert_eq!(fat.indexes, 0, "opening read an index");
5319 assert!(
5322 fat.opening.bytes < thin.opening.bytes * 2,
5323 "opening the thin file read {} bytes and the fat one read {}",
5324 thin.opening.bytes,
5325 fat.opening.bytes
5326 );
5327 }
5328
5329 #[test]
5337 fn two_opens_of_one_file_cost_the_same_and_the_second_is_not_cheaper() {
5338 let path = path("open-twice");
5339 let mut writer =
5340 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5341 .expect("new file");
5342 for part in 0..STRIPE_PARTS * 3 {
5343 let chunk = Chunk::new(vec![
5344 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
5345 .expect("integers"),
5346 ])
5347 .expect("matching rows");
5348 writer.append(&chunk).expect("one part");
5349 }
5350 writer.finish().expect("commit");
5351
5352 let first = Reader::open(&path).expect("open");
5353 for part in 0..first.parts() {
5356 first.read(part, &[0]).expect("a part");
5357 }
5358 assert!(first.reads().pages > 0, "the scan has to have read something");
5359 let second = Reader::open(&path).expect("open again");
5360
5361 assert_eq!(first.reads().opening, second.reads().opening);
5362 assert_eq!(
5363 second.reads().pages,
5364 0,
5365 "the second open read a page off the back of the first"
5366 );
5367 assert_eq!(second.reads().indexes, 0, "the second open read an index it inherited");
5368 fs::remove_file(path).expect("remove scratch file");
5369 }
5370
5371 #[test]
5379 fn an_index_is_read_once_per_stripe_however_often_the_page_is_evicted() {
5380 let path = path("index-cache");
5381 let mut writer =
5382 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5383 .expect("new file");
5384 let parts = STRIPE_PARTS * (CACHED_STRIPES_PER_COLUMN + 2);
5385 for part in 0..parts {
5386 let id = part as i32;
5387 let chunk = Chunk::new(vec![
5388 Vector::from_values(LogicalType::Integer, &[Value::Integer(id)]).expect("integers"),
5389 ])
5390 .expect("matching rows");
5391 writer.append(&chunk).expect("one part");
5392 }
5393 writer.finish().expect("commit");
5394
5395 let reader = Reader::open(&path).expect("reopen from disk");
5396 let stripes = reader.table().stripes().len();
5397 assert!(stripes > CACHED_STRIPES_PER_COLUMN, "the page cache has to be too small for this");
5398 for _ in 0..2 {
5400 for part in 0..parts {
5401 let chunk = reader.read(part, &[0]).expect("a part");
5402 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5403 }
5404 }
5405 assert_eq!(reader.indexes.load(Atomic::Relaxed), stripes, "one index read per stripe");
5406 assert!(
5407 reader.pages.load(Atomic::Relaxed) > stripes,
5408 "the pages are the ones that get read again, which is what makes the index count mean \
5409 something"
5410 );
5411 fs::remove_file(path).expect("remove scratch file");
5412 }
5413
5414 #[test]
5423 fn a_worker_per_stripe_reads_its_page_once_when_the_cache_was_told_to_expect_it() {
5424 let workers = CACHED_STRIPES_PER_COLUMN + 4;
5425 let path = path("stripe-per-worker");
5426 let mut writer =
5427 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5428 .expect("new file");
5429 for part in 0..STRIPE_PARTS * workers {
5430 let chunk = Chunk::new(vec![
5431 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
5432 .expect("integers"),
5433 ])
5434 .expect("matching rows");
5435 writer.append(&chunk).expect("one part");
5436 }
5437 writer.finish().expect("commit");
5438
5439 let read = |told: bool| {
5440 let reader = Reader::open(&path).expect("reopen from disk");
5441 assert_eq!(reader.table().stripes().len(), workers, "a stripe per worker");
5442 if told {
5443 reader.keep_stripes(workers);
5444 }
5445 let barrier = std::sync::Barrier::new(workers);
5446 std::thread::scope(|scope| {
5447 for (worker, run) in reader.stripe_parts().into_iter().enumerate() {
5448 let reader = &reader;
5449 let barrier = &barrier;
5450 scope.spawn(move || {
5451 for part in run {
5452 barrier.wait();
5453 let chunk = reader.read(part, &[0]).expect("a part of my own stripe");
5454 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5455 }
5456 assert!(worker < workers);
5457 });
5458 }
5459 });
5460 reader.pages.load(Atomic::Relaxed)
5461 };
5462
5463 assert_eq!(read(true), workers, "one page read per stripe and no more");
5464 assert!(read(false) > workers, "a cache that small is read again on every part");
5465 fs::remove_file(path).expect("remove scratch file");
5466 }
5467
5468 #[test]
5473 fn a_damaged_index_page_is_an_error() {
5474 let path = path("damaged-index");
5475 let mut writer =
5476 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5477 .expect("new file");
5478 writer.append(&sample_ids()).expect("first part");
5479 writer.append(&sample_ids()).expect("second part");
5480 writer.finish().expect("commit");
5481
5482 let reader = Reader::open(&path).expect("valid directory");
5483 let index = reader.table.stripes[0].index;
5484 let mut byte = [0; 1];
5485 read_at(&reader.file, index.offset, &mut byte).expect("the first part length");
5486 let mut file = OpenOptions::new().write(true).open(&path).expect("open index page");
5487 file.seek(SeekFrom::Start(index.offset)).expect("index start");
5488 file.write_all(&[!byte[0]]).expect("damage the first part length");
5489 let error = reader.read(1, &[0]).expect_err("a damaged index must not be used");
5490 assert!(error.message().contains("index page section checksum differs"), "{error}");
5491 fs::remove_file(path).expect("remove scratch file");
5492 }
5493
5494 #[test]
5501 fn every_integer_width_round_trips_through_a_page() {
5502 let path = path("integer-widths");
5503 let columns = [
5504 (LogicalType::TinyInt, vec![Value::TinyInt(i8::MIN), Value::TinyInt(i8::MAX)]),
5505 (LogicalType::UTinyInt, vec![Value::UTinyInt(0), Value::UTinyInt(u8::MAX)]),
5506 (LogicalType::SmallInt, vec![Value::SmallInt(i16::MIN), Value::SmallInt(i16::MAX)]),
5507 (LogicalType::USmallInt, vec![Value::USmallInt(0), Value::USmallInt(u16::MAX)]),
5508 (LogicalType::Integer, vec![Value::Integer(i32::MIN), Value::Integer(i32::MAX)]),
5509 (LogicalType::UInteger, vec![Value::UInteger(0), Value::UInteger(u32::MAX)]),
5510 (LogicalType::BigInt, vec![Value::BigInt(i64::MIN), Value::BigInt(i64::MAX)]),
5511 (LogicalType::UBigInt, vec![Value::UBigInt(0), Value::UBigInt(u64::MAX)]),
5512 ];
5513 let fields = columns
5514 .iter()
5515 .enumerate()
5516 .map(|(at, (ty, _))| Field::required(format!("c{at}"), ty.clone()))
5517 .collect::<Vec<_>>();
5518 let vectors = columns
5519 .iter()
5520 .map(|(ty, values)| Vector::from_values(ty.clone(), values).expect("a vector"))
5521 .collect::<Vec<_>>();
5522 let mut writer = Writer::create(&path, "widths", fields).expect("new file");
5523 writer.append(&Chunk::new(vectors).expect("matching rows")).expect("one stripe");
5524 writer.finish().expect("commit");
5525
5526 let reader = Reader::open(&path).expect("reopen from disk");
5527 let wanted = (0..columns.len()).collect::<Vec<_>>();
5528 let read = reader.read(0, &wanted).expect("every column");
5529 assert_eq!(read.len(), 2);
5530 for (at, (ty, values)) in columns.iter().enumerate() {
5532 assert_eq!(read.value_at(0, at), values[0], "the low end of {ty}");
5533 assert_eq!(read.value_at(1, at), values[1], "the high end of {ty}");
5534 }
5535 fs::remove_file(path).expect("remove scratch file");
5536 }
5537
5538 #[test]
5539 fn numeric_frequency_candidates_keep_bounded_row_ordinals() {
5540 let path = path("frequency-ordinals");
5541 let mut writer =
5542 Writer::create(&path, "items", vec![Field::required("id", LogicalType::BigInt)])
5543 .expect("new file");
5544 let mut values = Vec::new();
5545 for leader in 0..10_i64 {
5546 values.extend(std::iter::repeat_n(leader, 100));
5547 }
5548 values.extend(1_000_i64..41_000);
5549 for part in values.chunks(1_024) {
5550 let vector = Vector::flat(LogicalType::BigInt, Data::Int64(part.to_vec().into()))
5551 .expect("big integers");
5552 writer.append(&Chunk::new(vec![vector]).expect("one column")).expect("one stripe");
5553 }
5554 writer.finish().expect("commit");
5555
5556 let reader = Reader::open(&path).expect("reopen from disk");
5557 let occurrences =
5558 reader.frequency_occurrences(0).expect("valid metadata").expect("bounded ordinals");
5559 assert!(occurrences.omitted_max < 100);
5560 assert!(occurrences.ordinals.len() <= FREQUENCY_ORDINALS);
5561 assert!(occurrences.ordinals.windows(2).all(|pair| pair[0] < pair[1]));
5562 assert_eq!(&occurrences.ordinals[..1_000], &(0_u64..1_000).collect::<Vec<_>>());
5563 fs::remove_file(path).expect("remove scratch file");
5564 }
5565
5566 #[test]
5572 fn a_file_from_another_format_says_which_format_it_is() {
5573 let older = path("older-format");
5574 let mut writer =
5575 Writer::create(&older, "items", vec![Field::new("id", LogicalType::Integer)])
5576 .expect("new file");
5577 let chunk = Chunk::new(vec![
5578 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
5579 .expect("integers"),
5580 ])
5581 .expect("chunk");
5582 writer.append(&chunk).expect("page written");
5583 writer.finish().expect("commit");
5584
5585 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
5586 file.seek(SeekFrom::Start(8)).expect("the version follows the magic");
5587 file.write_all(&(FORMAT - 1).to_le_bytes()).expect("write an older version");
5588 drop(file);
5589 let complaint = Reader::open(&older).expect_err("an older format is refused").to_string();
5590 assert!(complaint.contains(&format!("format {}", FORMAT - 1)), "{complaint}");
5591 assert!(complaint.contains(&format!("format {FORMAT}")), "{complaint}");
5592
5593 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
5594 file.seek(SeekFrom::Start(0)).expect("the magic is first");
5595 file.write_all(b"NOTRUDB!").expect("write another engine's magic");
5596 drop(file);
5597 let complaint = Reader::open(&older).expect_err("a foreign file is refused").to_string();
5598 assert!(complaint.contains("magic"), "{complaint}");
5599 assert!(!complaint.contains("format"), "a version has nothing to do with it: {complaint}");
5600 fs::remove_file(older).expect("remove scratch file");
5601 }
5602
5603 #[test]
5604 fn an_unfinished_or_damaged_file_does_not_answer_with_partial_rows() {
5605 let unfinished = path("unfinished");
5606 let mut writer =
5607 Writer::create(&unfinished, "items", vec![Field::new("id", LogicalType::Integer)])
5608 .expect("new file");
5609 let chunk = Chunk::new(vec![
5610 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
5611 .expect("integers"),
5612 ])
5613 .expect("chunk");
5614 writer.append(&chunk).expect("page written");
5615 drop(writer);
5616 assert!(Reader::open(&unfinished).is_err(), "no directory was committed");
5617 fs::remove_file(unfinished).expect("remove scratch file");
5618
5619 let damaged = path("damaged");
5620 let mut writer =
5621 Writer::create(&damaged, "items", vec![Field::new("id", LogicalType::Integer)])
5622 .expect("new file");
5623 writer.append(&chunk).expect("page written");
5624 writer.finish().expect("commit");
5625 let reader = Reader::open(&damaged).expect("valid directory");
5626 let mut file =
5627 OpenOptions::new().write(true).open(&damaged).expect("open for a damaged page");
5628 file.seek(SeekFrom::Start(HEADER + 1)).expect("inside first page");
5629 file.write_all(&[255]).expect("damage one byte");
5630 assert!(reader.read(0, &[0]).is_err(), "page checksum rejects corruption");
5631 fs::remove_file(damaged).expect("remove scratch file");
5632 }
5633
5634 #[test]
5635 fn damaged_lazy_dictionary_payload_is_an_error() {
5636 let path = path("damaged-dictionary");
5637 let mut writer = Writer::create(
5638 &path,
5639 "items",
5640 vec![
5641 Field::required("id", LogicalType::Integer),
5642 Field::new("text", LogicalType::Varchar),
5643 ],
5644 )
5645 .expect("new file");
5646 writer.append(&sample()).expect("stripe written");
5647 writer.finish().expect("commit");
5648
5649 let reader = Reader::open(&path).expect("valid directory");
5650 let dictionary = reader.table.dictionaries[1].expect("string dictionary page");
5651 let mut header = [0; DICTIONARY_HEADER];
5654 read_at(&reader.file, dictionary.offset, &mut header).expect("dictionary header");
5655 let index_len = dictionary_index_len(&header);
5656 let rank_len = last_rank_end(&reader.file, dictionary.offset, &header);
5657 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5658 file.seek(SeekFrom::Start(dictionary.offset + index_len + rank_len))
5659 .expect("inside dictionary payload");
5660 file.write_all(&[255]).expect("damage dictionary payload");
5661
5662 let chunk = reader.read(0, &[1]).expect("code page and dictionary index remain valid");
5663 let error =
5664 chunk.validate_external().expect_err("payload corruption must reach the caller");
5665 assert!(error.message().contains("payload checksum differs"), "{error}");
5666 fs::remove_file(path).expect("remove scratch file");
5667 }
5668
5669 #[test]
5676 fn a_dictionary_over_many_blocks_checks_every_block_of_it() {
5677 let path = path("dictionary-blocks");
5678 let value =
5679 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5680 let parts = 30;
5681 let per_part = 1000;
5682 let mut writer =
5683 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5684 .expect("new file");
5685 for part in 0..parts {
5686 let values = (0..per_part)
5687 .map(|row| Value::Varchar(value(part * per_part + row)))
5688 .collect::<Vec<_>>();
5689 let chunk = Chunk::new(vec![
5690 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5691 ])
5692 .expect("matching rows");
5693 writer.append(&chunk).expect("a part");
5694 }
5695 writer.finish().expect("commit");
5696
5697 let reader = Reader::open(&path).expect("reopen from disk");
5698 let dictionary = reader.table.dictionaries[0].expect("string dictionary page");
5699 assert!(
5700 parts * per_part > TEXT_PAYLOAD_VALUES * 4,
5701 "the dictionary has to be several blocks for this to be testing anything"
5702 );
5703 for part in [0, parts - 1] {
5704 let chunk = reader.read(part, &[0]).expect("a part");
5705 chunk.validate_external().expect("every payload block checks out");
5706 assert_eq!(chunk.value_at(0, 0), Value::Varchar(value(part * per_part)));
5707 }
5708
5709 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5710 file.seek(SeekFrom::Start(dictionary.offset + u64::from(dictionary.length) - 4))
5711 .expect("the last bytes of the page are payload");
5712 file.write_all(&[255]).expect("damage the last payload block");
5713 let reader = Reader::open(&path).expect("the directory and the index are untouched");
5714 let chunk = reader.read(parts - 1, &[0]).expect("the code page remains valid");
5715 let error = chunk.validate_external().expect_err("the damage must reach the caller");
5716 assert!(error.message().contains("payload checksum differs"), "{error}");
5717 fs::remove_file(path).expect("remove scratch file");
5718 }
5719
5720 #[test]
5730 fn values_of_different_lengths_read_back_out_of_packed_offsets() {
5731 let path = path("dictionary-offsets");
5732 let value = |row: usize| {
5733 if row % 511 == 3 { String::new() } else { "x".repeat(row % 97) + &format!("{row:05}") }
5734 };
5735 let rows = 5_000;
5736 let mut writer =
5737 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5738 .expect("new file");
5739 let values = (0..rows).map(|row| Value::Varchar(value(row))).collect::<Vec<_>>();
5740 for part in values.chunks(1_000) {
5741 let chunk =
5742 Chunk::new(vec![Vector::from_values(LogicalType::Varchar, part).expect("strings")])
5743 .expect("matching rows");
5744 writer.append(&chunk).expect("a part");
5745 }
5746 writer.finish().expect("commit");
5747
5748 let reader = Reader::open(&path).expect("reopen from disk");
5749 assert!(
5750 rows > TEXT_PAYLOAD_VALUES * 4,
5751 "the dictionary has to be several blocks for this to be testing anything"
5752 );
5753 for part in 0..rows / 1_000 {
5754 let chunk = reader.read(part, &[0]).expect("a part");
5755 for row in 0..1_000 {
5756 let row = part * 1_000 + row;
5757 assert_eq!(
5758 chunk.value_at(row % 1_000, 0),
5759 Value::Varchar(value(row)),
5760 "value {row}"
5761 );
5762 }
5763 }
5764 fs::remove_file(path).expect("remove scratch file");
5765 }
5766
5767 #[test]
5779 fn a_global_dictionary_is_opened_once_however_many_workers_ask_at_once() {
5780 let path = path("dictionary-once");
5781 let parts = 8;
5782 let per_part = 500;
5783 let value =
5784 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5785 let mut writer =
5786 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5787 .expect("new file");
5788 for part in 0..parts {
5789 let values = (0..per_part)
5790 .map(|row| Value::Varchar(value(part * per_part + row)))
5791 .collect::<Vec<_>>();
5792 let chunk = Chunk::new(vec![
5793 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5794 ])
5795 .expect("matching rows");
5796 writer.append(&chunk).expect("a part");
5797 }
5798 writer.finish().expect("commit");
5799
5800 let reader = Reader::open(&path).expect("reopen from disk");
5801 assert!(reader.table.dictionaries[0].is_some(), "the column has to have one to share");
5802 assert_eq!(reader.reads().dictionaries, 0, "opening the file does not open a dictionary");
5803
5804 let workers = 16;
5805 let gate = std::sync::Barrier::new(workers);
5806 std::thread::scope(|scope| {
5807 for worker in 0..workers {
5808 let reader = reader.clone();
5809 let gate = &gate;
5810 scope.spawn(move || {
5811 gate.wait();
5812 let chunk = reader.read(worker % parts, &[0]).expect("a part");
5813 assert_eq!(
5814 chunk.value_at(0, 0),
5815 Value::Varchar(value((worker % parts) * per_part))
5816 );
5817 });
5818 }
5819 });
5820
5821 assert_eq!(reader.reads().dictionaries, 1, "sixteen workers, one dictionary, one open");
5822 fs::remove_file(path).expect("remove scratch file");
5823 }
5824
5825 #[test]
5830 fn a_damaged_sorted_order_is_an_error() {
5831 let path = path("damaged-order");
5832 let mut writer = Writer::create(
5833 &path,
5834 "items",
5835 vec![
5836 Field::required("id", LogicalType::Integer),
5837 Field::new("text", LogicalType::Varchar),
5838 ],
5839 )
5840 .expect("new file");
5841 writer.append(&sample()).expect("stripe written");
5842 writer.finish().expect("commit");
5843
5844 let reader = Reader::open(&path).expect("valid directory");
5845 let page = reader.table.dictionaries[1].expect("string dictionary page");
5846 let mut header = [0; DICTIONARY_HEADER];
5847 read_at(&reader.file, page.offset, &mut header).expect("dictionary header");
5848 let index_len = dictionary_index_len(&header);
5849 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5850 file.seek(SeekFrom::Start(page.offset + index_len)).expect("the first head");
5851 file.write_all(&[255]).expect("damage the order");
5852
5853 let dictionary = reader.dictionary(1).expect("read").expect("a string column has one");
5854 let error = dictionary.compare_rank(0, b"anything").expect_err("a damaged order is caught");
5855 assert!(error.message().contains("rank checksum differs"), "{error}");
5856 fs::remove_file(path).expect("remove scratch file");
5857 }
5858
5859 #[test]
5863 fn a_global_dictionary_carries_the_sorted_order_of_its_values() {
5864 let spellings = ["overlong1z", "b", "", "overlong1a", "overlong", "ab", "a", "overlong1"];
5867 let path = path("dictionary-order");
5868 let mut writer =
5869 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
5870 .expect("new file");
5871 writer
5872 .append(
5873 &Chunk::new(vec![
5874 Vector::from_values(
5875 LogicalType::Varchar,
5876 &spellings.map(|text| Value::Varchar(text.into())),
5877 )
5878 .expect("strings"),
5879 ])
5880 .expect("one column"),
5881 )
5882 .expect("stripe written");
5883 writer.finish().expect("commit");
5884
5885 let reader = Reader::open(&path).expect("valid directory");
5886 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
5887 let count = dictionary.ranks().expect("a v10 file stores one");
5888 assert_eq!(count, spellings.len(), "every distinct value has a rank");
5889 let order = (0..count)
5890 .map(|rank| dictionary.code_at_rank(rank).expect("a code"))
5891 .collect::<Vec<_>>();
5892 let mut seen = order.clone();
5893 seen.sort_unstable();
5894 assert_eq!(seen, (0..spellings.len() as u32).collect::<Vec<_>>(), "a permutation of codes");
5895
5896 let ranked = order
5897 .iter()
5898 .map(|&code| {
5899 dictionary.try_bytes_at(code as usize).expect("read").expect("a value").to_vec()
5900 })
5901 .collect::<Vec<_>>();
5902 let mut expected = spellings.map(|text| text.as_bytes().to_vec()).to_vec();
5903 expected.sort();
5904 assert_eq!(ranked, expected, "rank order is value order");
5905
5906 for (rank, value) in expected.iter().enumerate() {
5909 assert_eq!(
5910 dictionary.compare_rank(rank, value).expect("compare"),
5911 Ordering::Equal,
5912 "rank {rank} is its own value"
5913 );
5914 if rank > 0 {
5915 assert_eq!(
5916 dictionary.compare_rank(rank - 1, value).expect("compare"),
5917 Ordering::Less,
5918 "rank {rank} follows the one before it"
5919 );
5920 }
5921 }
5922 fs::remove_file(path).expect("remove scratch file");
5923 }
5924
5925 #[test]
5926 fn damaged_membership_cannot_skip_a_string_page() {
5927 let path = path("damaged-membership");
5928 let mut writer = Writer::create(
5929 &path,
5930 "items",
5931 vec![
5932 Field::required("id", LogicalType::Integer),
5933 Field::new("text", LogicalType::Varchar),
5934 ],
5935 )
5936 .expect("new file");
5937 writer.append(&sample()).expect("stripe written");
5938 writer.finish().expect("commit");
5939
5940 let reader = Reader::open(&path).expect("valid directory");
5941 let membership = reader.table.stripes[0].memberships[1].expect("string membership");
5942 let mut file = OpenOptions::new().write(true).open(&path).expect("open membership page");
5943 file.seek(SeekFrom::Start(membership.offset)).expect("membership start");
5944 file.write_all(&[255]).expect("damage membership");
5945 let error = reader.skips_codes(0, 1, &[3]).expect_err("corruption must not skip rows");
5946 assert!(error.message().contains("membership page checksum differs"), "{error}");
5947 fs::remove_file(path).expect("remove scratch file");
5948 }
5949
5950 #[test]
5951 fn membership_delta_stream_is_sorted_exact_and_bounded() {
5952 let unique = unique_codes(&[900, 4, 4, 72, 9, u32::MAX]);
5953 assert_eq!(unique, [4, 9, 72, 900, u32::MAX]);
5954 let encoded = encode_membership(&unique);
5955 assert_eq!(
5956 decode_membership(&encoded).expect("valid membership"),
5957 [4, 9, 72, 900, u32::MAX]
5958 );
5959 let merged = merged_codes(vec![vec![4, 900], vec![9, 900, u32::MAX], vec![72]]);
5962 assert_eq!(merged, [4, 9, 72, 900, u32::MAX]);
5963 assert_eq!(
5964 decode_membership(&encode_membership(&merged)).expect("valid membership"),
5965 unique
5966 );
5967 assert!(decode_membership(&[1, 0x80]).is_err(), "a truncated varint is invalid");
5968 assert!(
5969 decode_membership(&[1, 0xff, 0xff, 0xff, 0xff, 0x10]).is_err(),
5970 "a value past u32 is invalid"
5971 );
5972 }
5973
5974 #[test]
5975 fn a_global_dictionary_may_be_larger_than_one_column_page() {
5976 let dictionary = Page {
5977 offset: HEADER,
5978 length: u32::try_from(MAX_PAGE + 1).expect("the page bound fits on disk"),
5979 hash: 0,
5980 };
5981 let table = Table {
5982 name: "items".to_owned(),
5983 fields: vec![Field::new("text", LogicalType::Varchar)],
5984 stripes: Vec::new(),
5985 rows: 0,
5986 dictionaries: vec![Some(dictionary)],
5987 distincts: vec![None],
5988 frequencies: vec![None],
5989 };
5990 let directory = encode_directory(&table).expect("directory");
5991 let file_size = dictionary.offset + u64::from(dictionary.length) + 1;
5992
5993 let decoded = decode_directory(&directory, file_size).expect("large lazy dictionary");
5994 assert_eq!(decoded.dictionaries[0].expect("dictionary").length, dictionary.length);
5995 }
5996
5997 #[test]
5998 fn a_column_with_one_value_everywhere_costs_almost_nothing_a_row() {
5999 let path = path("constant-codes");
6000 let mut writer =
6001 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
6002 .expect("new file");
6003 let empty = vec![Value::Varchar(String::new()); 1024];
6004 for _ in 0..4 {
6005 let column = Vector::from_values(LogicalType::Varchar, &empty).expect("strings");
6006 writer.append(&Chunk::new(vec![column]).expect("one column")).expect("a part");
6007 }
6008 writer.finish().expect("commit");
6009
6010 let reader = Reader::open(&path).expect("valid directory");
6011 let pages = reader.layout().columns.first().expect("one column").pages;
6012 assert!(pages < 256, "{pages} bytes of pages for 4,096 rows of one value");
6016 let read = reader.read(3, &[0]).expect("the last part back");
6017 assert_eq!(read.value_at(0, 0), Value::Varchar(String::new()));
6018 assert_eq!(read.value_at(1023, 0), Value::Varchar(String::new()));
6019 fs::remove_file(path).expect("remove scratch file");
6020 }
6021
6022 #[test]
6023 fn a_cascade_value_too_wide_for_its_column_is_refused_rather_than_cut() {
6024 let over = vec![i64::from(i32::MAX) + 1];
6027 let error = narrowed(&LogicalType::Integer, over).expect_err("a page that disagrees");
6028 assert!(format!("{error}").contains("not of its type"), "{error}");
6029 assert!(narrowed(&LogicalType::BigInt, vec![i64::MIN]).is_ok(), "bigint holds all of i64");
6030 assert!(narrowed(&LogicalType::Varchar, vec![0]).is_err(), "strings are not integers");
6031 }
6032
6033 #[test]
6034 fn a_code_stream_the_cascade_cannot_shrink_is_left_alone() {
6035 let mut state: u32 = 0x9e37_79b9;
6039 let spread: Vec<u32> = (0..1024)
6040 .map(|_| {
6041 state ^= state << 13;
6042 state ^= state >> 17;
6043 state ^= state << 5;
6044 state
6045 })
6046 .collect();
6047 assert_eq!(encoded_codes(&spread).expect("no failure"), None);
6048 let near: Vec<u32> = (0..1024).collect();
6049 let coded = encoded_codes(&near).expect("no failure").expect("counting up is packable");
6050 assert!(coded.len() < near.len() * 4, "{} bytes for a run of 1,024", coded.len());
6051 }
6052
6053 #[test]
6059 fn two_writes_of_the_same_rows_give_the_same_bytes() {
6060 fn written(path: &PathBuf) {
6061 let fields = (0..40)
6062 .map(|column| {
6063 let ty =
6064 if column % 4 == 0 { LogicalType::Varchar } else { LogicalType::BigInt };
6065 Field::new(format!("c{column}"), ty)
6066 })
6067 .collect::<Vec<_>>();
6068 let mut writer = Writer::create(path, "wide", fields).expect("new file");
6069 for part in 0..70_u64 {
6070 let columns = (0..40)
6071 .map(|column| {
6072 let values = (0..64_u64)
6073 .map(|row| {
6074 let seed = part.wrapping_mul(31).wrapping_add(row);
6075 if column % 4 == 0 {
6076 Value::Varchar(format!("v{}", seed % 17))
6077 } else {
6078 Value::BigInt(i64::try_from(seed % 97).expect("small"))
6079 }
6080 })
6081 .collect::<Vec<_>>();
6082 let ty = if column % 4 == 0 {
6083 LogicalType::Varchar
6084 } else {
6085 LogicalType::BigInt
6086 };
6087 Vector::from_values(ty, &values).expect("a column")
6088 })
6089 .collect::<Vec<_>>();
6090 writer.append(&Chunk::new(columns).expect("forty columns")).expect("a part");
6091 }
6092 writer.finish().expect("commit");
6093 }
6094
6095 let first = path("repeatable-one");
6096 let second = path("repeatable-two");
6097 written(&first);
6098 written(&second);
6099 let left = fs::read(&first).expect("the first file");
6100 let right = fs::read(&second).expect("the second file");
6101 assert_eq!(left.len(), right.len(), "two writes of the same rows differ in length");
6102 assert!(left == right, "two writes of the same rows differ in their bytes");
6103
6104 let reader = Reader::open(&first).expect("valid directory");
6107 assert_eq!(reader.table().rows(), 70 * 64);
6108 let read = reader.read(0, &[0, 1]).expect("the first part back");
6109 assert_eq!(read.value_at(0, 0), Value::Varchar("v0".to_owned()));
6110 assert_eq!(read.value_at(0, 1), Value::BigInt(0));
6111 fs::remove_file(first).expect("remove scratch file");
6112 fs::remove_file(second).expect("remove scratch file");
6113 }
6114}