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::{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 = 14;
51const HEADER: u64 = 80;
52const SLOT_BYTES: usize = 28;
53const MAX_PAGE: usize = 256 * 1024 * 1024;
54const MAX_DIRECTORY: usize = 128 * 1024 * 1024;
55const FREQUENCIES: &[u8; 8] = b"RUDBFQ2\0";
56const FREQUENCY_CANDIDATES: usize = 32_768;
57const FREQUENCY_ENTRIES: usize = 512;
58const FREQUENCY_BUILD_RANK: usize = 10;
59const FREQUENCY_ORDINALS: usize = 65_536;
60const MAX_FREQUENCY_WORKERS: usize = 16;
61
62const MAX_ENCODE_WORKERS: usize = 32;
69
70const SIEVE_BUDGET: usize = 8 * 1024;
76
77fn io(error: std::io::Error) -> Error {
78 Error::io(error.to_string())
79}
80
81fn invalid(message: &str) -> Error {
82 Error::invalid_input(format!("invalid rudb native file: {message}"))
83}
84
85fn sum(counts: impl Iterator<Item = u64>) -> u64 {
87 counts.fold(0, u64::saturating_add)
88}
89
90fn span_bytes(spans: &[Span], at: usize) -> u64 {
92 spans.get(at).map_or(0, |span| u64::from(span.length))
93}
94
95fn page_bytes(pages: &[Option<Page>], at: usize) -> u64 {
97 pages.get(at).and_then(Option::as_ref).map_or(0, Page::bytes)
98}
99
100fn checksum(bytes: &[u8]) -> u64 {
101 const P1: u64 = 11_400_714_785_074_694_791;
102 const P2: u64 = 14_029_467_366_897_019_727;
103 const P3: u64 = 1_609_587_929_392_839_161;
104 const P4: u64 = 9_650_029_242_287_828_579;
105 const P5: u64 = 2_870_177_450_012_600_261;
106 let round = |state: u64, word: u64| {
107 state.wrapping_add(word.wrapping_mul(P2)).rotate_left(31).wrapping_mul(P1)
108 };
109 let merge = |state: u64, lane: u64| (state ^ round(0, lane)).wrapping_mul(P1).wrapping_add(P4);
110 let word =
111 |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("eight checksum bytes"));
112
113 let mut at = 0;
114 let mut hash = if bytes.len() >= 32 {
115 let mut one = P1.wrapping_add(P2);
116 let mut two = P2;
117 let mut three = 0;
118 let mut four = 0_u64.wrapping_sub(P1);
119 while at + 32 <= bytes.len() {
120 one = round(one, word(at));
121 two = round(two, word(at + 8));
122 three = round(three, word(at + 16));
123 four = round(four, word(at + 24));
124 at += 32;
125 }
126 let combined = one
127 .rotate_left(1)
128 .wrapping_add(two.rotate_left(7))
129 .wrapping_add(three.rotate_left(12))
130 .wrapping_add(four.rotate_left(18));
131 merge(merge(merge(merge(combined, one), two), three), four)
132 } else {
133 P5
134 };
135 hash = hash.wrapping_add(bytes.len() as u64);
136 while at + 8 <= bytes.len() {
137 hash ^= round(0, word(at));
138 hash = hash.rotate_left(27).wrapping_mul(P1).wrapping_add(P4);
139 at += 8;
140 }
141 if at + 4 <= bytes.len() {
142 let tail = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four checksum bytes"));
143 hash ^= u64::from(tail).wrapping_mul(P1);
144 hash = hash.rotate_left(23).wrapping_mul(P2).wrapping_add(P3);
145 at += 4;
146 }
147 while at < bytes.len() {
148 hash ^= u64::from(bytes[at]).wrapping_mul(P5);
149 hash = hash.rotate_left(11).wrapping_mul(P1);
150 at += 1;
151 }
152 hash ^= hash >> 33;
153 hash = hash.wrapping_mul(P2);
154 hash ^= hash >> 29;
155 hash = hash.wrapping_mul(P3);
156 hash ^ (hash >> 32)
157}
158
159#[derive(Debug, Clone, Copy)]
160struct Slot {
161 offset: u64,
162 length: u32,
163 generation: u64,
164 hash: u64,
165}
166
167impl Slot {
168 fn bytes(self) -> [u8; SLOT_BYTES] {
169 let mut result = [0; SLOT_BYTES];
170 result[..8].copy_from_slice(&self.offset.to_le_bytes());
171 result[8..12].copy_from_slice(&self.length.to_le_bytes());
172 result[12..20].copy_from_slice(&self.generation.to_le_bytes());
173 result[20..28].copy_from_slice(&self.hash.to_le_bytes());
174 result
175 }
176
177 fn read(bytes: &[u8]) -> Self {
178 Self {
179 offset: u64::from_le_bytes(bytes[..8].try_into().expect("eight bytes")),
180 length: u32::from_le_bytes(bytes[8..12].try_into().expect("four bytes")),
181 generation: u64::from_le_bytes(bytes[12..20].try_into().expect("eight bytes")),
182 hash: u64::from_le_bytes(bytes[20..28].try_into().expect("eight bytes")),
183 }
184 }
185}
186
187#[derive(Debug, Clone, Copy)]
188struct Page {
189 offset: u64,
190 length: u32,
191 hash: u64,
192}
193
194impl Page {
195 fn bytes(&self) -> u64 {
197 u64::from(self.length)
198 }
199}
200
201#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
202enum FrequencyValue {
203 Null,
204 Integer(i128),
205 Code(u32),
206}
207
208#[derive(Debug, Clone)]
209struct FrequencyEntry {
210 value: FrequencyValue,
211 count: u64,
212}
213
214#[derive(Debug, Clone)]
219struct FrequencySummary {
220 entries: Vec<FrequencyEntry>,
221 omitted_max: u64,
222 ordinals: Vec<u64>,
223}
224
225#[derive(Debug, Clone, PartialEq, Eq)]
227pub struct FrequencyOccurrences {
228 pub omitted_max: u64,
230 pub ordinals: Vec<u64>,
232}
233
234#[derive(Debug, Clone, Copy, Default)]
241struct Span {
242 offset: u64,
243 length: u32,
244}
245
246#[derive(Debug, Clone)]
248pub struct Stripe {
249 rows: usize,
250 parts: Vec<u32>,
253 index: Span,
257 pages: Vec<Span>,
258 memberships: Vec<Option<Page>>,
259 sieves: Vec<Option<Page>>,
262 zone: Zone,
263}
264
265impl Stripe {
266 #[must_use]
268 pub fn rows(&self) -> usize {
269 self.rows
270 }
271
272 #[must_use]
274 pub fn parts(&self) -> usize {
275 self.parts.len()
276 }
277}
278
279#[derive(Debug, Clone)]
281pub struct Table {
282 name: String,
283 fields: Vec<Field>,
284 stripes: Vec<Stripe>,
285 rows: usize,
286 dictionaries: Vec<Option<Page>>,
287 frequencies: Vec<Option<FrequencySummary>>,
288}
289
290impl Table {
291 #[must_use]
293 pub fn name(&self) -> &str {
294 &self.name
295 }
296
297 #[must_use]
299 pub fn fields(&self) -> &[Field] {
300 &self.fields
301 }
302
303 #[must_use]
305 pub fn rows(&self) -> usize {
306 self.rows
307 }
308
309 #[must_use]
311 pub fn stripes(&self) -> &[Stripe] {
312 &self.stripes
313 }
314}
315
316#[derive(Debug, Clone)]
318pub struct ColumnLayout {
319 pub name: String,
321 pub kind: String,
323 pub pages: u64,
325 pub memberships: u64,
327 pub sieves: u64,
329 pub dictionary: u64,
331}
332
333impl ColumnLayout {
334 #[must_use]
336 pub fn total(&self) -> u64 {
337 self.pages
338 .saturating_add(self.memberships)
339 .saturating_add(self.sieves)
340 .saturating_add(self.dictionary)
341 }
342}
343
344#[derive(Debug, Clone)]
355pub struct Layout {
356 pub file: u64,
358 pub rows: usize,
360 pub stripes: usize,
362 pub parts: usize,
364 pub columns: Vec<ColumnLayout>,
366 pub indexes: u64,
369 pub directory: u64,
371 pub header: u64,
373}
374
375impl Layout {
376 #[must_use]
378 pub fn columns_total(&self) -> u64 {
379 self.columns.iter().map(ColumnLayout::total).fold(0, u64::saturating_add)
380 }
381
382 #[must_use]
388 pub fn unaccounted(&self) -> u64 {
389 self.file
390 .saturating_sub(self.columns_total())
391 .saturating_sub(self.indexes)
392 .saturating_sub(self.directory)
393 .saturating_sub(self.header)
394 }
395}
396
397#[derive(Debug)]
399struct GlobalDictionary {
400 primary: HashMap<u64, u32>,
401 collisions: HashMap<u64, Vec<u32>>,
402 offsets: Vec<u32>,
403 payload: Vec<u8>,
404 counts: Vec<u64>,
405 nulls: u64,
406}
407
408impl GlobalDictionary {
409 fn new() -> Self {
410 Self {
411 primary: HashMap::new(),
412 collisions: HashMap::new(),
413 offsets: vec![0],
414 payload: Vec::new(),
415 counts: Vec::new(),
416 nulls: 0,
417 }
418 }
419
420 fn bytes(&self, code: u32) -> Option<&[u8]> {
421 let start = *self.offsets.get(code as usize)? as usize;
422 let end = *self.offsets.get(code as usize + 1)? as usize;
423 self.payload.get(start..end)
424 }
425
426 fn code(&mut self, text: &str) -> Result<u32> {
427 let hash = checksum(text.as_bytes());
428 if let Some(&code) = self.primary.get(&hash) {
429 if self.bytes(code) == Some(text.as_bytes()) {
430 return Ok(code);
431 }
432 if let Some(codes) = self.collisions.get(&hash) {
433 if let Some(code) =
434 codes.iter().copied().find(|&code| self.bytes(code) == Some(text.as_bytes()))
435 {
436 return Ok(code);
437 }
438 }
439 let code = self.insert(text)?;
440 self.collisions.entry(hash).or_default().push(code);
441 return Ok(code);
442 }
443 let code = self.insert(text)?;
444 self.primary.insert(hash, code);
445 Ok(code)
446 }
447
448 fn insert(&mut self, text: &str) -> Result<u32> {
449 let code = u32::try_from(self.offsets.len() - 1)
450 .map_err(|_| invalid("global dictionary has too many values"))?;
451 self.payload.extend_from_slice(text.as_bytes());
452 self.offsets.push(
453 u32::try_from(self.payload.len())
454 .map_err(|_| invalid("global dictionary payload exceeds 4 GiB"))?,
455 );
456 self.counts.push(0);
457 Ok(code)
458 }
459
460 fn ranked(&self) -> Vec<(u64, u32)> {
480 let count = self.offsets.len() - 1;
481 let mut ranked = (0..count)
482 .map(|code| {
483 let code = code as u32;
484 (head(self.bytes(code).unwrap_or_default()), code)
485 })
486 .collect::<Vec<_>>();
487 ranked.sort_unstable_by(|left, right| {
488 left.0.cmp(&right.0).then_with(|| self.bytes(left.1).cmp(&self.bytes(right.1)))
489 });
490 ranked
491 }
492
493 fn observe(&mut self, code: u32, null: bool) -> Result<()> {
494 if null {
495 self.nulls = self.nulls.saturating_add(1);
496 return Ok(());
497 }
498 let count = self
499 .counts
500 .get_mut(code as usize)
501 .ok_or_else(|| invalid("global dictionary count code is out of range"))?;
502 *count = count.saturating_add(1);
503 Ok(())
504 }
505}
506
507#[derive(Debug)]
509pub struct Writer {
510 file: File,
511 at: u64,
519 table: Table,
520 generation: u64,
521 order: Vec<((u64, u64), (u64, u64))>,
524 next_order: u64,
525 dictionaries: Vec<Option<GlobalDictionary>>,
526 pending: Vec<PendingChunk>,
527}
528
529#[derive(Debug)]
537struct PendingChunk {
538 order: (u64, u64),
539 chunk: Chunk,
540}
541
542#[derive(Debug)]
548struct ColumnStripe {
549 pages: Vec<Vec<u8>>,
550 codes: Vec<Option<Vec<u32>>>,
551 sieves: Vec<Option<Sieve>>,
552 ranges: Vec<Range>,
553}
554
555fn weight(ty: &LogicalType) -> usize {
563 match ty {
564 LogicalType::Varchar | LogicalType::Blob => 64,
565 LogicalType::BigInt
566 | LogicalType::UBigInt
567 | LogicalType::Timestamp
568 | LogicalType::Double
569 | LogicalType::Decimal { .. } => 8,
570 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date | LogicalType::Float => 4,
571 LogicalType::SmallInt | LogicalType::USmallInt => 2,
572 _ => 1,
573 }
574}
575
576const STRIPE_PARTS: usize = 64;
583
584const INDEX_ENTRY: usize = size_of::<u32>() + size_of::<u64>();
586
587fn index_section(parts: usize) -> Result<usize> {
589 parts
590 .checked_mul(INDEX_ENTRY)
591 .and_then(|bytes| bytes.checked_add(size_of::<u64>()))
592 .ok_or_else(|| invalid("index page length overflow"))
593}
594
595impl Writer {
596 pub fn create(
602 path: impl AsRef<Path>,
603 name: impl Into<String>,
604 fields: Vec<Field>,
605 ) -> Result<Self> {
606 for field in &fields {
607 type_tag(&field.ty)?;
608 }
609 let file =
610 OpenOptions::new().write(true).read(true).create_new(true).open(path).map_err(io)?;
611 let mut header = [0; HEADER as usize];
612 header[..8].copy_from_slice(MAGIC);
613 header[8..12].copy_from_slice(&FORMAT.to_le_bytes());
614 write_at(&file, 0, &header)?;
615 Ok(Self {
616 file,
617 at: HEADER,
618 dictionaries: fields
619 .iter()
620 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
621 .collect(),
622 table: Table {
623 name: name.into(),
624 dictionaries: vec![None; fields.len()],
625 fields,
626 stripes: Vec::new(),
627 rows: 0,
628 frequencies: Vec::new(),
629 },
630 generation: 1,
631 order: Vec::new(),
632 next_order: 0,
633 pending: Vec::with_capacity(STRIPE_PARTS),
634 })
635 }
636
637 fn put(&mut self, bytes: &[u8]) -> Result<()> {
642 write_at(&self.file, self.at, bytes)?;
643 self.at = self
644 .at
645 .checked_add(bytes.len() as u64)
646 .ok_or_else(|| invalid("native file length overflow"))?;
647 Ok(())
648 }
649
650 pub fn append(&mut self, chunk: &Chunk) -> Result<()> {
656 let order = (self.next_order, 0);
657 self.next_order = self.next_order.saturating_add(1);
658 self.append_at(order, chunk)
659 }
660
661 pub fn append_at(&mut self, order: (u64, u64), chunk: &Chunk) -> Result<()> {
672 if chunk.is_empty() {
673 return Ok(());
674 }
675 if chunk.width() != self.table.fields.len() {
676 return Err(invalid("chunk width differs from table schema"));
677 }
678 for (index, field) in self.table.fields.iter().enumerate() {
679 if chunk.column(index)?.logical_type() != &field.ty {
680 return Err(invalid("chunk type differs from table schema"));
681 }
682 }
683 self.table.rows = self
684 .table
685 .rows
686 .checked_add(chunk.len())
687 .ok_or_else(|| invalid("row count overflow"))?;
688 if self.pending.last().is_some_and(|last| last.order > order) {
689 self.flush_pending()?;
690 }
691 self.pending.push(PendingChunk { order, chunk: chunk.clone() });
696 if self.pending.len() == STRIPE_PARTS {
697 self.flush_pending()?;
698 }
699 Ok(())
700 }
701
702 fn encode_column(
710 index: usize,
711 held: &[PendingChunk],
712 mut dictionary: Option<&mut GlobalDictionary>,
713 ) -> Result<ColumnStripe> {
714 let mut stripe = ColumnStripe {
715 pages: Vec::with_capacity(held.len()),
716 codes: Vec::with_capacity(held.len()),
717 sieves: Vec::with_capacity(held.len()),
718 ranges: Vec::with_capacity(held.len()),
719 };
720 for pending in held {
721 let column = pending.chunk.column(index)?;
722 let (bytes, unique) = encode(column, dictionary.as_deref_mut())?;
723 if bytes.len() > MAX_PAGE {
724 return Err(invalid("column page exceeds the configured bound"));
725 }
726 let range = Range::of(column);
729 let sieve = match dictionary {
734 Some(_) => None,
735 None => Sieve::of(column, &range, SIEVE_BUDGET),
736 };
737 stripe.pages.push(bytes);
738 stripe.codes.push(unique);
739 stripe.sieves.push(sieve);
740 stripe.ranges.push(range);
741 }
742 Ok(stripe)
743 }
744
745 fn encode_columns(&mut self, held: &[PendingChunk]) -> Result<Vec<ColumnStripe>> {
754 let width = self.table.fields.len();
755 let workers = std::thread::available_parallelism()
756 .map_or(1, usize::from)
757 .min(MAX_ENCODE_WORKERS)
758 .min(width);
759 if workers <= 1 || held.len() <= 1 {
760 return self
761 .dictionaries
762 .iter_mut()
763 .enumerate()
764 .map(|(index, dictionary)| Self::encode_column(index, held, dictionary.as_mut()))
765 .collect();
766 }
767 let mut jobs: Vec<(usize, Option<GlobalDictionary>)> =
770 std::mem::take(&mut self.dictionaries).into_iter().enumerate().collect();
771 jobs.sort_by_key(|(index, _)| weight(&self.table.fields[*index].ty));
773 let queue = Mutex::new(jobs);
774 let pieces = std::thread::scope(|scope| {
775 (0..workers)
776 .map(|_| {
777 scope.spawn(|| {
778 let mut mine = Vec::new();
779 loop {
780 let taken = queue
781 .lock()
782 .map_err(|_| Error::internal("a native encode worker panicked"))?
783 .pop();
784 let Some((index, mut dictionary)) = taken else { break };
785 let encoded = Self::encode_column(index, held, dictionary.as_mut())?;
786 mine.push((index, dictionary, encoded));
787 }
788 Ok(mine)
789 })
790 })
791 .collect::<Vec<_>>()
792 .into_iter()
793 .map(|handle| {
794 handle.join().map_err(|_| Error::internal("a native encode worker panicked"))?
795 })
796 .collect::<Result<Vec<_>>>()
797 })?;
798 let mut dictionaries: Vec<Option<GlobalDictionary>> = (0..width).map(|_| None).collect();
799 let mut encoded: Vec<Option<ColumnStripe>> = (0..width).map(|_| None).collect();
800 for piece in pieces {
801 for (index, dictionary, stripe) in piece {
802 dictionaries[index] = dictionary;
803 encoded[index] = Some(stripe);
804 }
805 }
806 self.dictionaries = dictionaries;
807 encoded
808 .into_iter()
809 .map(|stripe| stripe.ok_or_else(|| Error::internal("a column was never encoded")))
810 .collect()
811 }
812
813 fn flush_pending(&mut self) -> Result<()> {
815 if self.pending.is_empty() {
816 return Ok(());
817 }
818 let width = self.table.fields.len();
819 let mut held = std::mem::take(&mut self.pending);
822 let parts = held.len();
823 let encoded = self.encode_columns(&held)?;
824 let mut pages = Vec::with_capacity(width);
825 let mut memberships = vec![None; width];
826 let mut ranges = Vec::with_capacity(width);
827 let mut index = Vec::with_capacity(width.saturating_mul(index_section(parts)?));
828 for stripe in &encoded {
829 let offset = self.at;
830 let section = index.len();
831 let mut length = 0_usize;
832 for bytes in &stripe.pages {
833 write_at(&self.file, self.at + length as u64, bytes)?;
834 put_u32(
835 &mut index,
836 u32::try_from(bytes.len()).map_err(|_| invalid("part length overflow"))?,
837 );
838 put_u64(&mut index, checksum(bytes));
839 length = length
840 .checked_add(bytes.len())
841 .ok_or_else(|| invalid("column page length overflow"))?;
842 }
843 let hash = checksum(&index[section..]);
844 put_u64(&mut index, hash);
845 if length > MAX_PAGE {
846 return Err(invalid("column page exceeds the configured bound"));
847 }
848 self.at = self
849 .at
850 .checked_add(length as u64)
851 .ok_or_else(|| invalid("native file length overflow"))?;
852 pages.push(Span {
853 offset,
854 length: u32::try_from(length).map_err(|_| invalid("page length overflow"))?,
855 });
856 ranges.push(merged_range(stripe.ranges.iter().cloned()));
857 }
858 for (membership, stripe) in memberships.iter_mut().zip(&encoded) {
859 if stripe.codes.iter().all(Option::is_none) {
860 continue;
861 }
862 let lists = stripe
863 .codes
864 .iter()
865 .map(|codes| codes.clone().unwrap_or_default())
866 .collect::<Vec<_>>();
867 let bytes = encode_membership(&merged_codes(lists));
868 let offset = self.at;
869 self.put(&bytes)?;
870 *membership = Some(Page {
871 offset,
872 length: u32::try_from(bytes.len())
873 .map_err(|_| invalid("membership page length overflow"))?,
874 hash: checksum(&bytes),
875 });
876 }
877 let mut sieves = vec![None; width];
878 for (page, stripe) in sieves.iter_mut().zip(&encoded) {
879 if stripe.sieves.iter().all(Option::is_none) {
880 continue;
881 }
882 let bytes = encode_sieves(stripe.sieves.iter())?;
883 let offset = self.at;
884 self.put(&bytes)?;
885 *page = Some(Page {
886 offset,
887 length: u32::try_from(bytes.len())
888 .map_err(|_| invalid("sieve page length overflow"))?,
889 hash: checksum(&bytes),
890 });
891 }
892 let offset = self.at;
893 self.put(&index)?;
894 let index = Span {
895 offset,
896 length: u32::try_from(index.len())
897 .map_err(|_| invalid("index page length overflow"))?,
898 };
899 let mut rows = 0_usize;
900 let mut lengths = Vec::with_capacity(parts);
901 let mut span = None;
902 for pending in held.drain(..) {
903 let part = pending.chunk.len();
904 rows = rows.checked_add(part).ok_or_else(|| invalid("row count overflow"))?;
905 lengths.push(u32::try_from(part).map_err(|_| invalid("part row count overflow"))?);
906 span = Some(
907 span.map_or((pending.order, pending.order), |(first, _)| (first, pending.order)),
908 );
909 }
910 self.order.push(span.ok_or_else(|| invalid("a stripe was flushed with no parts"))?);
911 self.table.stripes.push(Stripe {
912 rows,
913 parts: lengths,
914 index,
915 pages,
916 memberships,
917 sieves,
918 zone: Zone::from_ranges(ranges),
919 });
920 self.pending = held;
922 Ok(())
923 }
924
925 fn numeric_frequency(&self, column: usize) -> Result<Option<FrequencySummary>> {
929 let ty = &self.table.fields[column].ty;
930 if !matches!(
931 ty,
932 LogicalType::TinyInt
933 | LogicalType::SmallInt
934 | LogicalType::Integer
935 | LogicalType::BigInt
936 | LogicalType::UTinyInt
937 | LogicalType::USmallInt
938 | LogicalType::UInteger
939 | LogicalType::UBigInt
940 | LogicalType::Date
941 | LogicalType::Timestamp
942 ) {
943 return Ok(None);
944 }
945 let mut candidates: HashMap<FrequencyValue, u32> = HashMap::new();
946 let mut decrements = 0_u64;
947 self.visit_numeric(column, |_, value| {
948 if let Some(count) = candidates.get_mut(&value) {
949 *count = count.saturating_add(1);
950 } else if candidates.len() < FREQUENCY_CANDIDATES {
951 candidates.insert(value, 1);
952 } else {
953 candidates.retain(|_, count| {
954 *count -= 1;
955 *count != 0
956 });
957 decrements = decrements.saturating_add(1);
958 }
959 })?;
960 let (exact, ordinals) = if decrements == 0 {
961 (
962 candidates
963 .into_iter()
964 .map(|(value, count)| (value, u64::from(count)))
965 .collect::<HashMap<_, _>>(),
966 Vec::new(),
967 )
968 } else {
969 let mut lower = candidates.values().copied().collect::<Vec<_>>();
970 lower.sort_unstable_by(|left, right| right.cmp(left));
971 if lower.len() < FREQUENCY_BUILD_RANK
972 || u64::from(lower[FREQUENCY_BUILD_RANK - 1]) <= decrements
973 {
974 return Ok(None);
975 }
976 let mut exact =
977 candidates.into_keys().map(|value| (value, 0_u64)).collect::<HashMap<_, _>>();
978 let mut ordinals = Vec::new();
979 let mut exceeded = false;
980 self.visit_numeric(column, |ordinal, value| {
981 if let Some(count) = exact.get_mut(&value) {
982 *count = count.saturating_add(1);
983 if !exceeded {
984 if ordinals.len() < FREQUENCY_ORDINALS {
985 ordinals.push(ordinal);
986 } else {
987 ordinals.clear();
988 exceeded = true;
989 }
990 }
991 }
992 })?;
993 (exact, ordinals)
994 };
995 let mut entries = exact
996 .into_iter()
997 .map(|(value, count)| FrequencyEntry { value, count })
998 .collect::<Vec<_>>();
999 entries.sort_unstable_by(|left, right| {
1000 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
1001 });
1002 let omitted_max =
1003 entries.get(FREQUENCY_ENTRIES).map_or(decrements, |entry| decrements.max(entry.count));
1004 entries.truncate(FREQUENCY_ENTRIES);
1005 Ok(Some(FrequencySummary { entries, omitted_max, ordinals }))
1006 }
1007
1008 fn visit_numeric(
1009 &self,
1010 column: usize,
1011 mut visit: impl FnMut(u64, FrequencyValue),
1012 ) -> Result<()> {
1013 let ty = &self.table.fields[column].ty;
1014 let mut start = 0_u64;
1015 for stripe in &self.table.stripes {
1016 let spans = read_index(&self.file, stripe, column)?;
1017 let page = stripe.pages[column];
1018 let mut bytes = vec![0; page.length as usize];
1019 read_at(&self.file, page.offset, &mut bytes)?;
1020 for (span, &rows) in spans.iter().zip(&stripe.parts) {
1021 let part = part_bytes(&bytes, *span)?;
1022 if checksum(part) != span.hash {
1023 return Err(invalid("column page checksum differs while building frequencies"));
1024 }
1025 let rows = rows as usize;
1026 let vector = decode(ty, rows, part, None)?;
1027 for row in 0..rows {
1029 let value = if vector.is_null_at(row) {
1030 FrequencyValue::Null
1031 } else {
1032 let widened = match vector.signed_at(row) {
1036 Some(value) => Some(value),
1037 None => match vector.value_at(row) {
1038 Value::UTinyInt(value) => Some(i128::from(value)),
1039 Value::USmallInt(value) => Some(i128::from(value)),
1040 Value::UInteger(value) => Some(i128::from(value)),
1041 Value::UBigInt(value) => Some(i128::from(value)),
1042 _ => None,
1043 },
1044 };
1045 FrequencyValue::Integer(widened.ok_or_else(|| {
1046 invalid("numeric frequency page did not contain an integer value")
1047 })?)
1048 };
1049 visit(start.saturating_add(row as u64), value);
1050 }
1051 start = start.saturating_add(rows as u64);
1052 }
1053 }
1054 Ok(())
1055 }
1056
1057 fn numeric_frequencies(&self) -> Result<Vec<Option<FrequencySummary>>> {
1059 let columns = self
1060 .table
1061 .fields
1062 .iter()
1063 .enumerate()
1064 .filter_map(|(column, field)| {
1065 matches!(
1066 field.ty,
1067 LogicalType::TinyInt
1068 | LogicalType::SmallInt
1069 | LogicalType::Integer
1070 | LogicalType::BigInt
1071 | LogicalType::UTinyInt
1072 | LogicalType::USmallInt
1073 | LogicalType::UInteger
1074 | LogicalType::UBigInt
1075 | LogicalType::Date
1076 | LogicalType::Timestamp
1077 )
1078 .then_some(column)
1079 })
1080 .collect::<Vec<_>>();
1081 let workers = std::thread::available_parallelism()
1082 .map_or(1, usize::from)
1083 .min(MAX_FREQUENCY_WORKERS)
1084 .min(columns.len());
1085 if workers <= 1 {
1086 let mut frequencies = vec![None; self.table.fields.len()];
1087 for column in columns {
1088 frequencies[column] = self.numeric_frequency(column)?;
1089 }
1090 return Ok(frequencies);
1091 }
1092 let width = columns.len().div_ceil(workers);
1093 let pieces = std::thread::scope(|scope| {
1094 columns
1095 .chunks(width)
1096 .map(|columns| {
1097 scope.spawn(|| {
1098 columns
1099 .iter()
1100 .map(|&column| Ok((column, self.numeric_frequency(column)?)))
1101 .collect::<Result<Vec<_>>>()
1102 })
1103 })
1104 .collect::<Vec<_>>()
1105 .into_iter()
1106 .map(|handle| {
1107 handle
1108 .join()
1109 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1110 })
1111 .collect::<Result<Vec<_>>>()
1112 })?;
1113 let mut frequencies = vec![None; self.table.fields.len()];
1114 for piece in pieces {
1115 for (column, summary) in piece {
1116 frequencies[column] = summary;
1117 }
1118 }
1119 Ok(frequencies)
1120 }
1121
1122 pub fn finish(mut self) -> Result<Table> {
1128 self.flush_pending()?;
1129 let mut stripes = std::mem::take(&mut self.order)
1130 .into_iter()
1131 .zip(std::mem::take(&mut self.table.stripes))
1132 .collect::<Vec<_>>();
1133 stripes.sort_by_key(|(order, _)| order.0);
1134 let mut previous: Option<(u64, u64)> = None;
1135 for ((first, last), _) in &stripes {
1136 if previous.is_some_and(|previous| previous >= *first) {
1137 return Err(invalid("chunks did not arrive in source order"));
1138 }
1139 previous = Some(*last);
1140 }
1141 self.table.stripes = stripes.into_iter().map(|(_, stripe)| stripe).collect();
1142 self.table.frequencies = self.numeric_frequencies()?;
1143 let dictionaries = std::mem::take(&mut self.dictionaries);
1144 let orders = rankings(&dictionaries)?;
1145 for (index, (dictionary, order)) in dictionaries.into_iter().zip(orders).enumerate() {
1146 let Some(dictionary) = dictionary else { continue };
1147 self.table.frequencies[index] = Some(code_frequency(&dictionary));
1148 let encoded = encode_global_dictionary(dictionary, &order)?;
1149 let offset = self.at;
1150 self.put(&encoded.index)?;
1151 self.put(&encoded.ranks)?;
1152 self.put(&encoded.payload)?;
1153 let length = encoded
1154 .index
1155 .len()
1156 .checked_add(encoded.ranks.len())
1157 .and_then(|len| len.checked_add(encoded.payload.len()))
1158 .ok_or_else(|| invalid("dictionary page length overflow"))?;
1159 self.table.dictionaries[index] = Some(Page {
1160 offset,
1161 length: u32::try_from(length)
1162 .map_err(|_| invalid("dictionary page length overflow"))?,
1163 hash: checksum(&encoded.index),
1164 });
1165 }
1166 let directory = encode_directory(&self.table)?;
1167 if directory.len() > MAX_DIRECTORY {
1168 return Err(invalid("directory exceeds the configured bound"));
1169 }
1170 let offset = self.at;
1171 self.put(&directory)?;
1172 self.file.sync_all().map_err(io)?;
1173 let slot = Slot {
1174 offset,
1175 length: u32::try_from(directory.len())
1176 .map_err(|_| invalid("directory length overflow"))?,
1177 generation: self.generation,
1178 hash: checksum(&directory),
1179 };
1180 write_at(&self.file, 16, &slot.bytes())?;
1183 self.file.sync_all().map_err(io)?;
1184 Ok(self.table)
1185 }
1186}
1187
1188#[derive(Debug, Clone)]
1190pub struct Reader {
1191 file: Arc<File>,
1192 table: Arc<Table>,
1193 dictionaries: Arc<Vec<OnceLock<Arc<Vector>>>>,
1194 sieves: Arc<Vec<Vec<SieveSlot>>>,
1198 places: Arc<Vec<Place>>,
1200 cache: Arc<Vec<Mutex<Cached>>>,
1201 pages: Arc<AtomicUsize>,
1204 indexes: Arc<AtomicUsize>,
1207 kept: Arc<AtomicUsize>,
1210 size: u64,
1212 directory: u64,
1214 opening: Opening,
1216}
1217
1218#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1230pub struct Opening {
1231 pub reads: u32,
1234 pub bytes: u64,
1236}
1237
1238#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1240pub struct Reads {
1241 pub opening: Opening,
1243 pub pages: usize,
1245 pub indexes: usize,
1247}
1248
1249#[derive(Debug, Clone, Copy)]
1251struct Place {
1252 stripe: u32,
1253 part: u32,
1254 rows: u32,
1255}
1256
1257#[derive(Debug, Clone, Copy)]
1259struct PartSpan {
1260 start: usize,
1261 length: usize,
1262 hash: u64,
1263}
1264
1265#[derive(Debug, Clone)]
1271struct CachedColumn {
1272 stripe: usize,
1273 index: Arc<Vec<PartSpan>>,
1274 page: Option<Arc<Vec<u8>>>,
1275}
1276
1277#[derive(Debug, Default)]
1297struct Cached {
1298 pages: Vec<Option<Arc<Vec<u8>>>>,
1299 order: VecDeque<usize>,
1300 loading: Vec<usize>,
1301 index: Vec<Option<Arc<Vec<PartSpan>>>>,
1302}
1303
1304const CACHED_STRIPES_PER_COLUMN: usize = 4;
1316
1317type SieveSlot = OnceLock<Arc<Vec<Option<Sieve>>>>;
1319
1320type CrossingCache = OnceLock<Box<[OnceLock<Result<Vec<u8>>>]>>;
1321
1322#[derive(Debug)]
1323struct NativeText {
1324 file: Arc<File>,
1325 offsets: Vec<u32>,
1326 ranks: usize,
1328 rank_at: u64,
1332 rank_hashes: Vec<u64>,
1333 rank_blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1334 code_ranks: OnceLock<Option<Vec<u32>>>,
1341 payload: u64,
1342 payload_len: usize,
1343 hashes: Vec<u64>,
1344 payload_extents: Vec<OnceLock<Result<Vec<u8>>>>,
1346 crossing: Vec<CrossingCache>,
1347}
1348
1349const TEXT_PAYLOAD_BLOCK: usize = 64 * 1024;
1350
1351const TEXT_PAYLOAD_EXTENT: usize = 8;
1372const TEXT_CROSSING_BLOCK: usize = 1024;
1373
1374const TEXT_RANK_BLOCK: usize = 512;
1384
1385const RANK_ENTRY: usize = size_of::<u64>() + size_of::<u32>();
1387
1388impl NativeText {
1389 fn payload_block(&self, block: usize) -> Result<Option<&[u8]>> {
1390 if block >= self.hashes.len() {
1391 return Ok(None);
1392 }
1393 let extent = block / TEXT_PAYLOAD_EXTENT;
1394 let Some(slot) = self.payload_extents.get(extent) else { return Ok(None) };
1395 let bytes = slot
1396 .get_or_init(|| {
1397 let start = extent
1398 .checked_mul(TEXT_PAYLOAD_EXTENT * TEXT_PAYLOAD_BLOCK)
1399 .ok_or_else(|| invalid("global dictionary block offset overflow"))?;
1400 let len = (TEXT_PAYLOAD_EXTENT * TEXT_PAYLOAD_BLOCK).min(
1401 self.payload_len
1402 .checked_sub(start)
1403 .ok_or_else(|| invalid("global dictionary block starts past payload"))?,
1404 );
1405 let mut bytes = vec![0; len];
1406 read_at(&self.file, self.payload + start as u64, &mut bytes)?;
1407 for (within, piece) in bytes.chunks(TEXT_PAYLOAD_BLOCK).enumerate() {
1410 if checksum(piece)
1411 != *self
1412 .hashes
1413 .get(extent * TEXT_PAYLOAD_EXTENT + within)
1414 .ok_or_else(|| invalid("global dictionary block has no checksum"))?
1415 {
1416 return Err(invalid("global dictionary payload checksum differs"));
1417 }
1418 }
1419 Ok(bytes)
1420 })
1421 .as_ref()
1422 .map_err(Clone::clone)?;
1423 let within = (block % TEXT_PAYLOAD_EXTENT) * TEXT_PAYLOAD_BLOCK;
1424 let end = (within + TEXT_PAYLOAD_BLOCK).min(bytes.len());
1425 Ok(bytes.get(within..end))
1426 }
1427
1428 fn rank_parts(&self, rank: usize) -> Result<(&[u8], usize)> {
1435 let slot = self
1436 .rank_blocks
1437 .get(rank / TEXT_RANK_BLOCK)
1438 .ok_or_else(|| invalid("global dictionary rank is past the order"))?;
1439 let block = slot
1440 .get_or_init(|| {
1441 let first = rank / TEXT_RANK_BLOCK * TEXT_RANK_BLOCK;
1442 let len = TEXT_RANK_BLOCK.min(self.ranks - first) * RANK_ENTRY;
1443 let mut bytes = vec![0; len];
1444 read_at(&self.file, self.rank_at + (first * RANK_ENTRY) as u64, &mut bytes)?;
1445 if checksum(&bytes)
1446 != *self
1447 .rank_hashes
1448 .get(rank / TEXT_RANK_BLOCK)
1449 .ok_or_else(|| invalid("global dictionary rank block has no checksum"))?
1450 {
1451 return Err(invalid("global dictionary rank checksum differs"));
1452 }
1453 Ok(bytes)
1454 })
1455 .as_ref()
1456 .map_err(Clone::clone)?;
1457 Ok((block.as_slice(), rank % TEXT_RANK_BLOCK))
1458 }
1459
1460 fn head_at(&self, rank: usize) -> Result<u64> {
1462 let (block, within) = self.rank_parts(rank)?;
1463 let at = within * size_of::<u64>();
1464 let bytes = block
1465 .get(at..at + size_of::<u64>())
1466 .ok_or_else(|| invalid("global dictionary rank block is short of heads"))?;
1467 Ok(u64::from_le_bytes(bytes.try_into().expect("eight bytes")))
1468 }
1469}
1470
1471impl TextSource for NativeText {
1472 fn len(&self) -> usize {
1473 self.offsets.len().saturating_sub(1)
1474 }
1475
1476 fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
1477 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1478 else {
1479 return Ok(None);
1480 };
1481 if start == end {
1482 return Ok(Some(&[]));
1483 }
1484 let first = start as usize / TEXT_PAYLOAD_BLOCK;
1485 let last = (end as usize - 1) / TEXT_PAYLOAD_BLOCK;
1486 if first == last {
1487 let Some(block) = self.payload_block(first)? else { return Ok(None) };
1488 let within = start as usize % TEXT_PAYLOAD_BLOCK;
1489 return Ok(block.get(within..within + (end - start) as usize));
1490 }
1491 let Some(crossing) = self.crossing.get(index / TEXT_CROSSING_BLOCK) else {
1492 return Ok(None);
1493 };
1494 let block = crossing.get_or_init(|| {
1495 (0..TEXT_CROSSING_BLOCK).map(|_| OnceLock::new()).collect::<Vec<_>>().into_boxed_slice()
1496 });
1497 block[index % TEXT_CROSSING_BLOCK]
1498 .get_or_init(|| {
1499 let mut bytes = Vec::with_capacity((end - start) as usize);
1500 for part in first..=last {
1501 let source = self
1502 .payload_block(part)?
1503 .ok_or_else(|| invalid("global dictionary block is missing"))?;
1504 let from = if part == first { start as usize % TEXT_PAYLOAD_BLOCK } else { 0 };
1505 let to = if part == last {
1506 (end as usize - 1) % TEXT_PAYLOAD_BLOCK + 1
1507 } else {
1508 source.len()
1509 };
1510 bytes.extend_from_slice(source.get(from..to).ok_or_else(|| {
1511 invalid("global dictionary value exceeds its payload block")
1512 })?);
1513 }
1514 Ok(bytes)
1515 })
1516 .as_ref()
1517 .map(|bytes| Some(bytes.as_slice()))
1518 .map_err(Clone::clone)
1519 }
1520
1521 fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
1522 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1523 else {
1524 return Ok(None);
1525 };
1526 Ok(Some((end - start) as usize))
1527 }
1528
1529 fn ranks(&self) -> Option<usize> {
1530 (self.ranks > 0).then_some(self.ranks)
1531 }
1532
1533 fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
1534 let settled = self.head_at(rank)?.cmp(&head(wanted));
1538 if settled != Ordering::Equal {
1539 return Ok(settled);
1540 }
1541 let code = self.code_at_rank(rank)?;
1542 let bytes = self
1543 .bytes_at(code as usize)?
1544 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
1545 Ok(bytes.cmp(wanted))
1546 }
1547
1548 fn code_at_rank(&self, rank: usize) -> Result<u32> {
1549 let (block, within) = self.rank_parts(rank)?;
1550 let heads = block.len() / RANK_ENTRY * size_of::<u64>();
1551 let at = heads + within * size_of::<u32>();
1552 let bytes = block
1553 .get(at..at + size_of::<u32>())
1554 .ok_or_else(|| invalid("global dictionary rank block is short of codes"))?;
1555 let code = u32::from_le_bytes(bytes.try_into().expect("four bytes"));
1556 if code as usize >= self.len() {
1557 return Err(invalid("global dictionary order names a code it does not have"));
1558 }
1559 Ok(code)
1560 }
1561
1562 fn code_ranks(&self) -> Option<&[u32]> {
1563 if self.ranks == 0 || self.ranks != self.len() {
1567 return None;
1568 }
1569 self.code_ranks
1570 .get_or_init(|| {
1571 let mut ranks = vec![u32::MAX; self.ranks];
1572 for first in (0..self.ranks).step_by(TEXT_RANK_BLOCK) {
1575 let (block, _) = self.rank_parts(first).ok()?;
1576 let heads = block.len() / RANK_ENTRY * size_of::<u64>();
1577 let codes = block.get(heads..)?;
1578 for (within, entry) in codes.chunks_exact(size_of::<u32>()).enumerate() {
1579 let code = u32::from_le_bytes(entry.try_into().ok()?) as usize;
1580 *ranks.get_mut(code)? = u32::try_from(first + within).ok()?;
1581 }
1582 }
1583 if ranks.contains(&u32::MAX) {
1584 return None;
1585 }
1586 Some(ranks)
1587 })
1588 .as_deref()
1589 }
1590
1591 fn footprint(&self) -> usize {
1592 self.offsets.capacity() * size_of::<u32>()
1593 + self
1594 .code_ranks
1595 .get()
1596 .and_then(Option::as_ref)
1597 .map_or(0, |ranks| ranks.capacity() * size_of::<u32>())
1598 + self.rank_hashes.capacity() * size_of::<u64>()
1599 + self.rank_blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1600 + self
1601 .rank_blocks
1602 .iter()
1603 .filter_map(OnceLock::get)
1604 .filter_map(|result| result.as_ref().ok())
1605 .map(Vec::capacity)
1606 .sum::<usize>()
1607 + self.payload_extents.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1608 + self.hashes.capacity() * size_of::<u64>()
1609 + self
1610 .payload_extents
1611 .iter()
1612 .filter_map(OnceLock::get)
1613 .filter_map(|result| result.as_ref().ok())
1614 .map(Vec::capacity)
1615 .sum::<usize>()
1616 + self.crossing.capacity() * size_of::<CrossingCache>()
1617 + self
1618 .crossing
1619 .iter()
1620 .filter_map(OnceLock::get)
1621 .map(|block| {
1622 block.len() * size_of::<OnceLock<Result<Vec<u8>>>>()
1623 + block
1624 .iter()
1625 .filter_map(OnceLock::get)
1626 .filter_map(|result| result.as_ref().ok())
1627 .map(Vec::capacity)
1628 .sum::<usize>()
1629 })
1630 .sum::<usize>()
1631 }
1632}
1633
1634fn places(table: &Table) -> Result<Vec<Place>> {
1636 let mut places = Vec::with_capacity(table.stripes.len().saturating_mul(STRIPE_PARTS));
1637 for (at, stripe) in table.stripes.iter().enumerate() {
1638 let index = u32::try_from(at).map_err(|_| invalid("too many stripes"))?;
1639 for (part, &rows) in stripe.parts.iter().enumerate() {
1640 places.push(Place {
1641 stripe: index,
1642 part: u32::try_from(part).map_err(|_| invalid("too many parts in a stripe"))?,
1643 rows,
1644 });
1645 }
1646 }
1647 Ok(places)
1648}
1649
1650fn read_index(file: &File, stripe: &Stripe, column: usize) -> Result<Vec<PartSpan>> {
1655 let parts = stripe.parts.len();
1656 let section = index_section(parts)?;
1657 let at = column.checked_mul(section).ok_or_else(|| invalid("index page offset overflow"))?;
1658 let end = at.checked_add(section).ok_or_else(|| invalid("index page offset overflow"))?;
1659 if end > stripe.index.length as usize {
1660 return Err(invalid("index page is shorter than its columns"));
1661 }
1662 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1663 let mut bytes = vec![0; section];
1664 let offset = stripe
1665 .index
1666 .offset
1667 .checked_add(at as u64)
1668 .ok_or_else(|| invalid("index page offset overflow"))?;
1669 read_at(file, offset, &mut bytes)?;
1670 let entries = section - size_of::<u64>();
1671 let stored = u64::from_le_bytes(bytes[entries..].try_into().expect("eight bytes"));
1672 if checksum(&bytes[..entries]) != stored {
1673 return Err(invalid(&format!(
1676 "index page section checksum differs, column {column} of {parts} parts at {offset}, \
1677 wanted {stored:016x} and got {:016x}",
1678 checksum(&bytes[..entries]),
1679 )));
1680 }
1681 let mut spans = Vec::with_capacity(parts);
1682 let mut start = 0_usize;
1683 for part in 0..parts {
1684 let at = part * INDEX_ENTRY;
1685 let length = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four bytes")) as usize;
1686 let hash = u64::from_le_bytes(bytes[at + 4..at + 12].try_into().expect("eight bytes"));
1687 spans.push(PartSpan { start, length, hash });
1688 start = start.checked_add(length).ok_or_else(|| invalid("column page length overflow"))?;
1689 }
1690 if start != page.length as usize {
1691 return Err(invalid("column page length differs from its index"));
1692 }
1693 Ok(spans)
1694}
1695
1696fn part_bytes(page: &[u8], span: PartSpan) -> Result<&[u8]> {
1698 let end = span.start.checked_add(span.length).ok_or_else(|| invalid("part range overflow"))?;
1699 page.get(span.start..end).ok_or_else(|| invalid("part exceeds its column page"))
1700}
1701
1702fn remember(cached: &mut Cached, held: &CachedColumn, kept: usize) {
1707 if let Some(slot) = cached.index.get_mut(held.stripe) {
1708 if slot.is_none() {
1709 *slot = Some(Arc::clone(&held.index));
1710 }
1711 }
1712 let Some(page) = held.page.clone() else { return };
1713 let Some(slot) = cached.pages.get_mut(held.stripe) else { return };
1714 if slot.is_none() {
1715 cached.order.push_back(held.stripe);
1716 }
1717 *slot = Some(page);
1718 while cached.order.len() > kept.max(1) {
1719 let Some(oldest) = cached.order.pop_front() else { break };
1720 if let Some(slot) = cached.pages.get_mut(oldest) {
1721 *slot = None;
1722 }
1723 }
1724}
1725
1726impl Reader {
1727 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1733 let mut file = File::open(path).map_err(io)?;
1734 let size = file.metadata().map_err(io)?.len();
1735 if size < HEADER {
1736 return Err(invalid("file is shorter than its header"));
1737 }
1738 let mut header = [0; HEADER as usize];
1739 file.read_exact(&mut header).map_err(io)?;
1740 let mut opening = Opening { reads: 1, bytes: HEADER };
1741 let version = u32::from_le_bytes([header[8], header[9], header[10], header[11]]);
1742 if &header[..8] != MAGIC {
1747 return Err(invalid("the header does not begin with a rudb native magic"));
1748 }
1749 if version != FORMAT {
1750 return Err(invalid(&format!(
1751 "the file is format {version} and this build reads format {FORMAT}, so it has to \
1752 be written again"
1753 )));
1754 }
1755 let mut selected = None;
1756 for start in [16, 16 + SLOT_BYTES] {
1757 let slot = Slot::read(&header[start..start + SLOT_BYTES]);
1758 if slot.generation == 0 || slot.length == 0 || slot.length as usize > MAX_DIRECTORY {
1759 continue;
1760 }
1761 let Some(end) = slot.offset.checked_add(u64::from(slot.length)) else { continue };
1762 if slot.offset < HEADER || end > size {
1763 continue;
1764 }
1765 let mut bytes = vec![0; slot.length as usize];
1766 file.seek(SeekFrom::Start(slot.offset)).map_err(io)?;
1767 file.read_exact(&mut bytes).map_err(io)?;
1768 opening.reads += 1;
1769 opening.bytes += u64::from(slot.length);
1770 if checksum(&bytes) == slot.hash
1771 && selected
1772 .as_ref()
1773 .is_none_or(|(old, _): &(Slot, Vec<u8>)| old.generation < slot.generation)
1774 {
1775 selected = Some((slot, bytes));
1776 }
1777 }
1778 let (slot, bytes) =
1779 selected.ok_or_else(|| invalid("no committed directory slot is valid"))?;
1780 let table = decode_directory(&bytes, size)?;
1781 let places = places(&table)?;
1782 let dictionaries = (0..table.fields.len()).map(|_| OnceLock::new()).collect();
1783 let stripes = table.stripes.len();
1784 let cache = (0..table.fields.len())
1785 .map(|_| {
1786 Mutex::new(Cached {
1787 pages: (0..stripes).map(|_| None).collect(),
1788 index: (0..stripes).map(|_| None).collect(),
1789 ..Cached::default()
1790 })
1791 })
1792 .collect::<Vec<_>>();
1793 let sieves = (0..table.fields.len())
1794 .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
1795 .collect();
1796 Ok(Self {
1797 file: Arc::new(file),
1798 table: Arc::new(table),
1799 dictionaries: Arc::new(dictionaries),
1800 sieves: Arc::new(sieves),
1801 places: Arc::new(places),
1802 cache: Arc::new(cache),
1803 pages: Arc::new(AtomicUsize::new(0)),
1804 indexes: Arc::new(AtomicUsize::new(0)),
1805 kept: Arc::new(AtomicUsize::new(CACHED_STRIPES_PER_COLUMN)),
1806 size,
1807 directory: u64::from(slot.length),
1808 opening,
1809 })
1810 }
1811
1812 #[must_use]
1819 pub fn reads(&self) -> Reads {
1820 Reads {
1821 opening: self.opening,
1822 pages: self.pages.load(Atomic::Relaxed),
1823 indexes: self.indexes.load(Atomic::Relaxed),
1824 }
1825 }
1826
1827 #[must_use]
1832 pub fn layout(&self) -> Layout {
1833 let table = &self.table;
1834 let stripes = table.stripes.as_slice();
1835 let columns = table
1836 .fields
1837 .iter()
1838 .enumerate()
1839 .map(|(at, field)| ColumnLayout {
1840 name: field.name.clone(),
1841 kind: field.ty.to_string(),
1842 pages: sum(stripes.iter().map(|stripe| span_bytes(&stripe.pages, at))),
1843 memberships: sum(stripes.iter().map(|stripe| page_bytes(&stripe.memberships, at))),
1844 sieves: sum(stripes.iter().map(|stripe| page_bytes(&stripe.sieves, at))),
1845 dictionary: page_bytes(&table.dictionaries, at),
1846 })
1847 .collect();
1848 Layout {
1849 file: self.size,
1850 rows: table.rows,
1851 stripes: stripes.len(),
1852 parts: self.places.len(),
1853 columns,
1854 indexes: sum(stripes.iter().map(|stripe| u64::from(stripe.index.length))),
1855 directory: self.directory,
1856 header: HEADER,
1857 }
1858 }
1859
1860 #[must_use]
1862 pub fn parts(&self) -> usize {
1863 self.places.len()
1864 }
1865
1866 #[must_use]
1873 pub fn stripe_parts(&self) -> Vec<std::ops::Range<usize>> {
1874 let mut runs = Vec::with_capacity(self.table.stripes.len());
1875 let mut start = 0;
1876 for stripe in &self.table.stripes {
1877 let end = start + stripe.parts.len();
1878 runs.push(start..end);
1879 start = end;
1880 }
1881 runs
1882 }
1883
1884 pub fn keep_stripes(&self, stripes: usize) {
1891 self.kept.fetch_max(stripes, Atomic::Relaxed);
1892 }
1893
1894 #[must_use]
1896 pub fn part_rows(&self, at: usize) -> usize {
1897 self.places.get(at).map_or(0, |place| place.rows as usize)
1898 }
1899
1900 #[must_use]
1902 pub fn table(&self) -> &Table {
1903 &self.table
1904 }
1905
1906 pub fn top_frequencies(&self, column: usize, top: usize) -> Result<Option<Vec<(Value, u64)>>> {
1915 let field = self
1916 .table
1917 .fields
1918 .get(column)
1919 .ok_or_else(|| invalid("frequency column index out of range"))?;
1920 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1921 return Ok(None);
1922 };
1923 if top == 0 || summary.entries.len() < top {
1924 return Ok(None);
1925 }
1926 let boundary = summary.entries[top - 1].count;
1927 if boundary <= summary.omitted_max {
1928 return Ok(None);
1929 }
1930 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
1931 }
1932
1933 pub fn exact_frequencies(&self, column: usize) -> Result<Option<Vec<(Value, u64)>>> {
1953 let field = self
1954 .table
1955 .fields
1956 .get(column)
1957 .ok_or_else(|| invalid("frequency column index out of range"))?;
1958 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1959 return Ok(None);
1960 };
1961 if summary.omitted_max > 0 {
1962 return Ok(None);
1963 }
1964 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
1965 }
1966
1967 fn decode_frequencies(
1969 &self,
1970 column: usize,
1971 ty: &LogicalType,
1972 entries: &[FrequencyEntry],
1973 ) -> Result<Vec<(Value, u64)>> {
1974 let dictionary = if *ty == LogicalType::Varchar { self.dictionary(column)? } else { None };
1975 let mut out = Vec::with_capacity(entries.len());
1976 for entry in entries {
1977 let value = match entry.value {
1978 FrequencyValue::Null => Value::Null,
1979 FrequencyValue::Integer(value) => match *ty {
1980 LogicalType::TinyInt => Value::TinyInt(
1981 i8::try_from(value)
1982 .map_err(|_| invalid("frequency TINYINT is out of range"))?,
1983 ),
1984 LogicalType::UTinyInt => Value::UTinyInt(
1985 u8::try_from(value)
1986 .map_err(|_| invalid("frequency UTINYINT is out of range"))?,
1987 ),
1988 LogicalType::USmallInt => Value::USmallInt(
1989 u16::try_from(value)
1990 .map_err(|_| invalid("frequency USMALLINT is out of range"))?,
1991 ),
1992 LogicalType::UInteger => Value::UInteger(
1993 u32::try_from(value)
1994 .map_err(|_| invalid("frequency UINTEGER is out of range"))?,
1995 ),
1996 LogicalType::UBigInt => Value::UBigInt(
1997 u64::try_from(value)
1998 .map_err(|_| invalid("frequency UBIGINT is out of range"))?,
1999 ),
2000 LogicalType::SmallInt => Value::SmallInt(
2001 i16::try_from(value)
2002 .map_err(|_| invalid("frequency SMALLINT is out of range"))?,
2003 ),
2004 LogicalType::Integer => Value::Integer(
2005 i32::try_from(value)
2006 .map_err(|_| invalid("frequency INTEGER is out of range"))?,
2007 ),
2008 LogicalType::BigInt => Value::BigInt(
2009 i64::try_from(value)
2010 .map_err(|_| invalid("frequency BIGINT is out of range"))?,
2011 ),
2012 LogicalType::Date => Value::Date(
2013 i32::try_from(value)
2014 .map_err(|_| invalid("frequency DATE is out of range"))?,
2015 ),
2016 LogicalType::Timestamp => Value::Timestamp(
2017 i64::try_from(value)
2018 .map_err(|_| invalid("frequency TIMESTAMP is out of range"))?,
2019 ),
2020 _ => return Err(invalid("integer frequency belongs to another type")),
2021 },
2022 FrequencyValue::Code(code) => dictionary
2023 .as_ref()
2024 .ok_or_else(|| invalid("frequency code has no dictionary"))?
2025 .try_value_at(code as usize)?,
2026 };
2027 out.push((value, entry.count));
2028 }
2029 Ok(out)
2030 }
2031
2032 pub fn frequency_occurrences(&self, column: usize) -> Result<Option<FrequencyOccurrences>> {
2042 self.table
2043 .fields
2044 .get(column)
2045 .ok_or_else(|| invalid("frequency column index out of range"))?;
2046 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2047 return Ok(None);
2048 };
2049 if summary.ordinals.is_empty() {
2050 return Ok(None);
2051 }
2052 Ok(Some(FrequencyOccurrences {
2053 omitted_max: summary.omitted_max,
2054 ordinals: summary.ordinals.clone(),
2055 }))
2056 }
2057
2058 pub fn distinct_values(&self, column: usize) -> Result<Option<u64>> {
2078 if self.null_count(column)? > 0 {
2079 return Ok(None);
2080 }
2081 Ok(self.dictionary(column)?.map(|dictionary| dictionary.len() as u64))
2082 }
2083
2084 pub fn null_count(&self, column: usize) -> Result<u64> {
2095 if column >= self.table.fields.len() {
2096 return Err(invalid("null count column index out of range"));
2097 }
2098 let mut nulls = 0_u64;
2099 for stripe in &self.table.stripes {
2100 let range = stripe
2101 .zone
2102 .column(column)
2103 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2104 nulls = nulls
2105 .checked_add(range.nulls as u64)
2106 .ok_or_else(|| invalid("null count overflow"))?;
2107 }
2108 Ok(nulls)
2109 }
2110
2111 pub fn text_extremes(&self, column: usize) -> Result<Option<(Value, Value)>> {
2126 if self.null_count(column)? > 0 {
2127 return Ok(None);
2128 }
2129 let Some(dictionary) = self.dictionary(column)? else { return Ok(None) };
2130 let Some(ranks) = dictionary.ranks() else { return Ok(None) };
2131 if ranks == 0 {
2132 return Ok(None);
2133 }
2134 let low = text_at_rank(&dictionary, 0)?;
2135 let high = text_at_rank(&dictionary, ranks - 1)?;
2136 Ok(Some((low, high)))
2137 }
2138
2139 pub fn exact_extremes(&self, column: usize) -> Result<Option<(Bound, Bound)>> {
2162 if column >= self.table.fields.len() {
2163 return Err(invalid("extremes column index out of range"));
2164 }
2165 let mut low: Option<Bound> = None;
2166 let mut high: Option<Bound> = None;
2167 for stripe in &self.table.stripes {
2168 let range = stripe
2169 .zone
2170 .column(column)
2171 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2172 if !range.exact {
2173 return Ok(None);
2174 }
2175 let (Some(small), Some(large)) = (range.low.as_ref(), range.high.as_ref()) else {
2180 if stripe.rows > range.nulls {
2181 return Ok(None);
2182 }
2183 continue;
2184 };
2185 low = Some(low.map_or_else(|| small.clone(), |held| held.smaller(small.clone())));
2186 high = Some(high.map_or_else(|| large.clone(), |held| held.larger(large.clone())));
2187 }
2188 Ok(low.zip(high))
2189 }
2190
2191 pub fn exact_sum(&self, column: usize) -> Result<Option<(i128, u64)>> {
2204 if column >= self.table.fields.len() {
2205 return Err(invalid("sum column index out of range"));
2206 }
2207 let mut total = 0_i128;
2208 let mut rows = 0_u64;
2209 for stripe in &self.table.stripes {
2210 let range = stripe
2211 .zone
2212 .column(column)
2213 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2214 let Some(part) = range.sum else { return Ok(None) };
2215 let Some(sum) = total.checked_add(part) else { return Ok(None) };
2216 total = sum;
2217 rows = rows.saturating_add(stripe.rows as u64 - range.nulls as u64);
2218 }
2219 Ok(Some((total, rows)))
2220 }
2221
2222 fn dictionary(&self, column: usize) -> Result<Option<Arc<Vector>>> {
2223 let Some(page) = self.table.dictionaries[column] else { return Ok(None) };
2224 if let Some(dictionary) = self.dictionaries[column].get() {
2225 return Ok(Some(Arc::clone(dictionary)));
2226 }
2227 let dictionary = Arc::new(open_global_dictionary(
2228 Arc::clone(&self.file),
2229 page,
2230 &self.table.fields[column].ty,
2231 )?);
2232 let _ = self.dictionaries[column].set(Arc::clone(&dictionary));
2233 Ok(Some(self.dictionaries[column].get().map_or(dictionary, Arc::clone)))
2234 }
2235
2236 pub fn read(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2245 self.read_impl(part, columns, true)
2246 }
2247
2248 pub fn read_sparse(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2258 self.read_impl(part, columns, false)
2259 }
2260
2261 pub fn skips_codes(&self, part: usize, column: usize, candidates: &[u32]) -> Result<bool> {
2268 if candidates.is_empty() {
2269 return Ok(true);
2270 }
2271 if candidates.windows(2).any(|pair| pair[0] >= pair[1]) {
2272 return Err(Error::internal("native code candidates are not sorted and unique"));
2273 }
2274 let stripe = self.stripe_of(part)?;
2275 let Some(page) = stripe.memberships.get(column).copied().flatten() else {
2276 return Ok(false);
2277 };
2278 let mut bytes = vec![0; page.length as usize];
2279 read_at(&self.file, page.offset, &mut bytes)?;
2280 if checksum(&bytes) != page.hash {
2281 return Err(invalid("membership page checksum differs"));
2282 }
2283 let codes = decode_membership(&bytes)?;
2284 let mut left = 0;
2285 let mut right = 0;
2286 while left < codes.len() && right < candidates.len() {
2287 match codes[left].cmp(&candidates[right]) {
2288 Ordering::Less => left += 1,
2289 Ordering::Greater => right += 1,
2290 Ordering::Equal => return Ok(false),
2291 }
2292 }
2293 Ok(true)
2294 }
2295
2296 fn stripe_of(&self, part: usize) -> Result<&Stripe> {
2297 let place = self.places.get(part).ok_or_else(|| invalid("part index out of range"))?;
2298 self.table
2299 .stripes
2300 .get(place.stripe as usize)
2301 .ok_or_else(|| invalid("stripe index out of range"))
2302 }
2303
2304 fn held(&self, at: usize, stripe: &Stripe, column: usize, whole: bool) -> Result<CachedColumn> {
2321 let cache = self.cache.get(column).ok_or_else(|| invalid("column index out of range"))?;
2322 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2323 let known = cached.index.get(at).and_then(Clone::clone);
2324 let page = cached.pages.get(at).and_then(Clone::clone);
2325 if let Some(index) = known.clone() {
2326 if !whole || page.is_some() {
2327 return Ok(CachedColumn { stripe: at, index, page });
2328 }
2329 }
2330 if cached.loading.contains(&at) {
2331 drop(cached);
2332 if let Some(index) = known {
2336 return Ok(CachedColumn { stripe: at, index, page: None });
2337 }
2338 let held = self.page_of(stripe, column, at, false, None)?;
2339 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2340 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2341 return Ok(held);
2342 }
2343 cached.loading.push(at);
2344 drop(cached);
2345
2346 let read = self.page_of(stripe, column, at, whole, known);
2347
2348 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2352 if let Some(position) = cached.loading.iter().position(|loading| *loading == at) {
2353 cached.loading.remove(position);
2354 }
2355 let held = read?;
2356 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2357 Ok(held)
2358 }
2359
2360 fn page_of(
2366 &self,
2367 stripe: &Stripe,
2368 column: usize,
2369 at: usize,
2370 whole: bool,
2371 known: Option<Arc<Vec<PartSpan>>>,
2372 ) -> Result<CachedColumn> {
2373 let index = match known {
2374 Some(index) => index,
2375 None => {
2376 self.indexes.fetch_add(1, Atomic::Relaxed);
2377 Arc::new(read_index(&self.file, stripe, column)?)
2378 }
2379 };
2380 let page = if whole {
2381 self.pages.fetch_add(1, Atomic::Relaxed);
2382 let span = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2383 let mut bytes = vec![0; span.length as usize];
2384 read_at(&self.file, span.offset, &mut bytes)?;
2385 Some(Arc::new(bytes))
2386 } else {
2387 None
2388 };
2389 Ok(CachedColumn { stripe: at, index, page })
2390 }
2391
2392 fn read_impl(&self, at: usize, columns: &[usize], whole: bool) -> Result<Chunk> {
2393 let place = *self.places.get(at).ok_or_else(|| invalid("part index out of range"))?;
2394 let index = place.stripe as usize;
2395 let stripe =
2396 self.table.stripes.get(index).ok_or_else(|| invalid("stripe index out of range"))?;
2397 let rows = place.rows as usize;
2398 let mut picked = Vec::with_capacity(columns.len());
2399 for &column in columns {
2400 let field = self
2401 .table
2402 .fields
2403 .get(column)
2404 .ok_or_else(|| invalid("column index out of range"))?;
2405 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2406 let held = self.held(index, stripe, column, whole)?;
2407 let span = *held
2408 .index
2409 .get(place.part as usize)
2410 .ok_or_else(|| invalid("part index out of range"))?;
2411 let owned;
2412 let bytes = match &held.page {
2413 Some(held) => part_bytes(held, span)?,
2414 None => {
2415 let offset = page
2416 .offset
2417 .checked_add(span.start as u64)
2418 .ok_or_else(|| invalid("part range overflow"))?;
2419 let mut bytes = vec![0; span.length];
2420 read_at(&self.file, offset, &mut bytes)?;
2421 owned = bytes;
2422 &owned
2423 }
2424 };
2425 if checksum(bytes) != span.hash {
2426 return Err(invalid(&format!(
2427 "column page checksum differs, column {column} part {} at {}+{} of {} bytes, \
2428 wanted {:016x} and got {:016x}",
2429 place.part,
2430 page.offset,
2431 span.start,
2432 span.length,
2433 span.hash,
2434 checksum(bytes),
2435 )));
2436 }
2437 let dictionary = self.dictionary(column)?;
2438 picked.push(decode(&field.ty, rows, bytes, dictionary)?);
2439 }
2440 Chunk::with_rows(picked, rows)
2441 }
2442
2443 #[must_use]
2453 pub fn skips(&self, part: usize, probes: &[Probe]) -> bool {
2454 let Some(place) = self.places.get(part).copied() else { return false };
2455 let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
2456 if stripe.zone.skips(probes) {
2457 return true;
2458 }
2459 probes.iter().any(|probe| self.sifted(place, probe))
2460 }
2461
2462 fn sifted(&self, place: Place, probe: &Probe) -> bool {
2468 if probe.op != Op::Equal {
2469 return false;
2470 }
2471 match self.stripe_sieves(place.stripe as usize, probe.column) {
2472 Some(sieves) => sieves
2473 .get(place.part as usize)
2474 .and_then(Option::as_ref)
2475 .is_some_and(|sieve| sieve.excludes(&probe.value)),
2476 None => false,
2477 }
2478 }
2479
2480 fn stripe_sieves(&self, stripe: usize, column: usize) -> Option<&[Option<Sieve>]> {
2487 let slot = self.sieves.get(column)?.get(stripe)?;
2488 if let Some(held) = slot.get() {
2489 return Some(held);
2490 }
2491 let page = self.table.stripes.get(stripe)?.sieves.get(column).copied().flatten()?;
2492 let mut bytes = vec![0; page.length as usize];
2493 read_at(&self.file, page.offset, &mut bytes).ok()?;
2494 if checksum(&bytes) != page.hash {
2495 return None;
2496 }
2497 let sieves = Arc::new(decode_sieves(&bytes).ok()?);
2498 let _ = slot.set(sieves);
2499 slot.get().map(|held| held.as_slice())
2500 }
2501}
2502
2503fn text_at_rank(dictionary: &Vector, rank: usize) -> Result<Value> {
2505 let code = dictionary.code_at_rank(rank)? as usize;
2506 let text = dictionary
2507 .try_text_at(code)?
2508 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
2509 Ok(Value::Varchar(text.into()))
2510}
2511
2512#[cfg(unix)]
2517fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2518 use std::os::unix::fs::FileExt;
2519 while !bytes.is_empty() {
2520 let written = file.write_at(bytes, offset).map_err(io)?;
2521 if written == 0 {
2522 return Err(invalid("a write to the native file wrote nothing"));
2523 }
2524 offset += written as u64;
2525 bytes = &bytes[written..];
2526 }
2527 Ok(())
2528}
2529
2530#[cfg(windows)]
2532fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2533 use std::os::windows::fs::FileExt;
2534 while !bytes.is_empty() {
2535 let written = file.seek_write(bytes, offset).map_err(io)?;
2536 if written == 0 {
2537 return Err(invalid("a write to the native file wrote nothing"));
2538 }
2539 offset += written as u64;
2540 bytes = &bytes[written..];
2541 }
2542 Ok(())
2543}
2544
2545#[cfg(not(any(unix, windows)))]
2547fn write_at(file: &File, offset: u64, bytes: &[u8]) -> Result<()> {
2548 use std::io::Write;
2549 let mut file = file.try_clone().map_err(io)?;
2550 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2551 file.write_all(bytes).map_err(io)
2552}
2553
2554#[cfg(unix)]
2564fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2565 use std::os::unix::fs::FileExt;
2566 while !bytes.is_empty() {
2567 let read = file.read_at(bytes, offset).map_err(io)?;
2568 if read == 0 {
2569 return Err(invalid("column page ends before its declared length"));
2570 }
2571 offset += read as u64;
2572 bytes = &mut bytes[read..];
2573 }
2574 Ok(())
2575}
2576
2577#[cfg(windows)]
2583fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2584 use std::os::windows::fs::FileExt;
2585 while !bytes.is_empty() {
2586 let read = file.seek_read(bytes, offset).map_err(io)?;
2587 if read == 0 {
2588 return Err(invalid("column page ends before its declared length"));
2589 }
2590 offset += read as u64;
2591 bytes = &mut bytes[read..];
2592 }
2593 Ok(())
2594}
2595
2596#[cfg(not(any(unix, windows)))]
2601fn read_at(file: &File, offset: u64, bytes: &mut [u8]) -> Result<()> {
2602 let mut file = file.try_clone().map_err(io)?;
2603 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2604 file.read_exact(bytes).map_err(io)
2605}
2606
2607fn type_tag(ty: &LogicalType) -> Result<u8> {
2608 match ty {
2609 LogicalType::SmallInt => Ok(1),
2610 LogicalType::Integer => Ok(2),
2611 LogicalType::BigInt => Ok(3),
2612 LogicalType::Varchar => Ok(4),
2613 LogicalType::Date => Ok(5),
2614 LogicalType::Timestamp => Ok(6),
2615 LogicalType::Boolean => Ok(7),
2616 LogicalType::TinyInt => Ok(8),
2617 LogicalType::UTinyInt => Ok(9),
2618 LogicalType::USmallInt => Ok(10),
2619 LogicalType::UInteger => Ok(11),
2620 LogicalType::UBigInt => Ok(12),
2621 _ => Err(Error::not_implemented(format!("native storage for {ty}"))),
2622 }
2623}
2624
2625fn tag_type(tag: u8) -> Result<LogicalType> {
2626 match tag {
2627 1 => Ok(LogicalType::SmallInt),
2628 2 => Ok(LogicalType::Integer),
2629 3 => Ok(LogicalType::BigInt),
2630 4 => Ok(LogicalType::Varchar),
2631 5 => Ok(LogicalType::Date),
2632 6 => Ok(LogicalType::Timestamp),
2633 7 => Ok(LogicalType::Boolean),
2634 8 => Ok(LogicalType::TinyInt),
2635 9 => Ok(LogicalType::UTinyInt),
2636 10 => Ok(LogicalType::USmallInt),
2637 11 => Ok(LogicalType::UInteger),
2638 12 => Ok(LogicalType::UBigInt),
2639 _ => Err(invalid("column type tag is unknown")),
2640 }
2641}
2642
2643fn put_u16(out: &mut Vec<u8>, value: u16) {
2644 out.extend_from_slice(&value.to_le_bytes());
2645}
2646fn put_u32(out: &mut Vec<u8>, value: u32) {
2647 out.extend_from_slice(&value.to_le_bytes());
2648}
2649fn put_u64(out: &mut Vec<u8>, value: u64) {
2650 out.extend_from_slice(&value.to_le_bytes());
2651}
2652fn put_var_u64(out: &mut Vec<u8>, mut value: u64) {
2653 while value >= 0x80 {
2654 out.push((value as u8 & 0x7f) | 0x80);
2655 value >>= 7;
2656 }
2657 out.push(value as u8);
2658}
2659
2660fn frequency_order(left: FrequencyValue, right: FrequencyValue) -> Ordering {
2661 match (left, right) {
2662 (FrequencyValue::Null, FrequencyValue::Null) => Ordering::Equal,
2663 (FrequencyValue::Null, _) => Ordering::Less,
2664 (_, FrequencyValue::Null) => Ordering::Greater,
2665 (FrequencyValue::Integer(left), FrequencyValue::Integer(right)) => left.cmp(&right),
2666 (FrequencyValue::Code(left), FrequencyValue::Code(right)) => left.cmp(&right),
2667 (FrequencyValue::Integer(_), FrequencyValue::Code(_)) => Ordering::Less,
2668 (FrequencyValue::Code(_), FrequencyValue::Integer(_)) => Ordering::Greater,
2669 }
2670}
2671
2672fn code_frequency(dictionary: &GlobalDictionary) -> FrequencySummary {
2673 let mut entries = dictionary
2674 .counts
2675 .iter()
2676 .enumerate()
2677 .filter(|(_, count)| **count != 0)
2678 .map(|(code, &count)| FrequencyEntry { value: FrequencyValue::Code(code as u32), count })
2679 .collect::<Vec<_>>();
2680 if dictionary.nulls != 0 {
2681 entries.push(FrequencyEntry { value: FrequencyValue::Null, count: dictionary.nulls });
2682 }
2683 entries.sort_unstable_by(|left, right| {
2684 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
2685 });
2686 let omitted_max = entries.get(FREQUENCY_ENTRIES).map_or(0, |entry| entry.count);
2687 entries.truncate(FREQUENCY_ENTRIES);
2688 FrequencySummary { entries, omitted_max, ordinals: Vec::new() }
2689}
2690
2691fn encode_directory(table: &Table) -> Result<Vec<u8>> {
2692 let mut out = DIRECTORY.to_vec();
2693 let name = table.name.as_bytes();
2694 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
2695 out.extend_from_slice(name);
2696 put_u16(&mut out, u16::try_from(table.fields.len()).map_err(|_| invalid("too many columns"))?);
2697 for field in &table.fields {
2698 let name = field.name.as_bytes();
2699 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?);
2700 out.extend_from_slice(name);
2701 out.push(type_tag(&field.ty)?);
2702 out.push(u8::from(field.not_null));
2703 }
2704 for dictionary in &table.dictionaries {
2705 match dictionary {
2706 None => out.push(0),
2707 Some(page) => {
2708 out.push(1);
2709 put_u64(&mut out, page.offset);
2710 put_u32(&mut out, page.length);
2711 put_u64(&mut out, page.hash);
2712 }
2713 }
2714 }
2715 put_u64(&mut out, u64::try_from(table.rows).map_err(|_| invalid("row count overflow"))?);
2716 put_u32(&mut out, u32::try_from(table.stripes.len()).map_err(|_| invalid("too many stripes"))?);
2717 for stripe in &table.stripes {
2718 put_u32(
2719 &mut out,
2720 u32::try_from(stripe.parts.len()).map_err(|_| invalid("too many parts in a stripe"))?,
2721 );
2722 for &rows in &stripe.parts {
2723 put_u32(&mut out, rows);
2724 }
2725 put_u64(&mut out, stripe.index.offset);
2726 put_u32(&mut out, stripe.index.length);
2727 for page in &stripe.pages {
2728 put_u64(&mut out, page.offset);
2729 put_u32(&mut out, page.length);
2730 }
2731 for (field, membership) in table.fields.iter().zip(&stripe.memberships) {
2732 if field.ty != LogicalType::Varchar {
2733 continue;
2734 }
2735 let page =
2736 membership.ok_or_else(|| invalid("string page has no code membership index"))?;
2737 put_u64(&mut out, page.offset);
2738 put_u32(&mut out, page.length);
2739 put_u64(&mut out, page.hash);
2740 }
2741 for sieve in &stripe.sieves {
2742 match sieve {
2743 None => out.push(0),
2744 Some(page) => {
2745 out.push(1);
2746 put_u64(&mut out, page.offset);
2747 put_u32(&mut out, page.length);
2748 put_u64(&mut out, page.hash);
2749 }
2750 }
2751 }
2752 for range in stripe.zone.columns() {
2753 put_bound(&mut out, range.low.as_ref())?;
2754 put_bound(&mut out, range.high.as_ref())?;
2755 put_u32(
2756 &mut out,
2757 u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?,
2758 );
2759 out.push(u8::from(range.exact));
2760 match range.sum {
2761 None => out.push(0),
2762 Some(total) => {
2763 out.push(1);
2764 out.extend_from_slice(&total.to_le_bytes());
2765 }
2766 }
2767 }
2768 }
2769 out.extend_from_slice(FREQUENCIES);
2770 put_u16(
2771 &mut out,
2772 u16::try_from(table.frequencies.len())
2773 .map_err(|_| invalid("too many frequency columns"))?,
2774 );
2775 for summary in &table.frequencies {
2776 let Some(summary) = summary else {
2777 out.push(0);
2778 continue;
2779 };
2780 out.push(1);
2781 put_u64(&mut out, summary.omitted_max);
2782 put_u32(
2783 &mut out,
2784 u32::try_from(summary.entries.len())
2785 .map_err(|_| invalid("too many frequency entries"))?,
2786 );
2787 for entry in &summary.entries {
2788 match entry.value {
2789 FrequencyValue::Null => out.push(0),
2790 FrequencyValue::Integer(value) => {
2791 out.push(1);
2792 out.extend_from_slice(&value.to_le_bytes());
2793 }
2794 FrequencyValue::Code(value) => {
2795 out.push(2);
2796 put_u32(&mut out, value);
2797 }
2798 }
2799 put_u64(&mut out, entry.count);
2800 }
2801 put_u32(
2802 &mut out,
2803 u32::try_from(summary.ordinals.len())
2804 .map_err(|_| invalid("too many frequency ordinals"))?,
2805 );
2806 let mut previous = 0_u64;
2807 for (at, &ordinal) in summary.ordinals.iter().enumerate() {
2808 let delta = if at == 0 {
2809 ordinal
2810 } else {
2811 ordinal
2812 .checked_sub(previous)
2813 .ok_or_else(|| invalid("frequency ordinals are not ordered"))?
2814 };
2815 if at != 0 && delta == 0 {
2816 return Err(invalid("frequency ordinals are not unique"));
2817 }
2818 put_var_u64(&mut out, delta);
2819 previous = ordinal;
2820 }
2821 }
2822 Ok(out)
2823}
2824
2825struct Cursor<'a> {
2826 bytes: &'a [u8],
2827 at: usize,
2828}
2829impl<'a> Cursor<'a> {
2830 fn take(&mut self, len: usize) -> Result<&'a [u8]> {
2831 let end = self.at.checked_add(len).ok_or_else(|| invalid("directory offset overflow"))?;
2832 let bytes =
2833 self.bytes.get(self.at..end).ok_or_else(|| invalid("directory is truncated"))?;
2834 self.at = end;
2835 Ok(bytes)
2836 }
2837 fn u8(&mut self) -> Result<u8> {
2838 Ok(self.take(1)?[0])
2839 }
2840 fn u16(&mut self) -> Result<u16> {
2841 Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("two bytes")))
2842 }
2843 fn u32(&mut self) -> Result<u32> {
2844 Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("four bytes")))
2845 }
2846 fn u64(&mut self) -> Result<u64> {
2847 Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("eight bytes")))
2848 }
2849 fn var_u64(&mut self) -> Result<u64> {
2850 let mut value = 0_u64;
2851 for shift in (0..=63).step_by(7) {
2852 let byte = self.u8()?;
2853 let part = u64::from(byte & 0x7f);
2854 if shift == 63 && part > 1 {
2855 return Err(invalid("frequency ordinal varint overflows"));
2856 }
2857 value |= part << shift;
2858 if byte & 0x80 == 0 {
2859 return Ok(value);
2860 }
2861 }
2862 Err(invalid("frequency ordinal varint is too long"))
2863 }
2864 fn bound(&mut self) -> Result<Option<Bound>> {
2865 Ok(match self.u8()? {
2866 0 => None,
2867 1 => Some(Bound::Int(i128::from_le_bytes(
2868 self.take(16)?.try_into().expect("sixteen bytes"),
2869 ))),
2870 2 => Some(Bound::Real(f64::from_le_bytes(
2871 self.take(8)?.try_into().expect("eight bytes"),
2872 ))),
2873 3 => {
2874 let length = self.u32()? as usize;
2875 Some(Bound::Bytes(self.take(length)?.to_vec()))
2876 }
2877 _ => return Err(invalid("bound tag differs")),
2878 })
2879 }
2880 fn text(&mut self) -> Result<String> {
2881 let len = self.u16()? as usize;
2882 String::from_utf8(self.take(len)?.to_vec()).map_err(|_| invalid("name is not UTF-8"))
2883 }
2884}
2885
2886fn decode_directory(bytes: &[u8], size: u64) -> Result<Table> {
2887 let mut cur = Cursor { bytes, at: 0 };
2888 if cur.take(8)? != DIRECTORY {
2889 return Err(invalid("directory magic differs"));
2890 }
2891 let name = cur.text()?;
2892 let width = cur.u16()? as usize;
2893 let mut fields = Vec::with_capacity(width);
2894 for _ in 0..width {
2895 let name = cur.text()?;
2896 let ty = tag_type(cur.u8()?)?;
2897 let not_null = match cur.u8()? {
2898 0 => false,
2899 1 => true,
2900 _ => return Err(invalid("nullability flag differs")),
2901 };
2902 fields.push(Field { name, ty, not_null });
2903 }
2904 let mut dictionaries = Vec::with_capacity(width);
2905 for _ in 0..width {
2906 dictionaries.push(match cur.u8()? {
2907 0 => None,
2908 1 => {
2909 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2910 let end = page
2911 .offset
2912 .checked_add(u64::from(page.length))
2913 .ok_or_else(|| invalid("dictionary page offset overflow"))?;
2914 if page.offset < HEADER || end > size {
2919 return Err(invalid("dictionary page range is outside the file"));
2920 }
2921 Some(page)
2922 }
2923 _ => return Err(invalid("dictionary page tag differs")),
2924 });
2925 }
2926 let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
2927 let count = cur.u32()? as usize;
2928 let mut stripes = Vec::with_capacity(count);
2929 let mut total = 0_usize;
2930 for _ in 0..count {
2931 let count = cur.u32()? as usize;
2932 if count == 0 || count > STRIPE_PARTS {
2933 return Err(invalid("stripe part count is outside its bound"));
2934 }
2935 let mut parts = Vec::with_capacity(count);
2936 let mut stripe_rows = 0_usize;
2937 for _ in 0..count {
2938 let rows = cur.u32()?;
2939 if rows == 0 {
2940 return Err(invalid("empty part"));
2941 }
2942 parts.push(rows);
2943 stripe_rows = stripe_rows
2944 .checked_add(rows as usize)
2945 .ok_or_else(|| invalid("stripe row count overflow"))?;
2946 }
2947 total =
2948 total.checked_add(stripe_rows).ok_or_else(|| invalid("stripe row count overflow"))?;
2949 let index = Span { offset: cur.u64()?, length: cur.u32()? };
2950 let section = index_section(count)?;
2951 let wanted = section
2952 .checked_mul(width)
2953 .and_then(|bytes| u32::try_from(bytes).ok())
2954 .ok_or_else(|| invalid("index page length overflow"))?;
2955 let end = index
2956 .offset
2957 .checked_add(u64::from(index.length))
2958 .ok_or_else(|| invalid("index page offset overflow"))?;
2959 if index.offset < HEADER || end > size || index.length != wanted {
2960 return Err(invalid("index page range is outside the file"));
2961 }
2962 let mut pages = Vec::with_capacity(width);
2963 for _ in 0..width {
2964 let offset = cur.u64()?;
2965 let length = cur.u32()?;
2966 let end = offset
2967 .checked_add(u64::from(length))
2968 .ok_or_else(|| invalid("page offset overflow"))?;
2969 if offset < HEADER || end > size || length as usize > MAX_PAGE {
2970 return Err(invalid("page range is outside the file"));
2971 }
2972 pages.push(Span { offset, length });
2973 }
2974 let mut memberships = vec![None; width];
2975 for (column, field) in fields.iter().enumerate() {
2976 if field.ty != LogicalType::Varchar {
2977 continue;
2978 }
2979 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2980 let end = page
2981 .offset
2982 .checked_add(u64::from(page.length))
2983 .ok_or_else(|| invalid("membership page offset overflow"))?;
2984 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
2985 return Err(invalid("membership page range is outside the file"));
2986 }
2987 memberships[column] = Some(page);
2988 }
2989 let mut sieves = vec![None; width];
2990 for sieve in sieves.iter_mut().take(width) {
2991 match cur.u8()? {
2992 0 => continue,
2993 1 => {}
2994 _ => return Err(invalid("a sieve page has an unknown tag")),
2995 }
2996 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2997 let end = page
2998 .offset
2999 .checked_add(u64::from(page.length))
3000 .ok_or_else(|| invalid("sieve page offset overflow"))?;
3001 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3002 return Err(invalid("sieve page range is outside the file"));
3003 }
3004 *sieve = Some(page);
3005 }
3006 let mut ranges = Vec::with_capacity(width);
3007 for _ in 0..width {
3008 let low = cur.bound()?;
3009 let high = cur.bound()?;
3010 let nulls = cur.u32()? as usize;
3011 if nulls > stripe_rows {
3012 return Err(invalid("null count exceeds stripe rows"));
3013 }
3014 let exact = cur.u8()? != 0;
3015 let sum = match cur.u8()? {
3016 0 => None,
3017 1 => Some(i128::from_le_bytes(
3018 cur.take(16)?.try_into().map_err(|_| invalid("a stripe sum is truncated"))?,
3019 )),
3020 _ => return Err(invalid("a stripe sum has an unknown tag")),
3021 };
3022 ranges.push(Range { low, high, nulls, exact, sum });
3023 }
3024 stripes.push(Stripe {
3025 rows: stripe_rows,
3026 parts,
3027 index,
3028 pages,
3029 memberships,
3030 sieves,
3031 zone: Zone::from_ranges(ranges),
3032 });
3033 }
3034 if total != rows {
3035 return Err(invalid("table row count differs from stripes"));
3036 }
3037 let frequencies = if cur.at == bytes.len() {
3038 vec![None; width]
3039 } else {
3040 if cur.take(8)? != FREQUENCIES {
3041 return Err(invalid("directory extension magic differs"));
3042 }
3043 if cur.u16()? as usize != width {
3044 return Err(invalid("frequency column count differs"));
3045 }
3046 let mut frequencies = Vec::with_capacity(width);
3047 for field in &fields {
3048 let summary = match cur.u8()? {
3049 0 => None,
3050 1 => {
3051 let omitted_max = cur.u64()?;
3052 let count = cur.u32()? as usize;
3053 if count > FREQUENCY_ENTRIES {
3054 return Err(invalid("frequency entry count exceeds its bound"));
3055 }
3056 let mut entries = Vec::with_capacity(count);
3057 for _ in 0..count {
3059 let value = match cur.u8()? {
3060 0 => FrequencyValue::Null,
3061 1 => FrequencyValue::Integer(i128::from_le_bytes(
3062 cur.take(16)?.try_into().expect("sixteen bytes"),
3063 )),
3064 2 => FrequencyValue::Code(cur.u32()?),
3065 _ => return Err(invalid("frequency value tag differs")),
3066 };
3067 let valid = matches!(
3068 (&field.ty, value),
3069 (_, FrequencyValue::Null)
3070 | (LogicalType::Varchar, FrequencyValue::Code(_))
3071 | (
3072 LogicalType::TinyInt
3073 | LogicalType::SmallInt
3074 | LogicalType::Integer
3075 | LogicalType::BigInt
3076 | LogicalType::UTinyInt
3077 | LogicalType::USmallInt
3078 | LogicalType::UInteger
3079 | LogicalType::UBigInt
3080 | LogicalType::Date
3081 | LogicalType::Timestamp,
3082 FrequencyValue::Integer(_),
3083 )
3084 );
3085 if !valid {
3086 return Err(invalid("frequency value does not match its column"));
3087 }
3088 let count = cur.u64()?;
3089 if count == 0 || count > rows as u64 {
3090 return Err(invalid("frequency count is outside the table"));
3091 }
3092 entries.push(FrequencyEntry { value, count });
3093 }
3094 if entries.windows(2).any(|pair| pair[0].count < pair[1].count) {
3095 return Err(invalid("frequency entries are not descending"));
3096 }
3097 let ordinals = {
3098 let ordinal_count = cur.u32()? as usize;
3099 if ordinal_count > FREQUENCY_ORDINALS || ordinal_count > rows {
3100 return Err(invalid("frequency ordinal count exceeds its bound"));
3101 }
3102 let mut ordinals = Vec::with_capacity(ordinal_count);
3103 let mut previous = 0_u64;
3104 for at in 0..ordinal_count {
3105 let delta = cur.var_u64()?;
3106 if at != 0 && delta == 0 {
3107 return Err(invalid("frequency ordinals are not increasing"));
3108 }
3109 let ordinal = if at == 0 {
3110 delta
3111 } else {
3112 previous
3113 .checked_add(delta)
3114 .ok_or_else(|| invalid("frequency ordinal overflows"))?
3115 };
3116 if ordinal >= rows as u64 {
3117 return Err(invalid("frequency ordinal is outside the table"));
3118 }
3119 ordinals.push(ordinal);
3120 previous = ordinal;
3121 }
3122 ordinals
3123 };
3124 Some(FrequencySummary { entries, omitted_max, ordinals })
3125 }
3126 _ => return Err(invalid("frequency summary tag differs")),
3127 };
3128 frequencies.push(summary);
3129 }
3130 frequencies
3131 };
3132 if cur.at != bytes.len() {
3133 return Err(invalid("directory has trailing bytes"));
3134 }
3135 Ok(Table { name, fields, stripes, rows, dictionaries, frequencies })
3136}
3137
3138fn put_bound(out: &mut Vec<u8>, bound: Option<&Bound>) -> Result<()> {
3139 match bound {
3140 None => out.push(0),
3141 Some(Bound::Int(value)) => {
3142 out.push(1);
3143 out.extend_from_slice(&value.to_le_bytes());
3144 }
3145 Some(Bound::Real(value)) => {
3146 out.push(2);
3147 out.extend_from_slice(&value.to_le_bytes());
3148 }
3149 Some(Bound::Bytes(value)) => {
3150 out.push(3);
3151 put_u32(out, u32::try_from(value.len()).map_err(|_| invalid("bound length overflow"))?);
3152 out.extend_from_slice(value);
3153 }
3154 }
3155 Ok(())
3156}
3157
3158#[derive(Debug)]
3175struct Codes;
3176
3177impl chooser::Chooser for Codes {
3178 fn name(&self) -> &'static str {
3179 "codes"
3180 }
3181
3182 fn narrow_strings(
3183 &self,
3184 _values: &[&[u8]],
3185 offered: &[string::Kind],
3186 _depth: u8,
3187 ) -> Vec<string::Kind> {
3188 offered.to_vec()
3191 }
3192
3193 fn narrow_integers(
3194 &self,
3195 _values: &[i64],
3196 offered: &[integer::Kind],
3197 depth: u8,
3198 ) -> Vec<integer::Kind> {
3199 let keep: &[integer::Kind] = if depth == 0 {
3200 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Rle]
3201 } else {
3202 &[integer::Kind::Constant, integer::Kind::Packed]
3203 };
3204 let narrowed: Vec<integer::Kind> =
3205 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3206 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3209 }
3210}
3211
3212#[derive(Debug)]
3222struct Fixed;
3223
3224impl chooser::Chooser for Fixed {
3225 fn name(&self) -> &'static str {
3226 "fixed"
3227 }
3228
3229 fn narrow_strings(
3230 &self,
3231 _values: &[&[u8]],
3232 offered: &[string::Kind],
3233 _depth: u8,
3234 ) -> Vec<string::Kind> {
3235 offered.to_vec()
3236 }
3237
3238 fn narrow_integers(
3239 &self,
3240 _values: &[i64],
3241 offered: &[integer::Kind],
3242 depth: u8,
3243 ) -> Vec<integer::Kind> {
3244 let keep: &[integer::Kind] = if depth == 0 {
3245 &[
3246 integer::Kind::Constant,
3247 integer::Kind::Packed,
3248 integer::Kind::Delta,
3249 integer::Kind::Rle,
3250 integer::Kind::Sparse,
3251 ]
3252 } else {
3253 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Delta]
3254 };
3255 let narrowed: Vec<integer::Kind> =
3256 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3257 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3258 }
3259}
3260
3261fn widened(data: &Data) -> Option<Vec<i64>> {
3268 match data {
3269 Data::Int8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3270 Data::UInt8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3271 Data::Int16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3272 Data::UInt16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3273 Data::Int32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3274 Data::UInt32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3275 Data::Int64(values) => Some(values.to_vec()),
3276 _ => None,
3277 }
3278}
3279
3280fn narrowed(ty: &LogicalType, values: Vec<i64>) -> Result<Data> {
3285 fn fit<T: TryFrom<i64>>(values: &[i64]) -> Result<Vec<T>> {
3286 values
3287 .iter()
3288 .map(|value| T::try_from(*value).map_err(|_| invalid("page value is not of its type")))
3289 .collect()
3290 }
3291 Ok(match ty {
3292 LogicalType::TinyInt => Data::Int8(fit::<i8>(&values)?.into()),
3293 LogicalType::UTinyInt => Data::UInt8(fit::<u8>(&values)?.into()),
3294 LogicalType::SmallInt => Data::Int16(fit::<i16>(&values)?.into()),
3295 LogicalType::USmallInt => Data::UInt16(fit::<u16>(&values)?.into()),
3296 LogicalType::Integer | LogicalType::Date => Data::Int32(fit::<i32>(&values)?.into()),
3297 LogicalType::UInteger => Data::UInt32(fit::<u32>(&values)?.into()),
3298 LogicalType::BigInt | LogicalType::Timestamp => Data::Int64(values.into()),
3299 _ => return Err(invalid("cascade codec belongs to a page that is not integers")),
3300 })
3301}
3302
3303fn plain_width(ty: &LogicalType) -> Option<usize> {
3306 Some(match ty {
3307 LogicalType::TinyInt | LogicalType::UTinyInt => 1,
3308 LogicalType::SmallInt | LogicalType::USmallInt => 2,
3309 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date => 4,
3310 LogicalType::BigInt | LogicalType::Timestamp => 8,
3311 _ => return None,
3312 })
3313}
3314
3315fn cascaded(
3321 flat: &Vector,
3322 ty: &LogicalType,
3323 packed: Option<&Packed<'_>>,
3324) -> Result<Option<Vec<u8>>> {
3325 let (Some(width), Some(data)) = (plain_width(ty), flat.data()) else { return Ok(None) };
3326 let Some(values) = widened(data) else { return Ok(None) };
3327 let plain = values.len().saturating_mul(width);
3328 let best = match packed {
3329 Some(packed) => plain.min(21 + size_of_val(packed.words())),
3331 None => plain,
3332 };
3333 let out = integer::encode_with(&values, &Fixed)?;
3334 Ok((out.len() < best).then_some(out))
3335}
3336
3337fn encoded_codes(codes: &[u32]) -> Result<Option<Vec<u8>>> {
3349 let wide: Vec<i64> = codes.iter().map(|code| i64::from(*code)).collect();
3350 let coded = integer::encode_with(&wide, &Codes)?;
3351 let plain = codes.len().saturating_mul(size_of::<u32>());
3352 Ok((coded.len() < plain).then_some(coded))
3353}
3354
3355fn encode(
3356 vector: &Vector,
3357 global: Option<&mut GlobalDictionary>,
3358) -> Result<(Vec<u8>, Option<Vec<u32>>)> {
3359 let ty = vector.logical_type();
3360 let flat = vector.flatten()?;
3362 let mut out = Vec::new();
3363 let mut global_codes = None;
3364 if let Some(global) = global {
3365 let mut codes = Vec::with_capacity(flat.len());
3366 for row in 0..flat.len() {
3367 let text = flat.text_at(row).unwrap_or("");
3368 let code = global.code(text)?;
3369 global.observe(code, flat.is_null_at(row))?;
3370 codes.push(code);
3371 }
3372 global_codes = Some(codes);
3373 }
3374 let membership = global_codes.as_deref().map(unique_codes);
3375 let dictionary = if global_codes.is_none() && ty == &LogicalType::Varchar {
3376 string_dictionary(&flat)?
3377 } else {
3378 None
3379 };
3380 let packed_vector = if dictionary.is_none() && global_codes.is_none() {
3381 Some(flat.bit_packed()?)
3382 } else {
3383 None
3384 };
3385 let packed = packed_vector.as_ref().and_then(Vector::packed_parts);
3386 let coded = match global_codes.as_deref() {
3387 Some(codes) => encoded_codes(codes)?,
3388 None => None,
3389 };
3390 let cascade = if dictionary.is_none() && global_codes.is_none() {
3394 cascaded(&flat, ty, packed.as_ref())?
3395 } else {
3396 None
3397 };
3398 out.push(if coded.is_some() {
3399 4
3400 } else if cascade.is_some() {
3401 5
3402 } else if global_codes.is_some() {
3403 3
3404 } else if dictionary.is_some() {
3405 1
3406 } else if packed.is_some() {
3407 2
3408 } else {
3409 0
3410 });
3411 let nulls = flat.validity();
3412 let flag = match nulls {
3413 Validity::AllValid => 0,
3414 Validity::AllInvalid => 1,
3415 Validity::Mask(_) => 2,
3416 };
3417 out.push(flag);
3418 if flag == 2 {
3419 for group in (0..vector.len()).step_by(8) {
3420 let mut bits = 0_u8;
3421 for bit in 0..8 {
3422 if group + bit < vector.len() && !flat.is_null_at(group + bit) {
3423 bits |= 1 << bit;
3424 }
3425 }
3426 out.push(bits);
3427 }
3428 }
3429 if let Some(coded) = coded {
3430 out.extend_from_slice(&coded);
3431 return Ok((out, membership));
3432 }
3433 if let Some(cascade) = cascade {
3434 out.extend_from_slice(&cascade);
3435 return Ok((out, membership));
3436 }
3437 if let Some(codes) = global_codes {
3438 for code in codes {
3439 put_u32(&mut out, code);
3440 }
3441 return Ok((out, membership));
3442 }
3443 if let Some(dictionary) = dictionary {
3444 out.extend_from_slice(&dictionary);
3445 return Ok((out, membership));
3446 }
3447 if let Some(packed) = packed {
3448 if packed.offset() != 0 {
3449 return Err(invalid("writer received a sliced packed vector"));
3450 }
3451 out.push(u8::try_from(packed.width()).map_err(|_| invalid("packed width overflow"))?);
3452 out.extend_from_slice(&packed.base().to_le_bytes());
3453 put_u32(
3454 &mut out,
3455 u32::try_from(packed.words().len()).map_err(|_| invalid("too many packed words"))?,
3456 );
3457 for word in packed.words() {
3458 put_u64(&mut out, *word);
3459 }
3460 return Ok((out, membership));
3461 }
3462 let data = flat.data().ok_or_else(|| invalid("scalar column did not flatten"))?;
3463 match (ty, data) {
3464 (LogicalType::TinyInt, Data::Int8(values)) => {
3465 for value in &**values {
3466 out.extend_from_slice(&value.to_le_bytes());
3467 }
3468 }
3469 (LogicalType::UTinyInt, Data::UInt8(values)) => {
3470 for value in &**values {
3471 out.extend_from_slice(&value.to_le_bytes());
3472 }
3473 }
3474 (LogicalType::SmallInt, Data::Int16(values)) => {
3475 for value in &**values {
3476 out.extend_from_slice(&value.to_le_bytes());
3477 }
3478 }
3479 (LogicalType::USmallInt, Data::UInt16(values)) => {
3480 for value in &**values {
3481 out.extend_from_slice(&value.to_le_bytes());
3482 }
3483 }
3484 (LogicalType::UInteger, Data::UInt32(values)) => {
3485 for value in &**values {
3486 out.extend_from_slice(&value.to_le_bytes());
3487 }
3488 }
3489 (LogicalType::UBigInt, Data::UInt64(values)) => {
3490 for value in &**values {
3491 out.extend_from_slice(&value.to_le_bytes());
3492 }
3493 }
3494 (LogicalType::Integer | LogicalType::Date, Data::Int32(values)) => {
3495 for value in &**values {
3496 out.extend_from_slice(&value.to_le_bytes());
3497 }
3498 }
3499 (LogicalType::BigInt | LogicalType::Timestamp, Data::Int64(values)) => {
3500 for value in &**values {
3501 out.extend_from_slice(&value.to_le_bytes());
3502 }
3503 }
3504 (LogicalType::Boolean, Data::Bool(values)) => {
3505 for value in &**values {
3506 out.push(u8::from(*value));
3507 }
3508 }
3509 (LogicalType::Varchar, Data::Varlen(values)) => {
3510 let mut bytes = Vec::new();
3511 put_u32(&mut out, 0);
3512 for row in 0..vector.len() {
3513 let value = values.bytes(row).ok_or_else(|| invalid("string view is invalid"))?;
3514 bytes.extend_from_slice(value);
3515 put_u32(
3516 &mut out,
3517 u32::try_from(bytes.len())
3518 .map_err(|_| invalid("string payload exceeds 4GiB"))?,
3519 );
3520 }
3521 out.extend_from_slice(&bytes);
3522 }
3523 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
3524 }
3525 Ok((out, membership))
3526}
3527
3528fn put_varint(out: &mut Vec<u8>, mut value: u32) {
3529 while value >= 0x80 {
3530 out.push((value as u8 & 0x7f) | 0x80);
3531 value >>= 7;
3532 }
3533 out.push(value as u8);
3534}
3535
3536fn unique_codes(codes: &[u32]) -> Vec<u32> {
3538 let mut unique = codes.to_vec();
3539 unique.sort_unstable();
3540 unique.dedup();
3541 unique
3542}
3543
3544fn merged_codes(lists: Vec<Vec<u32>>) -> Vec<u32> {
3550 let mut lists = lists;
3551 while lists.len() > 1 {
3552 let mut next = Vec::with_capacity(lists.len().div_ceil(2));
3553 for pair in lists.chunks(2) {
3554 match pair {
3555 [left, right] => next.push(merged_pair(left, right)),
3556 [only] => next.push(only.clone()),
3557 _ => {}
3558 }
3559 }
3560 lists = next;
3561 }
3562 lists.pop().unwrap_or_default()
3563}
3564
3565fn merged_pair(left: &[u32], right: &[u32]) -> Vec<u32> {
3566 let mut out = Vec::with_capacity(left.len().saturating_add(right.len()));
3567 let mut at = 0;
3568 let mut to = 0;
3569 while at < left.len() && to < right.len() {
3570 match left[at].cmp(&right[to]) {
3571 Ordering::Less => {
3572 out.push(left[at]);
3573 at += 1;
3574 }
3575 Ordering::Greater => {
3576 out.push(right[to]);
3577 to += 1;
3578 }
3579 Ordering::Equal => {
3580 out.push(left[at]);
3581 at += 1;
3582 to += 1;
3583 }
3584 }
3585 }
3586 out.extend_from_slice(&left[at..]);
3587 out.extend_from_slice(&right[to..]);
3588 out
3589}
3590
3591fn merged_range(ranges: impl Iterator<Item = Range>) -> Range {
3596 let mut merged = Range::default();
3597 let mut first = true;
3598 for range in ranges {
3599 merged.nulls = merged.nulls.saturating_add(range.nulls);
3600 merged.sum = match (merged.sum.take(), range.sum) {
3604 (Some(held), Some(next)) if !first => held.checked_add(next),
3605 (_, next) if first => next,
3606 _ => None,
3607 };
3608 merged.exact = if first { range.exact } else { merged.exact && range.exact };
3609 if first {
3610 merged.low = range.low;
3611 merged.high = range.high;
3612 first = false;
3613 continue;
3614 }
3615 merged.low = match (merged.low.take(), range.low) {
3616 (Some(held), Some(next)) => Some(held.smaller(next)),
3617 _ => None,
3618 };
3619 merged.high = match (merged.high.take(), range.high) {
3620 (Some(held), Some(next)) => Some(held.larger(next)),
3621 _ => None,
3622 };
3623 }
3624 merged
3625}
3626
3627fn encode_sieves<'a>(sieves: impl Iterator<Item = &'a Option<Sieve>>) -> Result<Vec<u8>> {
3633 let held: Vec<&Option<Sieve>> = sieves.collect();
3634 let mut out = Vec::new();
3635 put_u32(
3636 &mut out,
3637 u32::try_from(held.len()).map_err(|_| invalid("too many parts in a stripe"))?,
3638 );
3639 for sieve in &held {
3640 let length = sieve.as_ref().map_or(0, Sieve::len);
3641 put_u32(&mut out, u32::try_from(length).map_err(|_| invalid("sieve length overflow"))?);
3642 }
3643 for sieve in held.into_iter().flatten() {
3645 out.extend_from_slice(&sieve.to_bytes());
3646 }
3647 Ok(out)
3648}
3649
3650fn decode_sieves(bytes: &[u8]) -> Result<Vec<Option<Sieve>>> {
3656 let parts = u32::from_le_bytes(
3657 bytes
3658 .get(..4)
3659 .ok_or_else(|| invalid("sieve page is truncated"))?
3660 .try_into()
3661 .map_err(|_| invalid("sieve page is truncated"))?,
3662 ) as usize;
3663 let mut lengths = Vec::with_capacity(parts);
3664 for part in 0..parts {
3665 let at = 4 + part * 4;
3666 let field = bytes.get(at..at + 4).ok_or_else(|| invalid("sieve page is truncated"))?;
3667 lengths.push(u32::from_le_bytes(
3668 field.try_into().map_err(|_| invalid("sieve page is truncated"))?,
3669 ) as usize);
3670 }
3671 let mut at = 4 + parts * 4;
3672 let mut out = Vec::with_capacity(parts);
3673 for length in lengths {
3674 if length == 0 {
3675 out.push(None);
3676 continue;
3677 }
3678 let end = at.checked_add(length).ok_or_else(|| invalid("sieve page is truncated"))?;
3679 let field = bytes.get(at..end).ok_or_else(|| invalid("sieve page is truncated"))?;
3680 out.push(Sieve::from_bytes(field));
3681 at = end;
3682 }
3683 if at != bytes.len() {
3684 return Err(invalid("sieve page has trailing bytes"));
3685 }
3686 Ok(out)
3687}
3688
3689fn encode_membership(unique: &[u32]) -> Vec<u8> {
3695 let mut out = Vec::with_capacity(unique.len().saturating_mul(2).saturating_add(5));
3696 put_varint(&mut out, u32::try_from(unique.len()).unwrap_or(u32::MAX));
3697 let mut previous = 0;
3698 for (at, &code) in unique.iter().enumerate() {
3699 put_varint(&mut out, if at == 0 { code } else { code - previous });
3700 previous = code;
3701 }
3702 out
3703}
3704
3705fn take_varint(bytes: &[u8], at: &mut usize) -> Result<u32> {
3706 let mut value = 0_u32;
3707 for shift in (0..35).step_by(7) {
3708 let byte = *bytes.get(*at).ok_or_else(|| invalid("membership varint is truncated"))?;
3709 *at += 1;
3710 let part = u32::from(byte & 0x7f);
3711 if shift == 28 && part > 0x0f {
3712 return Err(invalid("membership varint overflow"));
3713 }
3714 value = value
3715 .checked_add(
3716 part.checked_shl(shift).ok_or_else(|| invalid("membership varint overflow"))?,
3717 )
3718 .ok_or_else(|| invalid("membership varint overflow"))?;
3719 if byte & 0x80 == 0 {
3720 return Ok(value);
3721 }
3722 }
3723 Err(invalid("membership varint is too long"))
3724}
3725
3726fn decode_membership(bytes: &[u8]) -> Result<Vec<u32>> {
3727 let mut at = 0;
3728 let count = take_varint(bytes, &mut at)? as usize;
3729 let mut codes = Vec::with_capacity(count);
3730 let mut previous = 0_u32;
3731 for index in 0..count {
3732 let delta = take_varint(bytes, &mut at)?;
3733 let code = if index == 0 {
3734 delta
3735 } else {
3736 previous.checked_add(delta).ok_or_else(|| invalid("membership code overflow"))?
3737 };
3738 if index > 0 && code <= previous {
3739 return Err(invalid("membership codes are not increasing"));
3740 }
3741 codes.push(code);
3742 previous = code;
3743 }
3744 if at != bytes.len() {
3745 return Err(invalid("membership page has trailing bytes"));
3746 }
3747 Ok(codes)
3748}
3749
3750fn string_dictionary(vector: &Vector) -> Result<Option<Vec<u8>>> {
3751 let mut by_text = HashMap::new();
3752 let mut values = Vec::new();
3753 let mut codes = Vec::with_capacity(vector.len());
3754 let mut plain_bytes = 0_usize;
3755 for row in 0..vector.len() {
3756 let text = vector.text_at(row).unwrap_or("");
3757 plain_bytes = plain_bytes.saturating_add(text.len());
3758 let code = match by_text.get(text) {
3759 Some(&code) => code,
3760 None => {
3761 let code = u32::try_from(values.len())
3762 .map_err(|_| invalid("too many dictionary values"))?;
3763 by_text.insert(text, code);
3764 values.push(text);
3765 code
3766 }
3767 };
3768 codes.push(code);
3769 }
3770 let dictionary_bytes = values.iter().map(|value| value.len()).sum::<usize>();
3771 let encoded = 8_usize
3772 .saturating_add((values.len() + 1).saturating_mul(4))
3773 .saturating_add(dictionary_bytes)
3774 .saturating_add(codes.len().saturating_mul(4));
3775 let plain = (vector.len() + 1).saturating_mul(4).saturating_add(plain_bytes);
3776 if encoded >= plain {
3777 return Ok(None);
3778 }
3779 let mut out = Vec::with_capacity(encoded);
3780 put_u32(
3781 &mut out,
3782 u32::try_from(values.len()).map_err(|_| invalid("too many dictionary values"))?,
3783 );
3784 put_u32(
3785 &mut out,
3786 u32::try_from(dictionary_bytes).map_err(|_| invalid("dictionary payload exceeds 4GiB"))?,
3787 );
3788 let mut offset = 0_u32;
3789 put_u32(&mut out, offset);
3790 for value in &values {
3791 offset = offset
3792 .checked_add(
3793 u32::try_from(value.len()).map_err(|_| invalid("dictionary value is too long"))?,
3794 )
3795 .ok_or_else(|| invalid("dictionary payload exceeds 4GiB"))?;
3796 put_u32(&mut out, offset);
3797 }
3798 for value in values {
3799 out.extend_from_slice(value.as_bytes());
3800 }
3801 for code in codes {
3802 put_u32(&mut out, code);
3803 }
3804 Ok(Some(out))
3805}
3806
3807struct EncodedDictionary {
3808 index: Vec<u8>,
3809 ranks: Vec<u8>,
3810 payload: Vec<u8>,
3811}
3812
3813fn head(bytes: &[u8]) -> u64 {
3815 let mut word = [0; 8];
3816 let take = bytes.len().min(8);
3817 word[..take].copy_from_slice(&bytes[..take]);
3818 u64::from_be_bytes(word)
3819}
3820
3821fn rankings(dictionaries: &[Option<GlobalDictionary>]) -> Result<Vec<Vec<(u64, u32)>>> {
3829 let present =
3830 dictionaries.iter().enumerate().filter(|(_, held)| held.is_some()).map(|(at, _)| at);
3831 let present = present.collect::<Vec<_>>();
3832 let mut orders = vec![Vec::new(); dictionaries.len()];
3833 let workers = std::thread::available_parallelism()
3834 .map_or(1, usize::from)
3835 .min(MAX_FREQUENCY_WORKERS)
3836 .min(present.len());
3837 if workers <= 1 {
3838 for at in present {
3839 if let Some(dictionary) = &dictionaries[at] {
3840 orders[at] = dictionary.ranked();
3841 }
3842 }
3843 return Ok(orders);
3844 }
3845 let width = present.len().div_ceil(workers);
3846 let pieces = std::thread::scope(|scope| {
3847 present
3848 .chunks(width)
3849 .map(|columns| {
3850 scope.spawn(|| {
3851 columns
3852 .iter()
3853 .filter_map(|&at| dictionaries[at].as_ref().map(|held| (at, held.ranked())))
3854 .collect::<Vec<_>>()
3855 })
3856 })
3857 .collect::<Vec<_>>()
3858 .into_iter()
3859 .map(|handle| {
3860 handle.join().map_err(|_| Error::internal("a dictionary sort worker panicked"))
3861 })
3862 .collect::<Result<Vec<_>>>()
3863 })?;
3864 for piece in pieces {
3865 for (at, order) in piece {
3866 orders[at] = order;
3867 }
3868 }
3869 Ok(orders)
3870}
3871
3872fn encode_global_dictionary(
3873 dictionary: GlobalDictionary,
3874 order: &[(u64, u32)],
3875) -> Result<EncodedDictionary> {
3876 let values = dictionary.offsets.len() - 1;
3877 if order.len() != values {
3878 return Err(invalid("global dictionary order does not cover its values"));
3879 }
3880 let payload_len = dictionary.payload.len();
3881 let blocks = payload_len.div_ceil(TEXT_PAYLOAD_BLOCK);
3882 let ranks = encode_ranks(order);
3883 let rank_blocks = values.div_ceil(TEXT_RANK_BLOCK);
3884 let mut index = Vec::with_capacity(12 + (values + 1) * 4 + (blocks + rank_blocks) * 8);
3885 put_u32(
3886 &mut index,
3887 u32::try_from(values).map_err(|_| invalid("global dictionary has too many values"))?,
3888 );
3889 put_u32(&mut index, TEXT_PAYLOAD_BLOCK as u32);
3890 put_u32(
3891 &mut index,
3892 u32::try_from(blocks).map_err(|_| invalid("global dictionary has too many blocks"))?,
3893 );
3894 for offset in dictionary.offsets {
3895 put_u32(&mut index, offset);
3896 }
3897 for block in dictionary.payload.chunks(TEXT_PAYLOAD_BLOCK) {
3898 put_u64(&mut index, checksum(block));
3899 }
3900 for block in ranks.chunks(TEXT_RANK_BLOCK * RANK_ENTRY) {
3901 put_u64(&mut index, checksum(block));
3902 }
3903 Ok(EncodedDictionary { index, ranks, payload: dictionary.payload })
3904}
3905
3906fn encode_ranks(order: &[(u64, u32)]) -> Vec<u8> {
3913 let mut out = Vec::with_capacity(order.len() * RANK_ENTRY);
3914 for block in order.chunks(TEXT_RANK_BLOCK) {
3915 for &(head, _) in block {
3916 put_u64(&mut out, head);
3917 }
3918 for &(_, code) in block {
3919 put_u32(&mut out, code);
3920 }
3921 }
3922 out
3923}
3924
3925fn open_global_dictionary(file: Arc<File>, page: Page, ty: &LogicalType) -> Result<Vector> {
3926 if ty != &LogicalType::Varchar {
3927 return Err(invalid("global dictionary belongs to a non-string column"));
3928 }
3929 let mut header = [0; 12];
3930 read_at(&file, page.offset, &mut header)?;
3931 let count = u32::from_le_bytes(header[0..4].try_into().expect("four bytes")) as usize;
3932 let block_size = u32::from_le_bytes(header[4..8].try_into().expect("four bytes")) as usize;
3933 let blocks = u32::from_le_bytes(header[8..12].try_into().expect("four bytes")) as usize;
3934 if block_size != TEXT_PAYLOAD_BLOCK {
3935 return Err(invalid("global dictionary block width differs"));
3936 }
3937 let offset_len = (count + 1)
3938 .checked_mul(4)
3939 .ok_or_else(|| invalid("global dictionary offset count overflow"))?;
3940 let ranks = count;
3945 let rank_blocks = ranks.div_ceil(TEXT_RANK_BLOCK);
3946 let rank_len =
3947 ranks.checked_mul(RANK_ENTRY).ok_or_else(|| invalid("global dictionary rank overflow"))?;
3948 let hash_len = blocks
3949 .checked_add(rank_blocks)
3950 .and_then(|count| count.checked_mul(8))
3951 .ok_or_else(|| invalid("global dictionary block count overflow"))?;
3952 let index_len = 12usize
3953 .checked_add(offset_len)
3954 .and_then(|len| len.checked_add(hash_len))
3955 .ok_or_else(|| invalid("global dictionary header overflow"))?;
3956 let body_len = index_len
3957 .checked_add(rank_len)
3958 .ok_or_else(|| invalid("global dictionary header overflow"))?;
3959 if body_len > page.length as usize {
3960 return Err(invalid("global dictionary offset index exceeds its page"));
3961 }
3962 let mut index = vec![0; index_len];
3963 index[..12].copy_from_slice(&header);
3964 read_at(&file, page.offset + 12, &mut index[12..])?;
3965 if checksum(&index) != page.hash {
3966 return Err(invalid("global dictionary index checksum differs"));
3967 }
3968 let offsets = index[12..12 + offset_len]
3969 .chunks_exact(4)
3970 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
3971 .collect::<Vec<_>>();
3972 let mut hashes = index[12 + offset_len..]
3973 .chunks_exact(8)
3974 .map(|part| u64::from_le_bytes(part.try_into().expect("eight bytes")))
3975 .collect::<Vec<_>>();
3976 let rank_hashes = hashes.split_off(blocks);
3977 let payload_len = page.length as usize - body_len;
3978 if blocks != payload_len.div_ceil(TEXT_PAYLOAD_BLOCK) {
3979 return Err(invalid("global dictionary block count differs from its payload"));
3980 }
3981 if offsets.first() != Some(&0)
3982 || offsets.last().copied().map(|last| last as usize) != Some(payload_len)
3983 || offsets.windows(2).any(|pair| pair[0] > pair[1])
3984 {
3985 return Err(invalid("global dictionary offsets do not bound the payload"));
3986 }
3987 let payload_extents = (0..payload_len.div_ceil(TEXT_PAYLOAD_BLOCK * TEXT_PAYLOAD_EXTENT))
3988 .map(|_| OnceLock::new())
3989 .collect();
3990 let crossing = (0..count.div_ceil(TEXT_CROSSING_BLOCK)).map(|_| OnceLock::new()).collect();
3991 Vector::external_text(
3992 LogicalType::Varchar,
3993 Arc::new(NativeText {
3994 file,
3995 offsets,
3996 ranks,
3997 rank_at: page.offset + index_len as u64,
3998 rank_hashes,
3999 rank_blocks: (0..rank_blocks).map(|_| OnceLock::new()).collect(),
4000 code_ranks: OnceLock::new(),
4001 payload: page.offset + body_len as u64,
4002 payload_len,
4003 hashes,
4004 payload_extents,
4005 crossing,
4006 }),
4007 )
4008}
4009
4010fn decode(
4011 ty: &LogicalType,
4012 rows: usize,
4013 bytes: &[u8],
4014 global: Option<Arc<Vector>>,
4015) -> Result<Vector> {
4016 let mut cur = Cursor { bytes, at: 0 };
4017 let codec = cur.u8()?;
4018 let flag = cur.u8()?;
4019 let validity = match flag {
4020 0 => Validity::AllValid,
4021 1 => Validity::AllInvalid,
4022 2 => {
4023 let mask = cur.take(rows.div_ceil(8))?;
4024 Validity::from_iter(rows, |row| mask[row / 8] >> (row % 8) & 1 == 1)
4025 }
4026 _ => return Err(invalid("page validity tag differs")),
4027 };
4028 if codec == 1 {
4029 if ty != &LogicalType::Varchar {
4030 return Err(invalid("dictionary codec belongs to a non-string page"));
4031 }
4032 let count = cur.u32()? as usize;
4033 let payload_len = cur.u32()? as usize;
4034 let offset_bytes = cur.take(
4035 (count + 1)
4036 .checked_mul(4)
4037 .ok_or_else(|| invalid("dictionary offset count overflow"))?,
4038 )?;
4039 let offsets = offset_bytes
4040 .chunks_exact(4)
4041 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4042 .collect::<Vec<_>>();
4043 let payload = cur.take(payload_len)?.to_vec();
4044 if offsets.first() != Some(&0)
4045 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4046 || offsets.windows(2).any(|pair| pair[0] > pair[1])
4047 {
4048 return Err(invalid("dictionary offsets do not bound the payload"));
4049 }
4050 let mut strings = StringColumn::over(Buffer::from_vec(payload));
4051 for pair in offsets.windows(2) {
4052 strings.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4053 }
4054 let mut codes = Vec::with_capacity(rows);
4055 for _ in 0..rows {
4056 codes.push(cur.u32()?);
4057 }
4058 if codes.iter().any(|code| *code as usize >= count) {
4059 return Err(invalid("dictionary code is out of range"));
4060 }
4061 if cur.at != bytes.len() {
4062 return Err(invalid("dictionary page has trailing bytes"));
4063 }
4064 let dictionary = Vector::flat(LogicalType::Varchar, Data::Varlen(strings))?;
4065 return Ok(Vector::dictionary(codes, dictionary)?.with_validity(validity));
4066 }
4067 if codec == 3 || codec == 4 {
4068 let dictionary = global.ok_or_else(|| invalid("global code page has no dictionary"))?;
4069 let codes = if codec == 4 {
4070 let wide = integer::decode(&bytes[cur.at..])?;
4073 if wide.len() != rows {
4074 return Err(invalid("encoded code page holds the wrong number of rows"));
4075 }
4076 wide.into_iter()
4077 .map(|code| u32::try_from(code).map_err(|_| invalid("code is not a code")))
4078 .collect::<Result<Vec<u32>>>()?
4079 } else {
4080 let mut codes = Vec::with_capacity(rows);
4081 for _ in 0..rows {
4082 codes.push(cur.u32()?);
4083 }
4084 if cur.at != bytes.len() {
4085 return Err(invalid("global code page has trailing bytes"));
4086 }
4087 codes
4088 };
4089 let highest = codes.iter().copied().max();
4090 return Ok(Vector::stable_dictionary_validated(codes, dictionary, highest)?
4091 .with_validity(validity));
4092 }
4093 if codec == 5 {
4094 let values = integer::decode(&bytes[cur.at..])?;
4096 if values.len() != rows {
4097 return Err(invalid("cascade page holds the wrong number of rows"));
4098 }
4099 let data = narrowed(ty, values)?;
4100 return Ok(Vector::flat(ty.clone(), data)?.with_validity(validity));
4101 }
4102 if codec == 2 {
4103 let width = u32::from(cur.u8()?);
4104 let base = i128::from_le_bytes(cur.take(16)?.try_into().expect("sixteen bytes"));
4105 let count = cur.u32()? as usize;
4106 let mut words = Vec::with_capacity(count);
4107 for _ in 0..count {
4108 words.push(cur.u64()?);
4109 }
4110 if cur.at != bytes.len() {
4111 return Err(invalid("packed page has trailing bytes"));
4112 }
4113 return Ok(Vector::packed(ty.clone(), words, width, base, rows)?.with_validity(validity));
4114 }
4115 if codec != 0 {
4116 return Err(invalid("page codec is unknown"));
4117 }
4118 let data = match ty {
4119 LogicalType::TinyInt => {
4120 let values = cur.take(rows)?;
4121 Data::Int8(values.iter().map(|item| *item as i8).collect::<Vec<_>>().into())
4122 }
4123 LogicalType::UTinyInt => Data::UInt8(cur.take(rows)?.to_vec().into()),
4124 LogicalType::SmallInt => {
4125 let values =
4126 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4127 Data::Int16(
4128 values
4129 .chunks_exact(2)
4130 .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
4131 .collect::<Vec<_>>()
4132 .into(),
4133 )
4134 }
4135 LogicalType::USmallInt => {
4136 let values =
4137 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4138 Data::UInt16(
4139 values
4140 .chunks_exact(2)
4141 .map(|item| u16::from_le_bytes(item.try_into().expect("two bytes")))
4142 .collect::<Vec<_>>()
4143 .into(),
4144 )
4145 }
4146 LogicalType::UInteger => {
4147 let values =
4148 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4149 Data::UInt32(
4150 values
4151 .chunks_exact(4)
4152 .map(|item| u32::from_le_bytes(item.try_into().expect("four bytes")))
4153 .collect::<Vec<_>>()
4154 .into(),
4155 )
4156 }
4157 LogicalType::UBigInt => {
4158 let values =
4159 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4160 Data::UInt64(
4161 values
4162 .chunks_exact(8)
4163 .map(|item| u64::from_le_bytes(item.try_into().expect("eight bytes")))
4164 .collect::<Vec<_>>()
4165 .into(),
4166 )
4167 }
4168 LogicalType::Integer | LogicalType::Date => {
4169 let values =
4170 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4171 Data::Int32(
4172 values
4173 .chunks_exact(4)
4174 .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
4175 .collect::<Vec<_>>()
4176 .into(),
4177 )
4178 }
4179 LogicalType::BigInt | LogicalType::Timestamp => {
4180 let values =
4181 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4182 Data::Int64(
4183 values
4184 .chunks_exact(8)
4185 .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
4186 .collect::<Vec<_>>()
4187 .into(),
4188 )
4189 }
4190 LogicalType::Boolean => {
4191 let values = cur.take(rows)?;
4192 if values.iter().any(|value| *value > 1) {
4193 return Err(invalid("boolean page has another value"));
4194 }
4195 Data::Bool(values.iter().map(|value| *value == 1).collect::<Vec<_>>().into())
4196 }
4197 LogicalType::Varchar => {
4198 let offset_bytes = cur
4199 .take((rows + 1).checked_mul(4).ok_or_else(|| invalid("offset count overflow"))?)?;
4200 let offsets = offset_bytes
4201 .chunks_exact(4)
4202 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4203 .collect::<Vec<_>>();
4204 let payload = cur.take(bytes.len() - cur.at)?.to_vec();
4205 if offsets.first() != Some(&0)
4206 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4207 || offsets.windows(2).any(|pair| pair[0] > pair[1])
4208 {
4209 return Err(invalid("string offsets do not bound the payload"));
4210 }
4211 let mut values = StringColumn::over(Buffer::from_vec(payload));
4212 for pair in offsets.windows(2) {
4213 values.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4214 }
4215 Data::Varlen(values)
4216 }
4217 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
4218 };
4219 if cur.at != bytes.len() {
4220 return Err(invalid("page has trailing bytes"));
4221 }
4222 Ok(Vector::flat(ty.clone(), data)?.with_validity(validity))
4223}
4224
4225#[cfg(test)]
4226mod tests {
4227 use std::fs;
4228 use std::io::{Seek, SeekFrom, Write};
4229 use std::path::PathBuf;
4230 use std::time::{SystemTime, UNIX_EPOCH};
4231
4232 use rudb_common::Value;
4233 use rudb_common::bounds::Op;
4234
4235 use super::*;
4236
4237 #[test]
4238 fn checksum_matches_fixed_vectors() {
4239 assert_eq!(checksum(b""), 0xef46_db37_51d8_e999);
4240 assert_eq!(checksum(b"a"), 0xd24e_c4f1_a98c_6e5b);
4241 assert_eq!(checksum(b"abc"), 0x44bc_2cf5_ad77_0999);
4242 }
4243
4244 fn path(label: &str) -> PathBuf {
4245 let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
4246 std::env::temp_dir().join(format!("rudb-native-{label}-{}-{stamp}.rdb", std::process::id()))
4247 }
4248
4249 #[test]
4251 fn a_read_at_an_offset_ignores_where_another_thread_left_the_cursor() {
4252 const SPANS: usize = 64;
4253 const SPAN: usize = 512;
4254 let path = path("positional");
4255 let content: Vec<u8> =
4256 (0..SPANS).flat_map(|span| std::iter::repeat_n(span as u8, SPAN)).collect();
4257 fs::write(&path, &content).expect("the file is written");
4258 let file = Arc::new(File::open(&path).expect("the file opens"));
4259 std::thread::scope(|scope| {
4260 for _ in 0..8 {
4261 let file = Arc::clone(&file);
4262 scope.spawn(move || {
4263 for _ in 0..64 {
4264 for span in 0..SPANS {
4265 let mut bytes = [0_u8; SPAN];
4266 read_at(&file, (span * SPAN) as u64, &mut bytes)
4267 .expect("the span reads");
4268 assert!(
4269 bytes.iter().all(|byte| *byte == span as u8),
4270 "span {span} came back as {}",
4271 bytes[0],
4272 );
4273 }
4274 }
4275 });
4276 }
4277 });
4278 let mut past = [0_u8; SPAN];
4279 let end = (SPANS * SPAN) as u64;
4280 let error = read_at(&file, end, &mut past).expect_err("a read past the end is refused");
4281 assert!(error.message().contains("ends before its declared length"), "{error}");
4282 drop(file);
4283 let _ = fs::remove_file(&path);
4284 }
4285
4286 #[test]
4292 fn a_writer_puts_a_page_where_it_said_it_did_wherever_the_cursor_has_got_to() {
4293 let path = path("cursor");
4294 let mut writer = Writer::create(
4295 &path,
4296 "items",
4297 vec![
4298 Field::required("id", LogicalType::Integer),
4299 Field::new("text", LogicalType::Varchar),
4300 ],
4301 )
4302 .expect("new file");
4303 writer.append(&sample()).expect("first part");
4304 writer.file.seek(SeekFrom::Start(0)).expect("the cursor goes back to the header");
4305 writer.append(&sample()).expect("second part");
4306 writer.file.seek(SeekFrom::Start(1)).expect("and somewhere useless again");
4307 writer.finish().expect("commit");
4308 let reader = Reader::open(&path).expect("reopen from disk");
4309 assert_eq!(reader.table().rows(), 6);
4310 let ids = reader.read(0, &[0]).expect("the integer page reads back");
4311 assert_eq!(ids.value_at(0, 0), Value::Integer(4));
4312 assert_eq!(ids.value_at(2, 0), Value::Integer(-2));
4313 let text = reader.read(1, &[1]).expect("the text page reads back");
4314 assert_eq!(text.value_at(1, 0), Value::Null);
4315 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4316 let end = reader.table().stripes().iter().flat_map(|stripe| {
4319 stripe
4320 .pages
4321 .iter()
4322 .map(|page| page.offset + u64::from(page.length))
4323 .chain(std::iter::once(stripe.index.offset + u64::from(stripe.index.length)))
4324 });
4325 let last = end.fold(HEADER, u64::max);
4326 let directory = fs::metadata(&path).expect("the file is there").len();
4327 assert!(last <= directory, "a page runs to {last} in a file of {directory} bytes");
4328 fs::remove_file(path).expect("remove scratch file");
4329 }
4330
4331 fn sample() -> Chunk {
4332 Chunk::new(vec![
4333 Vector::from_values(
4334 LogicalType::Integer,
4335 &[Value::Integer(4), Value::Integer(9), Value::Integer(-2)],
4336 )
4337 .expect("integers"),
4338 Vector::from_values(
4339 LogicalType::Varchar,
4340 &[
4341 Value::Varchar("alpha".into()),
4342 Value::Null,
4343 Value::Varchar("long text after a slash".into()),
4344 ],
4345 )
4346 .expect("strings"),
4347 ])
4348 .expect("matching rows")
4349 }
4350
4351 fn sample_ids() -> Chunk {
4352 Chunk::new(vec![
4353 Vector::flat(LogicalType::Integer, Data::Int32(vec![7, 8, 9].into()))
4354 .expect("integers"),
4355 ])
4356 .expect("one column")
4357 }
4358
4359 #[test]
4360 fn committed_file_reopens_and_reads_only_requested_columns() {
4361 let path = path("reopen");
4362 let mut writer = Writer::create(
4363 &path,
4364 "items",
4365 vec![
4366 Field::required("id", LogicalType::Integer),
4367 Field::new("text", LogicalType::Varchar),
4368 ],
4369 )
4370 .expect("new file");
4371 writer.append(&sample()).expect("first part");
4372 writer.append(&sample()).expect("second part");
4373 writer.finish().expect("commit");
4374 let reader = Reader::open(&path).expect("reopen from disk");
4375 assert_eq!(reader.table().rows(), 6);
4376 assert_eq!(reader.table().stripes().len(), 1);
4379 assert_eq!(reader.parts(), 2);
4380 assert_eq!(reader.part_rows(0), 3);
4381 assert_eq!(reader.part_rows(1), 3);
4382 let text = reader.read(1, &[1]).expect("only text page");
4383 assert_eq!(text.width(), 1);
4384 assert_eq!(text.value_at(1, 0), Value::Null);
4385 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4386 let sparse = reader.read_sparse(1, &[1]).expect("one part without its whole page");
4387 assert_eq!(sparse.width(), 1);
4388 assert_eq!(sparse.value_at(1, 0), Value::Null);
4389 assert_eq!(sparse.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4390 assert!(!reader.skips_codes(0, 1, &[0]).expect("alpha is in the stripe"));
4391 assert!(!reader.skips_codes(0, 1, &[2]).expect("long text is in the stripe"));
4392 assert!(reader.skips_codes(0, 1, &[3]).expect("unknown code is absent"));
4393 let count = reader.read(0, &[]).expect("no page is needed for count");
4394 assert_eq!(count.len(), 3);
4395 assert!(reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }]));
4396 assert!(!reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(0) }]));
4397 let integers = reader.top_frequencies(0, 1).expect("valid integer synopsis").expect("kept");
4398 assert_eq!(
4399 integers,
4400 vec![(Value::Integer(-2), 2), (Value::Integer(4), 2), (Value::Integer(9), 2),]
4401 );
4402 let strings = reader.top_frequencies(1, 1).expect("valid string synopsis").expect("kept");
4403 assert_eq!(strings.len(), 3);
4404 assert!(strings.contains(&(Value::Null, 2)));
4405 assert!(strings.contains(&(Value::Varchar("alpha".into()), 2)));
4406 assert!(strings.contains(&(Value::Varchar("long text after a slash".into()), 2)));
4407 fs::remove_file(path).expect("remove scratch file");
4408 }
4409
4410 #[test]
4416 fn parts_past_the_stripe_bound_start_a_new_stripe() {
4417 let path = path("stripe-bound");
4418 let mut writer = Writer::create(
4419 &path,
4420 "items",
4421 vec![
4422 Field::required("id", LogicalType::Integer),
4423 Field::new("text", LogicalType::Varchar),
4424 ],
4425 )
4426 .expect("new file");
4427 let parts = STRIPE_PARTS * 2 + 3;
4428 for part in 0..parts {
4429 let id = part as i32;
4430 let chunk = Chunk::new(vec![
4431 Vector::from_values(
4432 LogicalType::Integer,
4433 &[Value::Integer(id), Value::Integer(-id)],
4434 )
4435 .expect("integers"),
4436 Vector::from_values(
4437 LogicalType::Varchar,
4438 &[Value::Varchar(format!("value {part}")), Value::Null],
4439 )
4440 .expect("strings"),
4441 ])
4442 .expect("matching rows");
4443 writer.append(&chunk).expect("one part");
4444 }
4445 writer.finish().expect("commit");
4446
4447 let reader = Reader::open(&path).expect("reopen from disk");
4448 assert_eq!(reader.parts(), parts);
4449 assert_eq!(reader.table().rows(), parts * 2);
4450 assert_eq!(reader.table().stripes().len(), parts.div_ceil(STRIPE_PARTS));
4451 assert_eq!(reader.table().stripes()[0].parts(), STRIPE_PARTS);
4452 assert_eq!(reader.table().stripes()[0].rows(), STRIPE_PARTS * 2);
4453 assert_eq!(reader.table().stripes()[2].parts(), 3);
4454 for part in (0..parts).rev() {
4457 let dense = reader.read(part, &[0, 1]).expect("a whole page read");
4458 let sparse = reader.read_sparse(part, &[0, 1]).expect("one part read");
4459 for chunk in [&dense, &sparse] {
4460 assert_eq!(chunk.len(), 2, "part {part} has its own row count");
4461 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4462 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
4463 assert_eq!(chunk.value_at(0, 1), Value::Varchar(format!("value {part}")));
4464 assert_eq!(chunk.value_at(1, 1), Value::Null);
4465 }
4466 }
4467 let above = [Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }];
4470 assert!(reader.skips(0, &above), "the first stripe stops at 63");
4471 assert!(!reader.skips(STRIPE_PARTS * 2, &above), "the third stripe reaches 130");
4472 fs::remove_file(path).expect("remove scratch file");
4473 }
4474
4475 fn scattered(n: i64) -> i64 {
4477 n.wrapping_mul(-7_046_029_254_386_353_131)
4478 }
4479
4480 #[test]
4486 fn a_part_is_skipped_when_its_sieve_does_not_hold_the_constant() {
4487 let path = path("sieve-skip");
4488 let mut writer =
4489 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
4490 .expect("new file");
4491 let parts = STRIPE_PARTS + 3;
4492 let per_part = 8;
4493 for part in 0..parts {
4494 let held: Vec<Value> = (0..per_part)
4495 .map(|row| Value::BigInt(scattered((part * per_part + row) as i64)))
4496 .collect();
4497 let chunk =
4498 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
4499 .expect("one column");
4500 writer.append(&chunk).expect("one part");
4501 }
4502 writer.finish().expect("commit");
4503
4504 let reader = Reader::open(&path).expect("reopen from disk");
4505 let probe = |value: i64| Probe {
4506 column: 0,
4507 op: Op::Equal,
4508 value: Bound::Int(i128::from(scattered(value))),
4509 };
4510 for wanted in [0_i64, 9, (parts * per_part - 1) as i64] {
4511 let tests = [probe(wanted)];
4512 let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &tests)).collect();
4513 let home = wanted as usize / per_part;
4514 assert_eq!(kept, vec![home], "only the part holding {wanted} is read");
4515 }
4516 let absent = [probe((parts * per_part) as i64 + 1)];
4517 assert!((0..parts).all(|part| reader.skips(part, &absent)), "no part holds it");
4518 let tests = [probe(0)];
4521 assert!(
4522 reader.table().stripes().iter().all(|stripe| !stripe.zone.skips(&tests)),
4523 "the bounds rule out no stripe at all"
4524 );
4525 fs::remove_file(path).expect("remove scratch file");
4526 }
4527
4528 #[test]
4534 fn a_damaged_sieve_page_is_read_through_rather_than_refused() {
4535 let path = path("sieve-damaged");
4536 let mut writer =
4537 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
4538 .expect("new file");
4539 let held: Vec<Value> = (0..8).map(|row| Value::BigInt(scattered(row))).collect();
4540 let chunk =
4541 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
4542 .expect("one column");
4543 writer.append(&chunk).expect("one part");
4544 writer.finish().expect("commit");
4545
4546 let page =
4547 Reader::open(&path).expect("reopen").table.stripes[0].sieves[0].expect("a sieve page");
4548 let mut file = OpenOptions::new().write(true).open(&path).expect("open the sieve page");
4549 file.seek(SeekFrom::Start(page.offset + u64::from(page.length) - 1)).expect("seek");
4550 file.write_all(&[0xff]).expect("damage one byte");
4551 drop(file);
4552
4553 let reader = Reader::open(&path).expect("reopen the damaged file");
4554 let absent =
4555 [Probe { column: 0, op: Op::Equal, value: Bound::Int(i128::from(scattered(99))) }];
4556 assert!(!reader.skips(0, &absent), "a sieve that cannot be read skips nothing");
4557 assert_eq!(reader.read(0, &[0]).expect("the rows are untouched").len(), 8);
4558 fs::remove_file(path).expect("remove scratch file");
4559 }
4560
4561 #[test]
4572 fn workers_that_want_the_same_stripe_read_it_once() {
4573 let path = path("single-flight");
4574 let mut writer =
4575 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4576 .expect("new file");
4577 for part in 0..STRIPE_PARTS {
4578 let id = part as i32;
4579 let chunk = Chunk::new(vec![
4580 Vector::from_values(
4581 LogicalType::Integer,
4582 &[Value::Integer(id), Value::Integer(-id)],
4583 )
4584 .expect("integers"),
4585 ])
4586 .expect("matching rows");
4587 writer.append(&chunk).expect("one part");
4588 }
4589 writer.finish().expect("commit");
4590
4591 let reader = Reader::open(&path).expect("reopen from disk");
4592 assert_eq!(reader.table().stripes().len(), 1, "one stripe is the point of the test");
4593 let barrier = std::sync::Barrier::new(8);
4594 std::thread::scope(|scope| {
4595 for worker in 0..8 {
4596 let reader = &reader;
4597 let barrier = &barrier;
4598 scope.spawn(move || {
4599 barrier.wait();
4600 for part in (worker..STRIPE_PARTS).step_by(8) {
4601 let chunk = reader.read(part, &[0]).expect("a whole page read");
4602 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4603 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
4604 }
4605 });
4606 }
4607 });
4608 assert_eq!(reader.pages.load(Atomic::Relaxed), 1, "one stripe, one page read, whoever won");
4609 fs::remove_file(path).expect("remove scratch file");
4610 }
4611
4612 #[test]
4625 fn opening_costs_the_same_over_a_thousand_times_the_rows() {
4626 let opened = |label: &str, rows_per_part: i32| {
4627 let path = path(label);
4628 let mut writer =
4629 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4630 .expect("new file");
4631 for part in 0..STRIPE_PARTS * 3 {
4632 let values = (0..rows_per_part)
4636 .map(|row| {
4637 Value::Integer((part as i32 * rows_per_part + row).wrapping_mul(2_654_435))
4638 })
4639 .collect::<Vec<_>>();
4640 let chunk = Chunk::new(vec![
4641 Vector::from_values(LogicalType::Integer, &values).expect("integers"),
4642 ])
4643 .expect("matching rows");
4644 writer.append(&chunk).expect("one part");
4645 }
4646 writer.finish().expect("commit");
4647 let reader = Reader::open(&path).expect("reopen from disk");
4648 let size = fs::metadata(&path).expect("the file is there").len();
4649 let out = (reader.reads(), reader.table().stripes().len(), size);
4650 fs::remove_file(path).expect("remove scratch file");
4651 out
4652 };
4653
4654 let (thin, thin_stripes, thin_size) = opened("open-thin", 1);
4655 let (fat, fat_stripes, fat_size) = opened("open-fat", 1000);
4656 assert_eq!(
4657 thin_stripes, fat_stripes,
4658 "the same stripe count is what makes this a fair ask"
4659 );
4660 assert!(
4661 fat_size > thin_size * 50,
4662 "the fat file has to actually be larger, and it is {fat_size} against {thin_size}"
4663 );
4664
4665 assert_eq!(thin.opening.reads, fat.opening.reads, "the same reads either way");
4666 assert_eq!(thin.pages, 0, "opening read a page");
4667 assert_eq!(fat.pages, 0, "opening read a page");
4668 assert_eq!(thin.indexes, 0, "opening read an index");
4669 assert_eq!(fat.indexes, 0, "opening read an index");
4670 assert!(
4673 fat.opening.bytes < thin.opening.bytes * 2,
4674 "opening the thin file read {} bytes and the fat one read {}",
4675 thin.opening.bytes,
4676 fat.opening.bytes
4677 );
4678 }
4679
4680 #[test]
4688 fn two_opens_of_one_file_cost_the_same_and_the_second_is_not_cheaper() {
4689 let path = path("open-twice");
4690 let mut writer =
4691 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4692 .expect("new file");
4693 for part in 0..STRIPE_PARTS * 3 {
4694 let chunk = Chunk::new(vec![
4695 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
4696 .expect("integers"),
4697 ])
4698 .expect("matching rows");
4699 writer.append(&chunk).expect("one part");
4700 }
4701 writer.finish().expect("commit");
4702
4703 let first = Reader::open(&path).expect("open");
4704 for part in 0..first.parts() {
4707 first.read(part, &[0]).expect("a part");
4708 }
4709 assert!(first.reads().pages > 0, "the scan has to have read something");
4710 let second = Reader::open(&path).expect("open again");
4711
4712 assert_eq!(first.reads().opening, second.reads().opening);
4713 assert_eq!(
4714 second.reads().pages,
4715 0,
4716 "the second open read a page off the back of the first"
4717 );
4718 assert_eq!(second.reads().indexes, 0, "the second open read an index it inherited");
4719 fs::remove_file(path).expect("remove scratch file");
4720 }
4721
4722 #[test]
4730 fn an_index_is_read_once_per_stripe_however_often_the_page_is_evicted() {
4731 let path = path("index-cache");
4732 let mut writer =
4733 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4734 .expect("new file");
4735 let parts = STRIPE_PARTS * (CACHED_STRIPES_PER_COLUMN + 2);
4736 for part in 0..parts {
4737 let id = part as i32;
4738 let chunk = Chunk::new(vec![
4739 Vector::from_values(LogicalType::Integer, &[Value::Integer(id)]).expect("integers"),
4740 ])
4741 .expect("matching rows");
4742 writer.append(&chunk).expect("one part");
4743 }
4744 writer.finish().expect("commit");
4745
4746 let reader = Reader::open(&path).expect("reopen from disk");
4747 let stripes = reader.table().stripes().len();
4748 assert!(stripes > CACHED_STRIPES_PER_COLUMN, "the page cache has to be too small for this");
4749 for _ in 0..2 {
4751 for part in 0..parts {
4752 let chunk = reader.read(part, &[0]).expect("a part");
4753 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4754 }
4755 }
4756 assert_eq!(reader.indexes.load(Atomic::Relaxed), stripes, "one index read per stripe");
4757 assert!(
4758 reader.pages.load(Atomic::Relaxed) > stripes,
4759 "the pages are the ones that get read again, which is what makes the index count mean \
4760 something"
4761 );
4762 fs::remove_file(path).expect("remove scratch file");
4763 }
4764
4765 #[test]
4774 fn a_worker_per_stripe_reads_its_page_once_when_the_cache_was_told_to_expect_it() {
4775 let workers = CACHED_STRIPES_PER_COLUMN + 4;
4776 let path = path("stripe-per-worker");
4777 let mut writer =
4778 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4779 .expect("new file");
4780 for part in 0..STRIPE_PARTS * workers {
4781 let chunk = Chunk::new(vec![
4782 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
4783 .expect("integers"),
4784 ])
4785 .expect("matching rows");
4786 writer.append(&chunk).expect("one part");
4787 }
4788 writer.finish().expect("commit");
4789
4790 let read = |told: bool| {
4791 let reader = Reader::open(&path).expect("reopen from disk");
4792 assert_eq!(reader.table().stripes().len(), workers, "a stripe per worker");
4793 if told {
4794 reader.keep_stripes(workers);
4795 }
4796 let barrier = std::sync::Barrier::new(workers);
4797 std::thread::scope(|scope| {
4798 for (worker, run) in reader.stripe_parts().into_iter().enumerate() {
4799 let reader = &reader;
4800 let barrier = &barrier;
4801 scope.spawn(move || {
4802 for part in run {
4803 barrier.wait();
4804 let chunk = reader.read(part, &[0]).expect("a part of my own stripe");
4805 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4806 }
4807 assert!(worker < workers);
4808 });
4809 }
4810 });
4811 reader.pages.load(Atomic::Relaxed)
4812 };
4813
4814 assert_eq!(read(true), workers, "one page read per stripe and no more");
4815 assert!(read(false) > workers, "a cache that small is read again on every part");
4816 fs::remove_file(path).expect("remove scratch file");
4817 }
4818
4819 #[test]
4824 fn a_damaged_index_page_is_an_error() {
4825 let path = path("damaged-index");
4826 let mut writer =
4827 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4828 .expect("new file");
4829 writer.append(&sample_ids()).expect("first part");
4830 writer.append(&sample_ids()).expect("second part");
4831 writer.finish().expect("commit");
4832
4833 let reader = Reader::open(&path).expect("valid directory");
4834 let index = reader.table.stripes[0].index;
4835 let mut byte = [0; 1];
4836 read_at(&reader.file, index.offset, &mut byte).expect("the first part length");
4837 let mut file = OpenOptions::new().write(true).open(&path).expect("open index page");
4838 file.seek(SeekFrom::Start(index.offset)).expect("index start");
4839 file.write_all(&[!byte[0]]).expect("damage the first part length");
4840 let error = reader.read(1, &[0]).expect_err("a damaged index must not be used");
4841 assert!(error.message().contains("index page section checksum differs"), "{error}");
4842 fs::remove_file(path).expect("remove scratch file");
4843 }
4844
4845 #[test]
4852 fn every_integer_width_round_trips_through_a_page() {
4853 let path = path("integer-widths");
4854 let columns = [
4855 (LogicalType::TinyInt, vec![Value::TinyInt(i8::MIN), Value::TinyInt(i8::MAX)]),
4856 (LogicalType::UTinyInt, vec![Value::UTinyInt(0), Value::UTinyInt(u8::MAX)]),
4857 (LogicalType::SmallInt, vec![Value::SmallInt(i16::MIN), Value::SmallInt(i16::MAX)]),
4858 (LogicalType::USmallInt, vec![Value::USmallInt(0), Value::USmallInt(u16::MAX)]),
4859 (LogicalType::Integer, vec![Value::Integer(i32::MIN), Value::Integer(i32::MAX)]),
4860 (LogicalType::UInteger, vec![Value::UInteger(0), Value::UInteger(u32::MAX)]),
4861 (LogicalType::BigInt, vec![Value::BigInt(i64::MIN), Value::BigInt(i64::MAX)]),
4862 (LogicalType::UBigInt, vec![Value::UBigInt(0), Value::UBigInt(u64::MAX)]),
4863 ];
4864 let fields = columns
4865 .iter()
4866 .enumerate()
4867 .map(|(at, (ty, _))| Field::required(format!("c{at}"), ty.clone()))
4868 .collect::<Vec<_>>();
4869 let vectors = columns
4870 .iter()
4871 .map(|(ty, values)| Vector::from_values(ty.clone(), values).expect("a vector"))
4872 .collect::<Vec<_>>();
4873 let mut writer = Writer::create(&path, "widths", fields).expect("new file");
4874 writer.append(&Chunk::new(vectors).expect("matching rows")).expect("one stripe");
4875 writer.finish().expect("commit");
4876
4877 let reader = Reader::open(&path).expect("reopen from disk");
4878 let wanted = (0..columns.len()).collect::<Vec<_>>();
4879 let read = reader.read(0, &wanted).expect("every column");
4880 assert_eq!(read.len(), 2);
4881 for (at, (ty, values)) in columns.iter().enumerate() {
4883 assert_eq!(read.value_at(0, at), values[0], "the low end of {ty}");
4884 assert_eq!(read.value_at(1, at), values[1], "the high end of {ty}");
4885 }
4886 fs::remove_file(path).expect("remove scratch file");
4887 }
4888
4889 #[test]
4890 fn numeric_frequency_candidates_keep_bounded_row_ordinals() {
4891 let path = path("frequency-ordinals");
4892 let mut writer =
4893 Writer::create(&path, "items", vec![Field::required("id", LogicalType::BigInt)])
4894 .expect("new file");
4895 let mut values = Vec::new();
4896 for leader in 0..10_i64 {
4897 values.extend(std::iter::repeat_n(leader, 100));
4898 }
4899 values.extend(1_000_i64..41_000);
4900 for part in values.chunks(1_024) {
4901 let vector = Vector::flat(LogicalType::BigInt, Data::Int64(part.to_vec().into()))
4902 .expect("big integers");
4903 writer.append(&Chunk::new(vec![vector]).expect("one column")).expect("one stripe");
4904 }
4905 writer.finish().expect("commit");
4906
4907 let reader = Reader::open(&path).expect("reopen from disk");
4908 let occurrences =
4909 reader.frequency_occurrences(0).expect("valid metadata").expect("bounded ordinals");
4910 assert!(occurrences.omitted_max < 100);
4911 assert!(occurrences.ordinals.len() <= FREQUENCY_ORDINALS);
4912 assert!(occurrences.ordinals.windows(2).all(|pair| pair[0] < pair[1]));
4913 assert_eq!(&occurrences.ordinals[..1_000], &(0_u64..1_000).collect::<Vec<_>>());
4914 fs::remove_file(path).expect("remove scratch file");
4915 }
4916
4917 #[test]
4923 fn a_file_from_another_format_says_which_format_it_is() {
4924 let older = path("older-format");
4925 let mut writer =
4926 Writer::create(&older, "items", vec![Field::new("id", LogicalType::Integer)])
4927 .expect("new file");
4928 let chunk = Chunk::new(vec![
4929 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
4930 .expect("integers"),
4931 ])
4932 .expect("chunk");
4933 writer.append(&chunk).expect("page written");
4934 writer.finish().expect("commit");
4935
4936 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
4937 file.seek(SeekFrom::Start(8)).expect("the version follows the magic");
4938 file.write_all(&(FORMAT - 1).to_le_bytes()).expect("write an older version");
4939 drop(file);
4940 let complaint = Reader::open(&older).expect_err("an older format is refused").to_string();
4941 assert!(complaint.contains(&format!("format {}", FORMAT - 1)), "{complaint}");
4942 assert!(complaint.contains(&format!("format {FORMAT}")), "{complaint}");
4943
4944 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
4945 file.seek(SeekFrom::Start(0)).expect("the magic is first");
4946 file.write_all(b"NOTRUDB!").expect("write another engine's magic");
4947 drop(file);
4948 let complaint = Reader::open(&older).expect_err("a foreign file is refused").to_string();
4949 assert!(complaint.contains("magic"), "{complaint}");
4950 assert!(!complaint.contains("format"), "a version has nothing to do with it: {complaint}");
4951 fs::remove_file(older).expect("remove scratch file");
4952 }
4953
4954 #[test]
4955 fn an_unfinished_or_damaged_file_does_not_answer_with_partial_rows() {
4956 let unfinished = path("unfinished");
4957 let mut writer =
4958 Writer::create(&unfinished, "items", vec![Field::new("id", LogicalType::Integer)])
4959 .expect("new file");
4960 let chunk = Chunk::new(vec![
4961 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
4962 .expect("integers"),
4963 ])
4964 .expect("chunk");
4965 writer.append(&chunk).expect("page written");
4966 drop(writer);
4967 assert!(Reader::open(&unfinished).is_err(), "no directory was committed");
4968 fs::remove_file(unfinished).expect("remove scratch file");
4969
4970 let damaged = path("damaged");
4971 let mut writer =
4972 Writer::create(&damaged, "items", vec![Field::new("id", LogicalType::Integer)])
4973 .expect("new file");
4974 writer.append(&chunk).expect("page written");
4975 writer.finish().expect("commit");
4976 let reader = Reader::open(&damaged).expect("valid directory");
4977 let mut file =
4978 OpenOptions::new().write(true).open(&damaged).expect("open for a damaged page");
4979 file.seek(SeekFrom::Start(HEADER + 1)).expect("inside first page");
4980 file.write_all(&[255]).expect("damage one byte");
4981 assert!(reader.read(0, &[0]).is_err(), "page checksum rejects corruption");
4982 fs::remove_file(damaged).expect("remove scratch file");
4983 }
4984
4985 #[test]
4986 fn damaged_lazy_dictionary_payload_is_an_error() {
4987 let path = path("damaged-dictionary");
4988 let mut writer = Writer::create(
4989 &path,
4990 "items",
4991 vec![
4992 Field::required("id", LogicalType::Integer),
4993 Field::new("text", LogicalType::Varchar),
4994 ],
4995 )
4996 .expect("new file");
4997 writer.append(&sample()).expect("stripe written");
4998 writer.finish().expect("commit");
4999
5000 let reader = Reader::open(&path).expect("valid directory");
5001 let dictionary = reader.table.dictionaries[1].expect("string dictionary page");
5002 let mut header = [0; 12];
5005 read_at(&reader.file, dictionary.offset, &mut header).expect("dictionary header");
5006 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
5007 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
5008 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
5009 let index_len =
5010 12 + (count + 1) * 4 + (blocks + rank_blocks) * 8 + count * RANK_ENTRY as u64;
5011 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5012 file.seek(SeekFrom::Start(dictionary.offset + index_len))
5013 .expect("inside dictionary payload");
5014 file.write_all(&[255]).expect("damage dictionary payload");
5015
5016 let chunk = reader.read(0, &[1]).expect("code page and dictionary index remain valid");
5017 let error =
5018 chunk.validate_external().expect_err("payload corruption must reach the caller");
5019 assert!(error.message().contains("payload checksum differs"), "{error}");
5020 fs::remove_file(path).expect("remove scratch file");
5021 }
5022
5023 #[test]
5030 fn a_dictionary_over_one_extent_checks_every_block_of_it() {
5031 let path = path("dictionary-extents");
5032 let value =
5033 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5034 let parts = 30;
5035 let per_part = 1000;
5036 let mut writer =
5037 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5038 .expect("new file");
5039 for part in 0..parts {
5040 let values = (0..per_part)
5041 .map(|row| Value::Varchar(value(part * per_part + row)))
5042 .collect::<Vec<_>>();
5043 let chunk = Chunk::new(vec![
5044 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5045 ])
5046 .expect("matching rows");
5047 writer.append(&chunk).expect("a part");
5048 }
5049 writer.finish().expect("commit");
5050
5051 let reader = Reader::open(&path).expect("reopen from disk");
5052 let dictionary = reader.table.dictionaries[0].expect("string dictionary page");
5053 assert!(
5054 dictionary.length as usize > TEXT_PAYLOAD_BLOCK * TEXT_PAYLOAD_EXTENT,
5055 "the dictionary has to be over one extent for this to be testing anything"
5056 );
5057 for part in [0, parts - 1] {
5058 let chunk = reader.read(part, &[0]).expect("a part");
5059 chunk.validate_external().expect("every payload block checks out");
5060 assert_eq!(chunk.value_at(0, 0), Value::Varchar(value(part * per_part)));
5061 }
5062
5063 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5064 file.seek(SeekFrom::Start(dictionary.offset + u64::from(dictionary.length) - 4))
5065 .expect("the last bytes of the page are payload");
5066 file.write_all(&[255]).expect("damage the last payload block");
5067 let reader = Reader::open(&path).expect("the directory and the index are untouched");
5068 let chunk = reader.read(parts - 1, &[0]).expect("the code page remains valid");
5069 let error = chunk.validate_external().expect_err("the damage must reach the caller");
5070 assert!(error.message().contains("payload checksum differs"), "{error}");
5071 fs::remove_file(path).expect("remove scratch file");
5072 }
5073
5074 #[test]
5079 fn a_damaged_sorted_order_is_an_error() {
5080 let path = path("damaged-order");
5081 let mut writer = Writer::create(
5082 &path,
5083 "items",
5084 vec![
5085 Field::required("id", LogicalType::Integer),
5086 Field::new("text", LogicalType::Varchar),
5087 ],
5088 )
5089 .expect("new file");
5090 writer.append(&sample()).expect("stripe written");
5091 writer.finish().expect("commit");
5092
5093 let reader = Reader::open(&path).expect("valid directory");
5094 let page = reader.table.dictionaries[1].expect("string dictionary page");
5095 let mut header = [0; 12];
5096 read_at(&reader.file, page.offset, &mut header).expect("dictionary header");
5097 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
5098 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
5099 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
5100 let index_len = 12 + (count + 1) * 4 + (blocks + rank_blocks) * 8;
5101 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5102 file.seek(SeekFrom::Start(page.offset + index_len)).expect("the first head");
5103 file.write_all(&[255]).expect("damage the order");
5104
5105 let dictionary = reader.dictionary(1).expect("read").expect("a string column has one");
5106 let error = dictionary.compare_rank(0, b"anything").expect_err("a damaged order is caught");
5107 assert!(error.message().contains("rank checksum differs"), "{error}");
5108 fs::remove_file(path).expect("remove scratch file");
5109 }
5110
5111 #[test]
5115 fn a_global_dictionary_carries_the_sorted_order_of_its_values() {
5116 let spellings = ["overlong1z", "b", "", "overlong1a", "overlong", "ab", "a", "overlong1"];
5119 let path = path("dictionary-order");
5120 let mut writer =
5121 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
5122 .expect("new file");
5123 writer
5124 .append(
5125 &Chunk::new(vec![
5126 Vector::from_values(
5127 LogicalType::Varchar,
5128 &spellings.map(|text| Value::Varchar(text.into())),
5129 )
5130 .expect("strings"),
5131 ])
5132 .expect("one column"),
5133 )
5134 .expect("stripe written");
5135 writer.finish().expect("commit");
5136
5137 let reader = Reader::open(&path).expect("valid directory");
5138 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
5139 let count = dictionary.ranks().expect("a v10 file stores one");
5140 assert_eq!(count, spellings.len(), "every distinct value has a rank");
5141 let order = (0..count)
5142 .map(|rank| dictionary.code_at_rank(rank).expect("a code"))
5143 .collect::<Vec<_>>();
5144 let mut seen = order.clone();
5145 seen.sort_unstable();
5146 assert_eq!(seen, (0..spellings.len() as u32).collect::<Vec<_>>(), "a permutation of codes");
5147
5148 let ranked = order
5149 .iter()
5150 .map(|&code| {
5151 dictionary.try_bytes_at(code as usize).expect("read").expect("a value").to_vec()
5152 })
5153 .collect::<Vec<_>>();
5154 let mut expected = spellings.map(|text| text.as_bytes().to_vec()).to_vec();
5155 expected.sort();
5156 assert_eq!(ranked, expected, "rank order is value order");
5157
5158 for (rank, value) in expected.iter().enumerate() {
5161 assert_eq!(
5162 dictionary.compare_rank(rank, value).expect("compare"),
5163 Ordering::Equal,
5164 "rank {rank} is its own value"
5165 );
5166 if rank > 0 {
5167 assert_eq!(
5168 dictionary.compare_rank(rank - 1, value).expect("compare"),
5169 Ordering::Less,
5170 "rank {rank} follows the one before it"
5171 );
5172 }
5173 }
5174 fs::remove_file(path).expect("remove scratch file");
5175 }
5176
5177 #[test]
5178 fn damaged_membership_cannot_skip_a_string_page() {
5179 let path = path("damaged-membership");
5180 let mut writer = Writer::create(
5181 &path,
5182 "items",
5183 vec![
5184 Field::required("id", LogicalType::Integer),
5185 Field::new("text", LogicalType::Varchar),
5186 ],
5187 )
5188 .expect("new file");
5189 writer.append(&sample()).expect("stripe written");
5190 writer.finish().expect("commit");
5191
5192 let reader = Reader::open(&path).expect("valid directory");
5193 let membership = reader.table.stripes[0].memberships[1].expect("string membership");
5194 let mut file = OpenOptions::new().write(true).open(&path).expect("open membership page");
5195 file.seek(SeekFrom::Start(membership.offset)).expect("membership start");
5196 file.write_all(&[255]).expect("damage membership");
5197 let error = reader.skips_codes(0, 1, &[3]).expect_err("corruption must not skip rows");
5198 assert!(error.message().contains("membership page checksum differs"), "{error}");
5199 fs::remove_file(path).expect("remove scratch file");
5200 }
5201
5202 #[test]
5203 fn membership_delta_stream_is_sorted_exact_and_bounded() {
5204 let unique = unique_codes(&[900, 4, 4, 72, 9, u32::MAX]);
5205 assert_eq!(unique, [4, 9, 72, 900, u32::MAX]);
5206 let encoded = encode_membership(&unique);
5207 assert_eq!(
5208 decode_membership(&encoded).expect("valid membership"),
5209 [4, 9, 72, 900, u32::MAX]
5210 );
5211 let merged = merged_codes(vec![vec![4, 900], vec![9, 900, u32::MAX], vec![72]]);
5214 assert_eq!(merged, [4, 9, 72, 900, u32::MAX]);
5215 assert_eq!(
5216 decode_membership(&encode_membership(&merged)).expect("valid membership"),
5217 unique
5218 );
5219 assert!(decode_membership(&[1, 0x80]).is_err(), "a truncated varint is invalid");
5220 assert!(
5221 decode_membership(&[1, 0xff, 0xff, 0xff, 0xff, 0x10]).is_err(),
5222 "a value past u32 is invalid"
5223 );
5224 }
5225
5226 #[test]
5227 fn a_global_dictionary_may_be_larger_than_one_column_page() {
5228 let dictionary = Page {
5229 offset: HEADER,
5230 length: u32::try_from(MAX_PAGE + 1).expect("the page bound fits on disk"),
5231 hash: 0,
5232 };
5233 let table = Table {
5234 name: "items".to_owned(),
5235 fields: vec![Field::new("text", LogicalType::Varchar)],
5236 stripes: Vec::new(),
5237 rows: 0,
5238 dictionaries: vec![Some(dictionary)],
5239 frequencies: vec![None],
5240 };
5241 let directory = encode_directory(&table).expect("directory");
5242 let file_size = dictionary.offset + u64::from(dictionary.length) + 1;
5243
5244 let decoded = decode_directory(&directory, file_size).expect("large lazy dictionary");
5245 assert_eq!(decoded.dictionaries[0].expect("dictionary").length, dictionary.length);
5246 }
5247
5248 #[test]
5249 fn a_column_with_one_value_everywhere_costs_almost_nothing_a_row() {
5250 let path = path("constant-codes");
5251 let mut writer =
5252 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
5253 .expect("new file");
5254 let empty = vec![Value::Varchar(String::new()); 1024];
5255 for _ in 0..4 {
5256 let column = Vector::from_values(LogicalType::Varchar, &empty).expect("strings");
5257 writer.append(&Chunk::new(vec![column]).expect("one column")).expect("a part");
5258 }
5259 writer.finish().expect("commit");
5260
5261 let reader = Reader::open(&path).expect("valid directory");
5262 let pages = reader.layout().columns.first().expect("one column").pages;
5263 assert!(pages < 256, "{pages} bytes of pages for 4,096 rows of one value");
5267 let read = reader.read(3, &[0]).expect("the last part back");
5268 assert_eq!(read.value_at(0, 0), Value::Varchar(String::new()));
5269 assert_eq!(read.value_at(1023, 0), Value::Varchar(String::new()));
5270 fs::remove_file(path).expect("remove scratch file");
5271 }
5272
5273 #[test]
5274 fn a_cascade_value_too_wide_for_its_column_is_refused_rather_than_cut() {
5275 let over = vec![i64::from(i32::MAX) + 1];
5278 let error = narrowed(&LogicalType::Integer, over).expect_err("a page that disagrees");
5279 assert!(format!("{error}").contains("not of its type"), "{error}");
5280 assert!(narrowed(&LogicalType::BigInt, vec![i64::MIN]).is_ok(), "bigint holds all of i64");
5281 assert!(narrowed(&LogicalType::Varchar, vec![0]).is_err(), "strings are not integers");
5282 }
5283
5284 #[test]
5285 fn a_code_stream_the_cascade_cannot_shrink_is_left_alone() {
5286 let mut state: u32 = 0x9e37_79b9;
5290 let spread: Vec<u32> = (0..1024)
5291 .map(|_| {
5292 state ^= state << 13;
5293 state ^= state >> 17;
5294 state ^= state << 5;
5295 state
5296 })
5297 .collect();
5298 assert_eq!(encoded_codes(&spread).expect("no failure"), None);
5299 let near: Vec<u32> = (0..1024).collect();
5300 let coded = encoded_codes(&near).expect("no failure").expect("counting up is packable");
5301 assert!(coded.len() < near.len() * 4, "{} bytes for a run of 1,024", coded.len());
5302 }
5303
5304 #[test]
5310 fn two_writes_of_the_same_rows_give_the_same_bytes() {
5311 fn written(path: &PathBuf) {
5312 let fields = (0..40)
5313 .map(|column| {
5314 let ty =
5315 if column % 4 == 0 { LogicalType::Varchar } else { LogicalType::BigInt };
5316 Field::new(format!("c{column}"), ty)
5317 })
5318 .collect::<Vec<_>>();
5319 let mut writer = Writer::create(path, "wide", fields).expect("new file");
5320 for part in 0..70_u64 {
5321 let columns = (0..40)
5322 .map(|column| {
5323 let values = (0..64_u64)
5324 .map(|row| {
5325 let seed = part.wrapping_mul(31).wrapping_add(row);
5326 if column % 4 == 0 {
5327 Value::Varchar(format!("v{}", seed % 17))
5328 } else {
5329 Value::BigInt(i64::try_from(seed % 97).expect("small"))
5330 }
5331 })
5332 .collect::<Vec<_>>();
5333 let ty = if column % 4 == 0 {
5334 LogicalType::Varchar
5335 } else {
5336 LogicalType::BigInt
5337 };
5338 Vector::from_values(ty, &values).expect("a column")
5339 })
5340 .collect::<Vec<_>>();
5341 writer.append(&Chunk::new(columns).expect("forty columns")).expect("a part");
5342 }
5343 writer.finish().expect("commit");
5344 }
5345
5346 let first = path("repeatable-one");
5347 let second = path("repeatable-two");
5348 written(&first);
5349 written(&second);
5350 let left = fs::read(&first).expect("the first file");
5351 let right = fs::read(&second).expect("the second file");
5352 assert_eq!(left.len(), right.len(), "two writes of the same rows differ in length");
5353 assert!(left == right, "two writes of the same rows differ in their bytes");
5354
5355 let reader = Reader::open(&first).expect("valid directory");
5358 assert_eq!(reader.table().rows(), 70 * 64);
5359 let read = reader.read(0, &[0, 1]).expect("the first part back");
5360 assert_eq!(read.value_at(0, 0), Value::Varchar("v0".to_owned()));
5361 assert_eq!(read.value_at(0, 1), Value::BigInt(0));
5362 fs::remove_file(first).expect("remove scratch file");
5363 fs::remove_file(second).expect("remove scratch file");
5364 }
5365}