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 = 15;
51const HEADER: u64 = 80;
52const SLOT_BYTES: usize = 28;
53const MAX_PAGE: usize = 256 * 1024 * 1024;
54const MAX_DIRECTORY: usize = 128 * 1024 * 1024;
55const FREQUENCIES: &[u8; 8] = b"RUDBFQ2\0";
56const FREQUENCY_CANDIDATES: usize = 32_768;
57const FREQUENCY_ENTRIES: usize = 512;
58const FREQUENCY_BUILD_RANK: usize = 10;
59const FREQUENCY_ORDINALS: usize = 65_536;
60const MAX_FREQUENCY_WORKERS: usize = 32;
67
68const MAX_ENCODE_WORKERS: usize = 32;
75
76const SIEVE_BUDGET: usize = 8 * 1024;
82
83fn io(error: std::io::Error) -> Error {
84 Error::io(error.to_string())
85}
86
87fn invalid(message: &str) -> Error {
88 Error::invalid_input(format!("invalid rudb native file: {message}"))
89}
90
91fn sum(counts: impl Iterator<Item = u64>) -> u64 {
93 counts.fold(0, u64::saturating_add)
94}
95
96fn span_bytes(spans: &[Span], at: usize) -> u64 {
98 spans.get(at).map_or(0, |span| u64::from(span.length))
99}
100
101fn page_bytes(pages: &[Option<Page>], at: usize) -> u64 {
103 pages.get(at).and_then(Option::as_ref).map_or(0, Page::bytes)
104}
105
106fn checksum(bytes: &[u8]) -> u64 {
107 const P1: u64 = 11_400_714_785_074_694_791;
108 const P2: u64 = 14_029_467_366_897_019_727;
109 const P3: u64 = 1_609_587_929_392_839_161;
110 const P4: u64 = 9_650_029_242_287_828_579;
111 const P5: u64 = 2_870_177_450_012_600_261;
112 let round = |state: u64, word: u64| {
113 state.wrapping_add(word.wrapping_mul(P2)).rotate_left(31).wrapping_mul(P1)
114 };
115 let merge = |state: u64, lane: u64| (state ^ round(0, lane)).wrapping_mul(P1).wrapping_add(P4);
116 let word =
117 |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("eight checksum bytes"));
118
119 let mut at = 0;
120 let mut hash = if bytes.len() >= 32 {
121 let mut one = P1.wrapping_add(P2);
122 let mut two = P2;
123 let mut three = 0;
124 let mut four = 0_u64.wrapping_sub(P1);
125 while at + 32 <= bytes.len() {
126 one = round(one, word(at));
127 two = round(two, word(at + 8));
128 three = round(three, word(at + 16));
129 four = round(four, word(at + 24));
130 at += 32;
131 }
132 let combined = one
133 .rotate_left(1)
134 .wrapping_add(two.rotate_left(7))
135 .wrapping_add(three.rotate_left(12))
136 .wrapping_add(four.rotate_left(18));
137 merge(merge(merge(merge(combined, one), two), three), four)
138 } else {
139 P5
140 };
141 hash = hash.wrapping_add(bytes.len() as u64);
142 while at + 8 <= bytes.len() {
143 hash ^= round(0, word(at));
144 hash = hash.rotate_left(27).wrapping_mul(P1).wrapping_add(P4);
145 at += 8;
146 }
147 if at + 4 <= bytes.len() {
148 let tail = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four checksum bytes"));
149 hash ^= u64::from(tail).wrapping_mul(P1);
150 hash = hash.rotate_left(23).wrapping_mul(P2).wrapping_add(P3);
151 at += 4;
152 }
153 while at < bytes.len() {
154 hash ^= u64::from(bytes[at]).wrapping_mul(P5);
155 hash = hash.rotate_left(11).wrapping_mul(P1);
156 at += 1;
157 }
158 hash ^= hash >> 33;
159 hash = hash.wrapping_mul(P2);
160 hash ^= hash >> 29;
161 hash = hash.wrapping_mul(P3);
162 hash ^ (hash >> 32)
163}
164
165#[derive(Debug, Clone, Copy)]
166struct Slot {
167 offset: u64,
168 length: u32,
169 generation: u64,
170 hash: u64,
171}
172
173impl Slot {
174 fn bytes(self) -> [u8; SLOT_BYTES] {
175 let mut result = [0; SLOT_BYTES];
176 result[..8].copy_from_slice(&self.offset.to_le_bytes());
177 result[8..12].copy_from_slice(&self.length.to_le_bytes());
178 result[12..20].copy_from_slice(&self.generation.to_le_bytes());
179 result[20..28].copy_from_slice(&self.hash.to_le_bytes());
180 result
181 }
182
183 fn read(bytes: &[u8]) -> Self {
184 Self {
185 offset: u64::from_le_bytes(bytes[..8].try_into().expect("eight bytes")),
186 length: u32::from_le_bytes(bytes[8..12].try_into().expect("four bytes")),
187 generation: u64::from_le_bytes(bytes[12..20].try_into().expect("eight bytes")),
188 hash: u64::from_le_bytes(bytes[20..28].try_into().expect("eight bytes")),
189 }
190 }
191}
192
193#[derive(Debug, Clone, Copy)]
194struct Page {
195 offset: u64,
196 length: u32,
197 hash: u64,
198}
199
200impl Page {
201 fn bytes(&self) -> u64 {
203 u64::from(self.length)
204 }
205}
206
207#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
208enum FrequencyValue {
209 Null,
210 Integer(i128),
211 Code(u32),
212}
213
214#[derive(Debug, Clone)]
215struct FrequencyEntry {
216 value: FrequencyValue,
217 count: u64,
218}
219
220#[derive(Debug, Clone)]
225struct FrequencySummary {
226 entries: Vec<FrequencyEntry>,
227 omitted_max: u64,
228 ordinals: Vec<u64>,
229}
230
231#[derive(Debug, Clone, PartialEq, Eq)]
233pub struct FrequencyOccurrences {
234 pub omitted_max: u64,
236 pub ordinals: Vec<u64>,
238}
239
240#[derive(Debug, Clone, Copy, Default)]
247struct Span {
248 offset: u64,
249 length: u32,
250}
251
252#[derive(Debug, Clone)]
254pub struct Stripe {
255 rows: usize,
256 parts: Vec<u32>,
259 index: Span,
263 pages: Vec<Span>,
264 memberships: Vec<Option<Page>>,
265 sieves: Vec<Option<Page>>,
268 zone: Zone,
269}
270
271impl Stripe {
272 #[must_use]
274 pub fn rows(&self) -> usize {
275 self.rows
276 }
277
278 #[must_use]
280 pub fn parts(&self) -> usize {
281 self.parts.len()
282 }
283}
284
285#[derive(Debug, Clone)]
287pub struct Table {
288 name: String,
289 fields: Vec<Field>,
290 stripes: Vec<Stripe>,
291 rows: usize,
292 dictionaries: Vec<Option<Page>>,
293 frequencies: Vec<Option<FrequencySummary>>,
294}
295
296impl Table {
297 #[must_use]
299 pub fn name(&self) -> &str {
300 &self.name
301 }
302
303 #[must_use]
305 pub fn fields(&self) -> &[Field] {
306 &self.fields
307 }
308
309 #[must_use]
311 pub fn rows(&self) -> usize {
312 self.rows
313 }
314
315 #[must_use]
317 pub fn stripes(&self) -> &[Stripe] {
318 &self.stripes
319 }
320}
321
322#[derive(Debug, Clone)]
324pub struct ColumnLayout {
325 pub name: String,
327 pub kind: String,
329 pub pages: u64,
331 pub memberships: u64,
333 pub sieves: u64,
335 pub dictionary: u64,
337}
338
339impl ColumnLayout {
340 #[must_use]
342 pub fn total(&self) -> u64 {
343 self.pages
344 .saturating_add(self.memberships)
345 .saturating_add(self.sieves)
346 .saturating_add(self.dictionary)
347 }
348}
349
350#[derive(Debug, Clone)]
361pub struct Layout {
362 pub file: u64,
364 pub rows: usize,
366 pub stripes: usize,
368 pub parts: usize,
370 pub columns: Vec<ColumnLayout>,
372 pub indexes: u64,
375 pub directory: u64,
377 pub header: u64,
379}
380
381impl Layout {
382 #[must_use]
384 pub fn columns_total(&self) -> u64 {
385 self.columns.iter().map(ColumnLayout::total).fold(0, u64::saturating_add)
386 }
387
388 #[must_use]
394 pub fn unaccounted(&self) -> u64 {
395 self.file
396 .saturating_sub(self.columns_total())
397 .saturating_sub(self.indexes)
398 .saturating_sub(self.directory)
399 .saturating_sub(self.header)
400 }
401}
402
403#[derive(Debug)]
405struct GlobalDictionary {
406 primary: HashMap<u64, u32>,
407 collisions: HashMap<u64, Vec<u32>>,
408 offsets: Vec<u32>,
409 payload: Vec<u8>,
410 counts: Vec<u64>,
411 nulls: u64,
412}
413
414impl GlobalDictionary {
415 fn new() -> Self {
416 Self {
417 primary: HashMap::new(),
418 collisions: HashMap::new(),
419 offsets: vec![0],
420 payload: Vec::new(),
421 counts: Vec::new(),
422 nulls: 0,
423 }
424 }
425
426 fn bytes(&self, code: u32) -> Option<&[u8]> {
427 let start = *self.offsets.get(code as usize)? as usize;
428 let end = *self.offsets.get(code as usize + 1)? as usize;
429 self.payload.get(start..end)
430 }
431
432 fn code(&mut self, text: &str) -> Result<u32> {
433 let hash = checksum(text.as_bytes());
434 if let Some(&code) = self.primary.get(&hash) {
435 if self.bytes(code) == Some(text.as_bytes()) {
436 return Ok(code);
437 }
438 if let Some(codes) = self.collisions.get(&hash) {
439 if let Some(code) =
440 codes.iter().copied().find(|&code| self.bytes(code) == Some(text.as_bytes()))
441 {
442 return Ok(code);
443 }
444 }
445 let code = self.insert(text)?;
446 self.collisions.entry(hash).or_default().push(code);
447 return Ok(code);
448 }
449 let code = self.insert(text)?;
450 self.primary.insert(hash, code);
451 Ok(code)
452 }
453
454 fn insert(&mut self, text: &str) -> Result<u32> {
455 let code = u32::try_from(self.offsets.len() - 1)
456 .map_err(|_| invalid("global dictionary has too many values"))?;
457 self.payload.extend_from_slice(text.as_bytes());
458 self.offsets.push(
459 u32::try_from(self.payload.len())
460 .map_err(|_| invalid("global dictionary payload exceeds 4 GiB"))?,
461 );
462 self.counts.push(0);
463 Ok(code)
464 }
465
466 fn ranked(&self) -> Vec<(u64, u32)> {
486 let count = self.offsets.len() - 1;
487 let mut ranked = (0..count)
488 .map(|code| {
489 let code = code as u32;
490 (head(self.bytes(code).unwrap_or_default()), code)
491 })
492 .collect::<Vec<_>>();
493 ranked.sort_unstable_by(|left, right| {
494 left.0.cmp(&right.0).then_with(|| self.bytes(left.1).cmp(&self.bytes(right.1)))
495 });
496 ranked
497 }
498
499 fn observe(&mut self, code: u32, null: bool) -> Result<()> {
500 if null {
501 self.nulls = self.nulls.saturating_add(1);
502 return Ok(());
503 }
504 let count = self
505 .counts
506 .get_mut(code as usize)
507 .ok_or_else(|| invalid("global dictionary count code is out of range"))?;
508 *count = count.saturating_add(1);
509 Ok(())
510 }
511}
512
513#[derive(Debug)]
515pub struct Writer {
516 file: File,
517 at: u64,
525 table: Table,
526 generation: u64,
527 order: Vec<((u64, u64), (u64, u64))>,
530 next_order: u64,
531 dictionaries: Vec<Option<GlobalDictionary>>,
532 pending: Vec<PendingChunk>,
533}
534
535#[derive(Debug)]
543struct PendingChunk {
544 order: (u64, u64),
545 chunk: Chunk,
546}
547
548#[derive(Debug)]
554struct ColumnStripe {
555 pages: Vec<Vec<u8>>,
556 codes: Vec<Option<Vec<u32>>>,
557 sieves: Vec<Option<Sieve>>,
558 ranges: Vec<Range>,
559}
560
561fn weight(ty: &LogicalType) -> usize {
569 match ty {
570 LogicalType::Varchar | LogicalType::Blob => 64,
571 LogicalType::BigInt
572 | LogicalType::UBigInt
573 | LogicalType::Timestamp
574 | LogicalType::Double
575 | LogicalType::Decimal { .. } => 8,
576 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date | LogicalType::Float => 4,
577 LogicalType::SmallInt | LogicalType::USmallInt => 2,
578 _ => 1,
579 }
580}
581
582pub const STRIPE_PARTS: usize = 64;
589
590const INDEX_ENTRY: usize = size_of::<u32>() + size_of::<u64>();
592
593fn index_section(parts: usize) -> Result<usize> {
595 parts
596 .checked_mul(INDEX_ENTRY)
597 .and_then(|bytes| bytes.checked_add(size_of::<u64>()))
598 .ok_or_else(|| invalid("index page length overflow"))
599}
600
601impl Writer {
602 pub fn create(
608 path: impl AsRef<Path>,
609 name: impl Into<String>,
610 fields: Vec<Field>,
611 ) -> Result<Self> {
612 for field in &fields {
613 type_tag(&field.ty)?;
614 }
615 let file =
616 OpenOptions::new().write(true).read(true).create_new(true).open(path).map_err(io)?;
617 let mut header = [0; HEADER as usize];
618 header[..8].copy_from_slice(MAGIC);
619 header[8..12].copy_from_slice(&FORMAT.to_le_bytes());
620 write_at(&file, 0, &header)?;
621 Ok(Self {
622 file,
623 at: HEADER,
624 dictionaries: fields
625 .iter()
626 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
627 .collect(),
628 table: Table {
629 name: name.into(),
630 dictionaries: vec![None; fields.len()],
631 fields,
632 stripes: Vec::new(),
633 rows: 0,
634 frequencies: Vec::new(),
635 },
636 generation: 1,
637 order: Vec::new(),
638 next_order: 0,
639 pending: Vec::with_capacity(STRIPE_PARTS),
640 })
641 }
642
643 fn put(&mut self, bytes: &[u8]) -> Result<()> {
648 write_at(&self.file, self.at, bytes)?;
649 self.at = self
650 .at
651 .checked_add(bytes.len() as u64)
652 .ok_or_else(|| invalid("native file length overflow"))?;
653 Ok(())
654 }
655
656 pub fn append(&mut self, chunk: &Chunk) -> Result<()> {
662 let order = (self.next_order, 0);
663 self.next_order = self.next_order.saturating_add(1);
664 self.append_at(order, chunk)
665 }
666
667 pub fn append_at(&mut self, order: (u64, u64), chunk: &Chunk) -> Result<()> {
678 if chunk.is_empty() {
679 return Ok(());
680 }
681 self.admit(chunk)?;
682 if self.pending.last().is_some_and(|last| last.order > order) {
683 self.flush_pending()?;
684 }
685 self.pending.push(PendingChunk { order, chunk: chunk.clone() });
690 if self.pending.len() == STRIPE_PARTS {
691 self.flush_pending()?;
692 }
693 Ok(())
694 }
695
696 pub fn append_stripe(&mut self, parts: Vec<((u64, u64), Chunk)>) -> Result<()> {
712 if parts.len() > STRIPE_PARTS {
713 return Err(invalid("a stripe was handed more parts than it holds"));
714 }
715 self.flush_pending()?;
718 for (order, chunk) in parts {
719 if chunk.is_empty() {
720 continue;
721 }
722 self.admit(&chunk)?;
723 self.pending.push(PendingChunk { order, chunk });
724 }
725 self.flush_pending()
726 }
727
728 fn admit(&mut self, chunk: &Chunk) -> Result<()> {
730 if chunk.width() != self.table.fields.len() {
731 return Err(invalid("chunk width differs from table schema"));
732 }
733 for (index, field) in self.table.fields.iter().enumerate() {
734 if chunk.column(index)?.logical_type() != &field.ty {
735 return Err(invalid("chunk type differs from table schema"));
736 }
737 }
738 self.table.rows = self
739 .table
740 .rows
741 .checked_add(chunk.len())
742 .ok_or_else(|| invalid("row count overflow"))?;
743 Ok(())
744 }
745
746 fn encode_column(
754 index: usize,
755 held: &[PendingChunk],
756 mut dictionary: Option<&mut GlobalDictionary>,
757 ) -> Result<ColumnStripe> {
758 let mut stripe = ColumnStripe {
759 pages: Vec::with_capacity(held.len()),
760 codes: Vec::with_capacity(held.len()),
761 sieves: Vec::with_capacity(held.len()),
762 ranges: Vec::with_capacity(held.len()),
763 };
764 for pending in held {
765 let column = pending.chunk.column(index)?;
766 let (bytes, unique) = encode(column, dictionary.as_deref_mut())?;
767 if bytes.len() > MAX_PAGE {
768 return Err(invalid("column page exceeds the configured bound"));
769 }
770 let range = Range::of(column);
773 let sieve = match dictionary {
778 Some(_) => None,
779 None => Sieve::of(column, &range, SIEVE_BUDGET),
780 };
781 stripe.pages.push(bytes);
782 stripe.codes.push(unique);
783 stripe.sieves.push(sieve);
784 stripe.ranges.push(range);
785 }
786 Ok(stripe)
787 }
788
789 fn encode_columns(&mut self, held: &[PendingChunk]) -> Result<Vec<ColumnStripe>> {
798 let width = self.table.fields.len();
799 let workers = std::thread::available_parallelism()
800 .map_or(1, usize::from)
801 .min(MAX_ENCODE_WORKERS)
802 .min(width);
803 if workers <= 1 || held.len() <= 1 {
804 return self
805 .dictionaries
806 .iter_mut()
807 .enumerate()
808 .map(|(index, dictionary)| Self::encode_column(index, held, dictionary.as_mut()))
809 .collect();
810 }
811 let mut jobs: Vec<(usize, Option<GlobalDictionary>)> =
814 std::mem::take(&mut self.dictionaries).into_iter().enumerate().collect();
815 jobs.sort_by_key(|(index, _)| weight(&self.table.fields[*index].ty));
817 let queue = Mutex::new(jobs);
818 let pieces = std::thread::scope(|scope| {
819 (0..workers)
820 .map(|_| {
821 scope.spawn(|| {
822 let mut mine = Vec::new();
823 loop {
824 let taken = queue
825 .lock()
826 .map_err(|_| Error::internal("a native encode worker panicked"))?
827 .pop();
828 let Some((index, mut dictionary)) = taken else { break };
829 let encoded = Self::encode_column(index, held, dictionary.as_mut())?;
830 mine.push((index, dictionary, encoded));
831 }
832 Ok(mine)
833 })
834 })
835 .collect::<Vec<_>>()
836 .into_iter()
837 .map(|handle| {
838 handle.join().map_err(|_| Error::internal("a native encode worker panicked"))?
839 })
840 .collect::<Result<Vec<_>>>()
841 })?;
842 let mut dictionaries: Vec<Option<GlobalDictionary>> = (0..width).map(|_| None).collect();
843 let mut encoded: Vec<Option<ColumnStripe>> = (0..width).map(|_| None).collect();
844 for piece in pieces {
845 for (index, dictionary, stripe) in piece {
846 dictionaries[index] = dictionary;
847 encoded[index] = Some(stripe);
848 }
849 }
850 self.dictionaries = dictionaries;
851 encoded
852 .into_iter()
853 .map(|stripe| stripe.ok_or_else(|| Error::internal("a column was never encoded")))
854 .collect()
855 }
856
857 fn flush_pending(&mut self) -> Result<()> {
859 if self.pending.is_empty() {
860 return Ok(());
861 }
862 let width = self.table.fields.len();
863 let mut held = std::mem::take(&mut self.pending);
866 let parts = held.len();
867 let encoded = self.encode_columns(&held)?;
868 let mut pages = Vec::with_capacity(width);
869 let mut memberships = vec![None; width];
870 let mut ranges = Vec::with_capacity(width);
871 let mut index = Vec::with_capacity(width.saturating_mul(index_section(parts)?));
872 for stripe in &encoded {
873 let offset = self.at;
874 let section = index.len();
875 let mut length = 0_usize;
876 for bytes in &stripe.pages {
877 write_at(&self.file, self.at + length as u64, bytes)?;
878 put_u32(
879 &mut index,
880 u32::try_from(bytes.len()).map_err(|_| invalid("part length overflow"))?,
881 );
882 put_u64(&mut index, checksum(bytes));
883 length = length
884 .checked_add(bytes.len())
885 .ok_or_else(|| invalid("column page length overflow"))?;
886 }
887 let hash = checksum(&index[section..]);
888 put_u64(&mut index, hash);
889 if length > MAX_PAGE {
890 return Err(invalid("column page exceeds the configured bound"));
891 }
892 self.at = self
893 .at
894 .checked_add(length as u64)
895 .ok_or_else(|| invalid("native file length overflow"))?;
896 pages.push(Span {
897 offset,
898 length: u32::try_from(length).map_err(|_| invalid("page length overflow"))?,
899 });
900 ranges.push(merged_range(stripe.ranges.iter().cloned()));
901 }
902 for (membership, stripe) in memberships.iter_mut().zip(&encoded) {
903 if stripe.codes.iter().all(Option::is_none) {
904 continue;
905 }
906 let lists = stripe
907 .codes
908 .iter()
909 .map(|codes| codes.clone().unwrap_or_default())
910 .collect::<Vec<_>>();
911 let bytes = encode_membership(&merged_codes(lists));
912 let offset = self.at;
913 self.put(&bytes)?;
914 *membership = Some(Page {
915 offset,
916 length: u32::try_from(bytes.len())
917 .map_err(|_| invalid("membership page length overflow"))?,
918 hash: checksum(&bytes),
919 });
920 }
921 let mut sieves = vec![None; width];
922 for (page, stripe) in sieves.iter_mut().zip(&encoded) {
923 if stripe.sieves.iter().all(Option::is_none) {
924 continue;
925 }
926 let bytes = encode_sieves(stripe.sieves.iter())?;
927 let offset = self.at;
928 self.put(&bytes)?;
929 *page = Some(Page {
930 offset,
931 length: u32::try_from(bytes.len())
932 .map_err(|_| invalid("sieve page length overflow"))?,
933 hash: checksum(&bytes),
934 });
935 }
936 let offset = self.at;
937 self.put(&index)?;
938 let index = Span {
939 offset,
940 length: u32::try_from(index.len())
941 .map_err(|_| invalid("index page length overflow"))?,
942 };
943 let mut rows = 0_usize;
944 let mut lengths = Vec::with_capacity(parts);
945 let mut span = None;
946 for pending in held.drain(..) {
947 let part = pending.chunk.len();
948 rows = rows.checked_add(part).ok_or_else(|| invalid("row count overflow"))?;
949 lengths.push(u32::try_from(part).map_err(|_| invalid("part row count overflow"))?);
950 span = Some(
951 span.map_or((pending.order, pending.order), |(first, _)| (first, pending.order)),
952 );
953 }
954 self.order.push(span.ok_or_else(|| invalid("a stripe was flushed with no parts"))?);
955 self.table.stripes.push(Stripe {
956 rows,
957 parts: lengths,
958 index,
959 pages,
960 memberships,
961 sieves,
962 zone: Zone::from_ranges(ranges),
963 });
964 self.pending = held;
966 Ok(())
967 }
968
969 fn numeric_frequency(&self, column: usize) -> Result<Option<FrequencySummary>> {
973 let ty = &self.table.fields[column].ty;
974 if !matches!(
975 ty,
976 LogicalType::TinyInt
977 | LogicalType::SmallInt
978 | LogicalType::Integer
979 | LogicalType::BigInt
980 | LogicalType::UTinyInt
981 | LogicalType::USmallInt
982 | LogicalType::UInteger
983 | LogicalType::UBigInt
984 | LogicalType::Date
985 | LogicalType::Timestamp
986 ) {
987 return Ok(None);
988 }
989 let mut candidates: HashMap<FrequencyValue, u32> = HashMap::new();
990 let mut decrements = 0_u64;
991 self.visit_numeric(column, |_, value| {
992 if let Some(count) = candidates.get_mut(&value) {
993 *count = count.saturating_add(1);
994 } else if candidates.len() < FREQUENCY_CANDIDATES {
995 candidates.insert(value, 1);
996 } else {
997 candidates.retain(|_, count| {
998 *count -= 1;
999 *count != 0
1000 });
1001 decrements = decrements.saturating_add(1);
1002 }
1003 })?;
1004 let (exact, ordinals) = if decrements == 0 {
1005 (
1006 candidates
1007 .into_iter()
1008 .map(|(value, count)| (value, u64::from(count)))
1009 .collect::<HashMap<_, _>>(),
1010 Vec::new(),
1011 )
1012 } else {
1013 let mut lower = candidates.values().copied().collect::<Vec<_>>();
1014 lower.sort_unstable_by(|left, right| right.cmp(left));
1015 if lower.len() < FREQUENCY_BUILD_RANK
1016 || u64::from(lower[FREQUENCY_BUILD_RANK - 1]) <= decrements
1017 {
1018 return Ok(None);
1019 }
1020 let mut exact =
1021 candidates.into_keys().map(|value| (value, 0_u64)).collect::<HashMap<_, _>>();
1022 let mut ordinals = Vec::new();
1023 let mut exceeded = false;
1024 self.visit_numeric(column, |ordinal, value| {
1025 if let Some(count) = exact.get_mut(&value) {
1026 *count = count.saturating_add(1);
1027 if !exceeded {
1028 if ordinals.len() < FREQUENCY_ORDINALS {
1029 ordinals.push(ordinal);
1030 } else {
1031 ordinals.clear();
1032 exceeded = true;
1033 }
1034 }
1035 }
1036 })?;
1037 (exact, ordinals)
1038 };
1039 let mut entries = exact
1040 .into_iter()
1041 .map(|(value, count)| FrequencyEntry { value, count })
1042 .collect::<Vec<_>>();
1043 entries.sort_unstable_by(|left, right| {
1044 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
1045 });
1046 let omitted_max =
1047 entries.get(FREQUENCY_ENTRIES).map_or(decrements, |entry| decrements.max(entry.count));
1048 entries.truncate(FREQUENCY_ENTRIES);
1049 Ok(Some(FrequencySummary { entries, omitted_max, ordinals }))
1050 }
1051
1052 fn visit_numeric(
1053 &self,
1054 column: usize,
1055 mut visit: impl FnMut(u64, FrequencyValue),
1056 ) -> Result<()> {
1057 let ty = &self.table.fields[column].ty;
1058 let mut start = 0_u64;
1059 for stripe in &self.table.stripes {
1060 let spans = read_index(&self.file, stripe, column)?;
1061 let page = stripe.pages[column];
1062 let mut bytes = vec![0; page.length as usize];
1063 read_at(&self.file, page.offset, &mut bytes)?;
1064 for (span, &rows) in spans.iter().zip(&stripe.parts) {
1065 let part = part_bytes(&bytes, *span)?;
1066 if checksum(part) != span.hash {
1067 return Err(invalid("column page checksum differs while building frequencies"));
1068 }
1069 let rows = rows as usize;
1070 let vector = decode(ty, rows, part, None)?;
1071 for row in 0..rows {
1073 let value = if vector.is_null_at(row) {
1074 FrequencyValue::Null
1075 } else {
1076 let widened = match vector.signed_at(row) {
1080 Some(value) => Some(value),
1081 None => match vector.value_at(row) {
1082 Value::UTinyInt(value) => Some(i128::from(value)),
1083 Value::USmallInt(value) => Some(i128::from(value)),
1084 Value::UInteger(value) => Some(i128::from(value)),
1085 Value::UBigInt(value) => Some(i128::from(value)),
1086 _ => None,
1087 },
1088 };
1089 FrequencyValue::Integer(widened.ok_or_else(|| {
1090 invalid("numeric frequency page did not contain an integer value")
1091 })?)
1092 };
1093 visit(start.saturating_add(row as u64), value);
1094 }
1095 start = start.saturating_add(rows as u64);
1096 }
1097 }
1098 Ok(())
1099 }
1100
1101 fn numeric_frequencies(&self) -> Result<Vec<Option<FrequencySummary>>> {
1109 let mut columns = self
1110 .table
1111 .fields
1112 .iter()
1113 .enumerate()
1114 .filter_map(|(column, field)| {
1115 matches!(
1116 field.ty,
1117 LogicalType::TinyInt
1118 | LogicalType::SmallInt
1119 | LogicalType::Integer
1120 | LogicalType::BigInt
1121 | LogicalType::UTinyInt
1122 | LogicalType::USmallInt
1123 | LogicalType::UInteger
1124 | LogicalType::UBigInt
1125 | LogicalType::Date
1126 | LogicalType::Timestamp
1127 )
1128 .then_some(column)
1129 })
1130 .collect::<Vec<_>>();
1131 let workers = std::thread::available_parallelism()
1132 .map_or(1, usize::from)
1133 .min(MAX_FREQUENCY_WORKERS)
1134 .min(columns.len());
1135 if workers <= 1 {
1136 let mut frequencies = vec![None; self.table.fields.len()];
1137 for column in columns {
1138 frequencies[column] = self.numeric_frequency(column)?;
1139 }
1140 return Ok(frequencies);
1141 }
1142 columns.sort_by_key(|&column| weight(&self.table.fields[column].ty));
1145 let queue = Mutex::new(columns);
1146 let pieces = std::thread::scope(|scope| {
1147 (0..workers)
1148 .map(|_| {
1149 scope.spawn(|| {
1150 let mut mine = Vec::new();
1151 loop {
1152 let taken = queue
1153 .lock()
1154 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1155 .pop();
1156 let Some(column) = taken else { break };
1157 mine.push((column, self.numeric_frequency(column)?));
1158 }
1159 Ok(mine)
1160 })
1161 })
1162 .collect::<Vec<_>>()
1163 .into_iter()
1164 .map(|handle| {
1165 handle
1166 .join()
1167 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1168 })
1169 .collect::<Result<Vec<_>>>()
1170 })?;
1171 let mut frequencies = vec![None; self.table.fields.len()];
1172 for piece in pieces {
1173 for (column, summary) in piece {
1174 frequencies[column] = summary;
1175 }
1176 }
1177 Ok(frequencies)
1178 }
1179
1180 pub fn finish(mut self) -> Result<Table> {
1186 self.flush_pending()?;
1187 let mut stripes = std::mem::take(&mut self.order)
1188 .into_iter()
1189 .zip(std::mem::take(&mut self.table.stripes))
1190 .collect::<Vec<_>>();
1191 stripes.sort_by_key(|(order, _)| order.0);
1192 let mut previous: Option<(u64, u64)> = None;
1193 for ((first, last), _) in &stripes {
1194 if previous.is_some_and(|previous| previous >= *first) {
1195 return Err(invalid("chunks did not arrive in source order"));
1196 }
1197 previous = Some(*last);
1198 }
1199 self.table.stripes = stripes.into_iter().map(|(_, stripe)| stripe).collect();
1200 self.table.frequencies = self.numeric_frequencies()?;
1201 let dictionaries = std::mem::take(&mut self.dictionaries);
1202 let orders = rankings(&dictionaries)?;
1203 for (index, (dictionary, order)) in dictionaries.into_iter().zip(orders).enumerate() {
1204 let Some(dictionary) = dictionary else { continue };
1205 self.table.frequencies[index] = Some(code_frequency(&dictionary));
1206 let encoded = encode_global_dictionary(dictionary, &order)?;
1207 let offset = self.at;
1208 self.put(&encoded.index)?;
1209 self.put(&encoded.ranks)?;
1210 for block in &encoded.payload {
1211 self.put(block)?;
1212 }
1213 let payload_len =
1214 encoded.payload.iter().try_fold(0_usize, |len, block| len.checked_add(block.len()));
1215 let length = payload_len
1216 .and_then(|len| len.checked_add(encoded.index.len()))
1217 .and_then(|len| len.checked_add(encoded.ranks.len()))
1218 .ok_or_else(|| invalid("dictionary page length overflow"))?;
1219 self.table.dictionaries[index] = Some(Page {
1220 offset,
1221 length: u32::try_from(length)
1222 .map_err(|_| invalid("dictionary page length overflow"))?,
1223 hash: checksum(&encoded.index),
1224 });
1225 }
1226 let directory = encode_directory(&self.table)?;
1227 if directory.len() > MAX_DIRECTORY {
1228 return Err(invalid("directory exceeds the configured bound"));
1229 }
1230 let offset = self.at;
1231 self.put(&directory)?;
1232 self.file.sync_all().map_err(io)?;
1233 let slot = Slot {
1234 offset,
1235 length: u32::try_from(directory.len())
1236 .map_err(|_| invalid("directory length overflow"))?,
1237 generation: self.generation,
1238 hash: checksum(&directory),
1239 };
1240 write_at(&self.file, 16, &slot.bytes())?;
1243 self.file.sync_all().map_err(io)?;
1244 Ok(self.table)
1245 }
1246}
1247
1248#[derive(Debug, Clone)]
1250pub struct Reader {
1251 file: Arc<File>,
1252 table: Arc<Table>,
1253 dictionaries: Arc<Vec<OnceLock<Arc<Vector>>>>,
1254 loading: Arc<Vec<Mutex<()>>>,
1263 opened: Arc<AtomicUsize>,
1267 sieves: Arc<Vec<Vec<SieveSlot>>>,
1271 places: Arc<Vec<Place>>,
1273 cache: Arc<Vec<Mutex<Cached>>>,
1274 pages: Arc<AtomicUsize>,
1277 indexes: Arc<AtomicUsize>,
1280 kept: Arc<AtomicUsize>,
1283 size: u64,
1285 directory: u64,
1287 opening: Opening,
1289}
1290
1291#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1303pub struct Opening {
1304 pub reads: u32,
1307 pub bytes: u64,
1309}
1310
1311#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1313pub struct Reads {
1314 pub opening: Opening,
1316 pub pages: usize,
1318 pub indexes: usize,
1320 pub dictionaries: usize,
1323}
1324
1325#[derive(Debug, Clone, Copy)]
1327struct Place {
1328 stripe: u32,
1329 part: u32,
1330 rows: u32,
1331}
1332
1333#[derive(Debug, Clone, Copy)]
1335struct PartSpan {
1336 start: usize,
1337 length: usize,
1338 hash: u64,
1339}
1340
1341#[derive(Debug, Clone)]
1347struct CachedColumn {
1348 stripe: usize,
1349 index: Arc<Vec<PartSpan>>,
1350 page: Option<Arc<Vec<u8>>>,
1351}
1352
1353#[derive(Debug, Default)]
1373struct Cached {
1374 pages: Vec<Option<Arc<Vec<u8>>>>,
1375 order: VecDeque<usize>,
1376 loading: Vec<usize>,
1377 index: Vec<Option<Arc<Vec<PartSpan>>>>,
1378}
1379
1380const CACHED_STRIPES_PER_COLUMN: usize = 4;
1392
1393type SieveSlot = OnceLock<Arc<Vec<Option<Sieve>>>>;
1395
1396#[derive(Debug)]
1397struct NativeText {
1398 file: Arc<File>,
1399 offsets: Vec<u32>,
1400 ranks: usize,
1402 rank_at: u64,
1406 rank_hashes: Vec<u64>,
1407 rank_blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1408 code_ranks: OnceLock<Option<Vec<u32>>>,
1415 payload: u64,
1416 ends: Vec<u64>,
1419 hashes: Vec<u64>,
1420 blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1422}
1423
1424const TEXT_PAYLOAD_VALUES: usize = 1024;
1440
1441const TEXT_RANK_BLOCK: usize = 512;
1451
1452const RANK_ENTRY: usize = size_of::<u64>() + size_of::<u32>();
1454
1455impl NativeText {
1456 fn payload_block(&self, block: usize) -> Result<Option<&[u8]>> {
1463 let Some(slot) = self.blocks.get(block) else { return Ok(None) };
1464 let bytes = slot
1465 .get_or_init(|| {
1466 let start = if block == 0 { 0 } else { self.ends[block - 1] };
1467 let end = self.ends[block];
1468 let len = end
1469 .checked_sub(start)
1470 .ok_or_else(|| invalid("global dictionary block ends before it starts"))?;
1471 let mut stored = vec![
1472 0;
1473 usize::try_from(len).map_err(|_| invalid(
1474 "global dictionary block does not fit in memory"
1475 ))?
1476 ];
1477 read_at(&self.file, self.payload + start, &mut stored)?;
1478 if checksum(&stored) != self.hashes[block] {
1479 return Err(invalid("global dictionary payload checksum differs"));
1480 }
1481 let first = block * TEXT_PAYLOAD_VALUES;
1482 let last = (first + TEXT_PAYLOAD_VALUES).min(self.offsets.len() - 1);
1483 let want = (self.offsets[last] - self.offsets[first]) as usize;
1484 let values = string::decode_flat(&stored)?;
1485 if values.len() != last - first {
1486 return Err(invalid("global dictionary block holds the wrong value count"));
1487 }
1488 let bytes = values.into_bytes();
1489 if bytes.len() != want {
1490 return Err(invalid("global dictionary block decodes to the wrong length"));
1491 }
1492 Ok(bytes)
1493 })
1494 .as_ref()
1495 .map_err(Clone::clone)?;
1496 Ok(Some(bytes.as_slice()))
1497 }
1498
1499 fn rank_parts(&self, rank: usize) -> Result<(&[u8], usize)> {
1506 let slot = self
1507 .rank_blocks
1508 .get(rank / TEXT_RANK_BLOCK)
1509 .ok_or_else(|| invalid("global dictionary rank is past the order"))?;
1510 let block = slot
1511 .get_or_init(|| {
1512 let first = rank / TEXT_RANK_BLOCK * TEXT_RANK_BLOCK;
1513 let len = TEXT_RANK_BLOCK.min(self.ranks - first) * RANK_ENTRY;
1514 let mut bytes = vec![0; len];
1515 read_at(&self.file, self.rank_at + (first * RANK_ENTRY) as u64, &mut bytes)?;
1516 if checksum(&bytes)
1517 != *self
1518 .rank_hashes
1519 .get(rank / TEXT_RANK_BLOCK)
1520 .ok_or_else(|| invalid("global dictionary rank block has no checksum"))?
1521 {
1522 return Err(invalid("global dictionary rank checksum differs"));
1523 }
1524 Ok(bytes)
1525 })
1526 .as_ref()
1527 .map_err(Clone::clone)?;
1528 Ok((block.as_slice(), rank % TEXT_RANK_BLOCK))
1529 }
1530
1531 fn head_at(&self, rank: usize) -> Result<u64> {
1533 let (block, within) = self.rank_parts(rank)?;
1534 let at = within * size_of::<u64>();
1535 let bytes = block
1536 .get(at..at + size_of::<u64>())
1537 .ok_or_else(|| invalid("global dictionary rank block is short of heads"))?;
1538 Ok(u64::from_le_bytes(bytes.try_into().expect("eight bytes")))
1539 }
1540}
1541
1542impl TextSource for NativeText {
1543 fn len(&self) -> usize {
1544 self.offsets.len().saturating_sub(1)
1545 }
1546
1547 fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
1548 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1549 else {
1550 return Ok(None);
1551 };
1552 if start == end {
1553 return Ok(Some(&[]));
1554 }
1555 let block = index / TEXT_PAYLOAD_VALUES;
1558 let Some(bytes) = self.payload_block(block)? else { return Ok(None) };
1559 let base = self.offsets[block * TEXT_PAYLOAD_VALUES];
1560 let within = (start - base) as usize;
1561 Ok(bytes.get(within..within + (end - start) as usize))
1562 }
1563
1564 fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
1565 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1566 else {
1567 return Ok(None);
1568 };
1569 Ok(Some((end - start) as usize))
1570 }
1571
1572 fn ranks(&self) -> Option<usize> {
1573 (self.ranks > 0).then_some(self.ranks)
1574 }
1575
1576 fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
1577 let settled = self.head_at(rank)?.cmp(&head(wanted));
1581 if settled != Ordering::Equal {
1582 return Ok(settled);
1583 }
1584 let code = self.code_at_rank(rank)?;
1585 let bytes = self
1586 .bytes_at(code as usize)?
1587 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
1588 Ok(bytes.cmp(wanted))
1589 }
1590
1591 fn code_at_rank(&self, rank: usize) -> Result<u32> {
1592 let (block, within) = self.rank_parts(rank)?;
1593 let heads = block.len() / RANK_ENTRY * size_of::<u64>();
1594 let at = heads + within * size_of::<u32>();
1595 let bytes = block
1596 .get(at..at + size_of::<u32>())
1597 .ok_or_else(|| invalid("global dictionary rank block is short of codes"))?;
1598 let code = u32::from_le_bytes(bytes.try_into().expect("four bytes"));
1599 if code as usize >= self.len() {
1600 return Err(invalid("global dictionary order names a code it does not have"));
1601 }
1602 Ok(code)
1603 }
1604
1605 fn code_ranks(&self) -> Option<&[u32]> {
1606 if self.ranks == 0 || self.ranks != self.len() {
1610 return None;
1611 }
1612 self.code_ranks
1613 .get_or_init(|| {
1614 let mut ranks = vec![u32::MAX; self.ranks];
1615 for first in (0..self.ranks).step_by(TEXT_RANK_BLOCK) {
1618 let (block, _) = self.rank_parts(first).ok()?;
1619 let heads = block.len() / RANK_ENTRY * size_of::<u64>();
1620 let codes = block.get(heads..)?;
1621 for (within, entry) in codes.chunks_exact(size_of::<u32>()).enumerate() {
1622 let code = u32::from_le_bytes(entry.try_into().ok()?) as usize;
1623 *ranks.get_mut(code)? = u32::try_from(first + within).ok()?;
1624 }
1625 }
1626 if ranks.contains(&u32::MAX) {
1627 return None;
1628 }
1629 Some(ranks)
1630 })
1631 .as_deref()
1632 }
1633
1634 fn footprint(&self) -> usize {
1635 self.offsets.capacity() * size_of::<u32>()
1636 + self
1637 .code_ranks
1638 .get()
1639 .and_then(Option::as_ref)
1640 .map_or(0, |ranks| ranks.capacity() * size_of::<u32>())
1641 + self.rank_hashes.capacity() * size_of::<u64>()
1642 + self.rank_blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1643 + self
1644 .rank_blocks
1645 .iter()
1646 .filter_map(OnceLock::get)
1647 .filter_map(|result| result.as_ref().ok())
1648 .map(Vec::capacity)
1649 .sum::<usize>()
1650 + self.blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1651 + self.hashes.capacity() * size_of::<u64>()
1652 + self.ends.capacity() * size_of::<u64>()
1653 + self
1654 .blocks
1655 .iter()
1656 .filter_map(OnceLock::get)
1657 .filter_map(|result| result.as_ref().ok())
1658 .map(Vec::capacity)
1659 .sum::<usize>()
1660 }
1661}
1662
1663fn places(table: &Table) -> Result<Vec<Place>> {
1665 let mut places = Vec::with_capacity(table.stripes.len().saturating_mul(STRIPE_PARTS));
1666 for (at, stripe) in table.stripes.iter().enumerate() {
1667 let index = u32::try_from(at).map_err(|_| invalid("too many stripes"))?;
1668 for (part, &rows) in stripe.parts.iter().enumerate() {
1669 places.push(Place {
1670 stripe: index,
1671 part: u32::try_from(part).map_err(|_| invalid("too many parts in a stripe"))?,
1672 rows,
1673 });
1674 }
1675 }
1676 Ok(places)
1677}
1678
1679fn read_index(file: &File, stripe: &Stripe, column: usize) -> Result<Vec<PartSpan>> {
1684 let parts = stripe.parts.len();
1685 let section = index_section(parts)?;
1686 let at = column.checked_mul(section).ok_or_else(|| invalid("index page offset overflow"))?;
1687 let end = at.checked_add(section).ok_or_else(|| invalid("index page offset overflow"))?;
1688 if end > stripe.index.length as usize {
1689 return Err(invalid("index page is shorter than its columns"));
1690 }
1691 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1692 let mut bytes = vec![0; section];
1693 let offset = stripe
1694 .index
1695 .offset
1696 .checked_add(at as u64)
1697 .ok_or_else(|| invalid("index page offset overflow"))?;
1698 read_at(file, offset, &mut bytes)?;
1699 let entries = section - size_of::<u64>();
1700 let stored = u64::from_le_bytes(bytes[entries..].try_into().expect("eight bytes"));
1701 if checksum(&bytes[..entries]) != stored {
1702 return Err(invalid(&format!(
1705 "index page section checksum differs, column {column} of {parts} parts at {offset}, \
1706 wanted {stored:016x} and got {:016x}",
1707 checksum(&bytes[..entries]),
1708 )));
1709 }
1710 let mut spans = Vec::with_capacity(parts);
1711 let mut start = 0_usize;
1712 for part in 0..parts {
1713 let at = part * INDEX_ENTRY;
1714 let length = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four bytes")) as usize;
1715 let hash = u64::from_le_bytes(bytes[at + 4..at + 12].try_into().expect("eight bytes"));
1716 spans.push(PartSpan { start, length, hash });
1717 start = start.checked_add(length).ok_or_else(|| invalid("column page length overflow"))?;
1718 }
1719 if start != page.length as usize {
1720 return Err(invalid("column page length differs from its index"));
1721 }
1722 Ok(spans)
1723}
1724
1725fn part_bytes(page: &[u8], span: PartSpan) -> Result<&[u8]> {
1727 let end = span.start.checked_add(span.length).ok_or_else(|| invalid("part range overflow"))?;
1728 page.get(span.start..end).ok_or_else(|| invalid("part exceeds its column page"))
1729}
1730
1731fn remember(cached: &mut Cached, held: &CachedColumn, kept: usize) {
1736 if let Some(slot) = cached.index.get_mut(held.stripe) {
1737 if slot.is_none() {
1738 *slot = Some(Arc::clone(&held.index));
1739 }
1740 }
1741 let Some(page) = held.page.clone() else { return };
1742 let Some(slot) = cached.pages.get_mut(held.stripe) else { return };
1743 if slot.is_none() {
1744 cached.order.push_back(held.stripe);
1745 }
1746 *slot = Some(page);
1747 while cached.order.len() > kept.max(1) {
1748 let Some(oldest) = cached.order.pop_front() else { break };
1749 if let Some(slot) = cached.pages.get_mut(oldest) {
1750 *slot = None;
1751 }
1752 }
1753}
1754
1755impl Reader {
1756 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1762 let mut file = File::open(path).map_err(io)?;
1763 let size = file.metadata().map_err(io)?.len();
1764 if size < HEADER {
1765 return Err(invalid("file is shorter than its header"));
1766 }
1767 let mut header = [0; HEADER as usize];
1768 file.read_exact(&mut header).map_err(io)?;
1769 let mut opening = Opening { reads: 1, bytes: HEADER };
1770 let version = u32::from_le_bytes([header[8], header[9], header[10], header[11]]);
1771 if &header[..8] != MAGIC {
1776 return Err(invalid("the header does not begin with a rudb native magic"));
1777 }
1778 if version != FORMAT {
1779 return Err(invalid(&format!(
1780 "the file is format {version} and this build reads format {FORMAT}, so it has to \
1781 be written again"
1782 )));
1783 }
1784 let mut selected = None;
1785 for start in [16, 16 + SLOT_BYTES] {
1786 let slot = Slot::read(&header[start..start + SLOT_BYTES]);
1787 if slot.generation == 0 || slot.length == 0 || slot.length as usize > MAX_DIRECTORY {
1788 continue;
1789 }
1790 let Some(end) = slot.offset.checked_add(u64::from(slot.length)) else { continue };
1791 if slot.offset < HEADER || end > size {
1792 continue;
1793 }
1794 let mut bytes = vec![0; slot.length as usize];
1795 file.seek(SeekFrom::Start(slot.offset)).map_err(io)?;
1796 file.read_exact(&mut bytes).map_err(io)?;
1797 opening.reads += 1;
1798 opening.bytes += u64::from(slot.length);
1799 if checksum(&bytes) == slot.hash
1800 && selected
1801 .as_ref()
1802 .is_none_or(|(old, _): &(Slot, Vec<u8>)| old.generation < slot.generation)
1803 {
1804 selected = Some((slot, bytes));
1805 }
1806 }
1807 let (slot, bytes) =
1808 selected.ok_or_else(|| invalid("no committed directory slot is valid"))?;
1809 let table = decode_directory(&bytes, size)?;
1810 let places = places(&table)?;
1811 let dictionaries = (0..table.fields.len()).map(|_| OnceLock::new()).collect();
1812 let table_fields = table.fields.len();
1813 let stripes = table.stripes.len();
1814 let cache = (0..table.fields.len())
1815 .map(|_| {
1816 Mutex::new(Cached {
1817 pages: (0..stripes).map(|_| None).collect(),
1818 index: (0..stripes).map(|_| None).collect(),
1819 ..Cached::default()
1820 })
1821 })
1822 .collect::<Vec<_>>();
1823 let sieves = (0..table.fields.len())
1824 .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
1825 .collect();
1826 Ok(Self {
1827 file: Arc::new(file),
1828 table: Arc::new(table),
1829 dictionaries: Arc::new(dictionaries),
1830 loading: Arc::new((0..table_fields).map(|_| Mutex::new(())).collect()),
1831 opened: Arc::new(AtomicUsize::new(0)),
1832 sieves: Arc::new(sieves),
1833 places: Arc::new(places),
1834 cache: Arc::new(cache),
1835 pages: Arc::new(AtomicUsize::new(0)),
1836 indexes: Arc::new(AtomicUsize::new(0)),
1837 kept: Arc::new(AtomicUsize::new(CACHED_STRIPES_PER_COLUMN)),
1838 size,
1839 directory: u64::from(slot.length),
1840 opening,
1841 })
1842 }
1843
1844 #[must_use]
1851 pub fn reads(&self) -> Reads {
1852 Reads {
1853 opening: self.opening,
1854 pages: self.pages.load(Atomic::Relaxed),
1855 indexes: self.indexes.load(Atomic::Relaxed),
1856 dictionaries: self.opened.load(Atomic::Relaxed),
1857 }
1858 }
1859
1860 #[must_use]
1865 pub fn layout(&self) -> Layout {
1866 let table = &self.table;
1867 let stripes = table.stripes.as_slice();
1868 let columns = table
1869 .fields
1870 .iter()
1871 .enumerate()
1872 .map(|(at, field)| ColumnLayout {
1873 name: field.name.clone(),
1874 kind: field.ty.to_string(),
1875 pages: sum(stripes.iter().map(|stripe| span_bytes(&stripe.pages, at))),
1876 memberships: sum(stripes.iter().map(|stripe| page_bytes(&stripe.memberships, at))),
1877 sieves: sum(stripes.iter().map(|stripe| page_bytes(&stripe.sieves, at))),
1878 dictionary: page_bytes(&table.dictionaries, at),
1879 })
1880 .collect();
1881 Layout {
1882 file: self.size,
1883 rows: table.rows,
1884 stripes: stripes.len(),
1885 parts: self.places.len(),
1886 columns,
1887 indexes: sum(stripes.iter().map(|stripe| u64::from(stripe.index.length))),
1888 directory: self.directory,
1889 header: HEADER,
1890 }
1891 }
1892
1893 #[must_use]
1895 pub fn parts(&self) -> usize {
1896 self.places.len()
1897 }
1898
1899 #[must_use]
1906 pub fn stripe_parts(&self) -> Vec<std::ops::Range<usize>> {
1907 let mut runs = Vec::with_capacity(self.table.stripes.len());
1908 let mut start = 0;
1909 for stripe in &self.table.stripes {
1910 let end = start + stripe.parts.len();
1911 runs.push(start..end);
1912 start = end;
1913 }
1914 runs
1915 }
1916
1917 pub fn keep_stripes(&self, stripes: usize) {
1924 self.kept.fetch_max(stripes, Atomic::Relaxed);
1925 }
1926
1927 #[must_use]
1929 pub fn part_rows(&self, at: usize) -> usize {
1930 self.places.get(at).map_or(0, |place| place.rows as usize)
1931 }
1932
1933 #[must_use]
1935 pub fn table(&self) -> &Table {
1936 &self.table
1937 }
1938
1939 pub fn top_frequencies(&self, column: usize, top: usize) -> Result<Option<Vec<(Value, u64)>>> {
1948 let field = self
1949 .table
1950 .fields
1951 .get(column)
1952 .ok_or_else(|| invalid("frequency column index out of range"))?;
1953 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1954 return Ok(None);
1955 };
1956 if top == 0 || summary.entries.len() < top {
1957 return Ok(None);
1958 }
1959 let boundary = summary.entries[top - 1].count;
1960 if boundary <= summary.omitted_max {
1961 return Ok(None);
1962 }
1963 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
1964 }
1965
1966 pub fn exact_frequencies(&self, column: usize) -> Result<Option<Vec<(Value, u64)>>> {
1986 let field = self
1987 .table
1988 .fields
1989 .get(column)
1990 .ok_or_else(|| invalid("frequency column index out of range"))?;
1991 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1992 return Ok(None);
1993 };
1994 if summary.omitted_max > 0 {
1995 return Ok(None);
1996 }
1997 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
1998 }
1999
2000 fn decode_frequencies(
2002 &self,
2003 column: usize,
2004 ty: &LogicalType,
2005 entries: &[FrequencyEntry],
2006 ) -> Result<Vec<(Value, u64)>> {
2007 let dictionary = if *ty == LogicalType::Varchar { self.dictionary(column)? } else { None };
2008 let mut out = Vec::with_capacity(entries.len());
2009 for entry in entries {
2010 let value = match entry.value {
2011 FrequencyValue::Null => Value::Null,
2012 FrequencyValue::Integer(value) => match *ty {
2013 LogicalType::TinyInt => Value::TinyInt(
2014 i8::try_from(value)
2015 .map_err(|_| invalid("frequency TINYINT is out of range"))?,
2016 ),
2017 LogicalType::UTinyInt => Value::UTinyInt(
2018 u8::try_from(value)
2019 .map_err(|_| invalid("frequency UTINYINT is out of range"))?,
2020 ),
2021 LogicalType::USmallInt => Value::USmallInt(
2022 u16::try_from(value)
2023 .map_err(|_| invalid("frequency USMALLINT is out of range"))?,
2024 ),
2025 LogicalType::UInteger => Value::UInteger(
2026 u32::try_from(value)
2027 .map_err(|_| invalid("frequency UINTEGER is out of range"))?,
2028 ),
2029 LogicalType::UBigInt => Value::UBigInt(
2030 u64::try_from(value)
2031 .map_err(|_| invalid("frequency UBIGINT is out of range"))?,
2032 ),
2033 LogicalType::SmallInt => Value::SmallInt(
2034 i16::try_from(value)
2035 .map_err(|_| invalid("frequency SMALLINT is out of range"))?,
2036 ),
2037 LogicalType::Integer => Value::Integer(
2038 i32::try_from(value)
2039 .map_err(|_| invalid("frequency INTEGER is out of range"))?,
2040 ),
2041 LogicalType::BigInt => Value::BigInt(
2042 i64::try_from(value)
2043 .map_err(|_| invalid("frequency BIGINT is out of range"))?,
2044 ),
2045 LogicalType::Date => Value::Date(
2046 i32::try_from(value)
2047 .map_err(|_| invalid("frequency DATE is out of range"))?,
2048 ),
2049 LogicalType::Timestamp => Value::Timestamp(
2050 i64::try_from(value)
2051 .map_err(|_| invalid("frequency TIMESTAMP is out of range"))?,
2052 ),
2053 _ => return Err(invalid("integer frequency belongs to another type")),
2054 },
2055 FrequencyValue::Code(code) => dictionary
2056 .as_ref()
2057 .ok_or_else(|| invalid("frequency code has no dictionary"))?
2058 .try_value_at(code as usize)?,
2059 };
2060 out.push((value, entry.count));
2061 }
2062 Ok(out)
2063 }
2064
2065 pub fn frequency_occurrences(&self, column: usize) -> Result<Option<FrequencyOccurrences>> {
2075 self.table
2076 .fields
2077 .get(column)
2078 .ok_or_else(|| invalid("frequency column index out of range"))?;
2079 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2080 return Ok(None);
2081 };
2082 if summary.ordinals.is_empty() {
2083 return Ok(None);
2084 }
2085 Ok(Some(FrequencyOccurrences {
2086 omitted_max: summary.omitted_max,
2087 ordinals: summary.ordinals.clone(),
2088 }))
2089 }
2090
2091 pub fn distinct_values(&self, column: usize) -> Result<Option<u64>> {
2111 if self.null_count(column)? > 0 {
2112 return Ok(None);
2113 }
2114 Ok(self.dictionary(column)?.map(|dictionary| dictionary.len() as u64))
2115 }
2116
2117 pub fn null_count(&self, column: usize) -> Result<u64> {
2128 if column >= self.table.fields.len() {
2129 return Err(invalid("null count column index out of range"));
2130 }
2131 let mut nulls = 0_u64;
2132 for stripe in &self.table.stripes {
2133 let range = stripe
2134 .zone
2135 .column(column)
2136 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2137 nulls = nulls
2138 .checked_add(range.nulls as u64)
2139 .ok_or_else(|| invalid("null count overflow"))?;
2140 }
2141 Ok(nulls)
2142 }
2143
2144 pub fn text_extremes(&self, column: usize) -> Result<Option<(Value, Value)>> {
2159 if self.null_count(column)? > 0 {
2160 return Ok(None);
2161 }
2162 let Some(dictionary) = self.dictionary(column)? else { return Ok(None) };
2163 let Some(ranks) = dictionary.ranks() else { return Ok(None) };
2164 if ranks == 0 {
2165 return Ok(None);
2166 }
2167 let low = text_at_rank(&dictionary, 0)?;
2168 let high = text_at_rank(&dictionary, ranks - 1)?;
2169 Ok(Some((low, high)))
2170 }
2171
2172 pub fn exact_extremes(&self, column: usize) -> Result<Option<(Bound, Bound)>> {
2195 if column >= self.table.fields.len() {
2196 return Err(invalid("extremes column index out of range"));
2197 }
2198 let mut low: Option<Bound> = None;
2199 let mut high: Option<Bound> = None;
2200 for stripe in &self.table.stripes {
2201 let range = stripe
2202 .zone
2203 .column(column)
2204 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2205 if !range.exact {
2206 return Ok(None);
2207 }
2208 let (Some(small), Some(large)) = (range.low.as_ref(), range.high.as_ref()) else {
2213 if stripe.rows > range.nulls {
2214 return Ok(None);
2215 }
2216 continue;
2217 };
2218 low = Some(low.map_or_else(|| small.clone(), |held| held.smaller(small.clone())));
2219 high = Some(high.map_or_else(|| large.clone(), |held| held.larger(large.clone())));
2220 }
2221 Ok(low.zip(high))
2222 }
2223
2224 pub fn exact_sum(&self, column: usize) -> Result<Option<(i128, u64)>> {
2237 if column >= self.table.fields.len() {
2238 return Err(invalid("sum column index out of range"));
2239 }
2240 let mut total = 0_i128;
2241 let mut rows = 0_u64;
2242 for stripe in &self.table.stripes {
2243 let range = stripe
2244 .zone
2245 .column(column)
2246 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2247 let Some(part) = range.sum else { return Ok(None) };
2248 let Some(sum) = total.checked_add(part) else { return Ok(None) };
2249 total = sum;
2250 rows = rows.saturating_add(stripe.rows as u64 - range.nulls as u64);
2251 }
2252 Ok(Some((total, rows)))
2253 }
2254
2255 fn dictionary(&self, column: usize) -> Result<Option<Arc<Vector>>> {
2264 let Some(page) = self.table.dictionaries[column] else { return Ok(None) };
2265 if let Some(dictionary) = self.dictionaries[column].get() {
2266 return Ok(Some(Arc::clone(dictionary)));
2267 }
2268 let _queued = self.loading[column].lock().map_err(|_| invalid("a poisoned dictionary"))?;
2269 if let Some(dictionary) = self.dictionaries[column].get() {
2270 return Ok(Some(Arc::clone(dictionary)));
2271 }
2272 self.opened.fetch_add(1, Atomic::Relaxed);
2273 let dictionary = Arc::new(open_global_dictionary(
2274 Arc::clone(&self.file),
2275 page,
2276 &self.table.fields[column].ty,
2277 )?);
2278 let _ = self.dictionaries[column].set(Arc::clone(&dictionary));
2279 Ok(Some(dictionary))
2280 }
2281
2282 pub fn read(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2291 self.read_impl(part, columns, true)
2292 }
2293
2294 pub fn read_sparse(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2304 self.read_impl(part, columns, false)
2305 }
2306
2307 pub fn skips_codes(&self, part: usize, column: usize, candidates: &[u32]) -> Result<bool> {
2314 if candidates.is_empty() {
2315 return Ok(true);
2316 }
2317 if candidates.windows(2).any(|pair| pair[0] >= pair[1]) {
2318 return Err(Error::internal("native code candidates are not sorted and unique"));
2319 }
2320 let stripe = self.stripe_of(part)?;
2321 let Some(page) = stripe.memberships.get(column).copied().flatten() else {
2322 return Ok(false);
2323 };
2324 let mut bytes = vec![0; page.length as usize];
2325 read_at(&self.file, page.offset, &mut bytes)?;
2326 if checksum(&bytes) != page.hash {
2327 return Err(invalid("membership page checksum differs"));
2328 }
2329 let codes = decode_membership(&bytes)?;
2330 let mut left = 0;
2331 let mut right = 0;
2332 while left < codes.len() && right < candidates.len() {
2333 match codes[left].cmp(&candidates[right]) {
2334 Ordering::Less => left += 1,
2335 Ordering::Greater => right += 1,
2336 Ordering::Equal => return Ok(false),
2337 }
2338 }
2339 Ok(true)
2340 }
2341
2342 fn stripe_of(&self, part: usize) -> Result<&Stripe> {
2343 let place = self.places.get(part).ok_or_else(|| invalid("part index out of range"))?;
2344 self.table
2345 .stripes
2346 .get(place.stripe as usize)
2347 .ok_or_else(|| invalid("stripe index out of range"))
2348 }
2349
2350 fn held(&self, at: usize, stripe: &Stripe, column: usize, whole: bool) -> Result<CachedColumn> {
2367 let cache = self.cache.get(column).ok_or_else(|| invalid("column index out of range"))?;
2368 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2369 let known = cached.index.get(at).and_then(Clone::clone);
2370 let page = cached.pages.get(at).and_then(Clone::clone);
2371 if let Some(index) = known.clone() {
2372 if !whole || page.is_some() {
2373 return Ok(CachedColumn { stripe: at, index, page });
2374 }
2375 }
2376 if cached.loading.contains(&at) {
2377 drop(cached);
2378 if let Some(index) = known {
2382 return Ok(CachedColumn { stripe: at, index, page: None });
2383 }
2384 let held = self.page_of(stripe, column, at, false, None)?;
2385 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2386 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2387 return Ok(held);
2388 }
2389 cached.loading.push(at);
2390 drop(cached);
2391
2392 let read = self.page_of(stripe, column, at, whole, known);
2393
2394 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2398 if let Some(position) = cached.loading.iter().position(|loading| *loading == at) {
2399 cached.loading.remove(position);
2400 }
2401 let held = read?;
2402 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2403 Ok(held)
2404 }
2405
2406 fn page_of(
2412 &self,
2413 stripe: &Stripe,
2414 column: usize,
2415 at: usize,
2416 whole: bool,
2417 known: Option<Arc<Vec<PartSpan>>>,
2418 ) -> Result<CachedColumn> {
2419 let index = match known {
2420 Some(index) => index,
2421 None => {
2422 self.indexes.fetch_add(1, Atomic::Relaxed);
2423 Arc::new(read_index(&self.file, stripe, column)?)
2424 }
2425 };
2426 let page = if whole {
2427 self.pages.fetch_add(1, Atomic::Relaxed);
2428 let span = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2429 let mut bytes = vec![0; span.length as usize];
2430 read_at(&self.file, span.offset, &mut bytes)?;
2431 Some(Arc::new(bytes))
2432 } else {
2433 None
2434 };
2435 Ok(CachedColumn { stripe: at, index, page })
2436 }
2437
2438 fn read_impl(&self, at: usize, columns: &[usize], whole: bool) -> Result<Chunk> {
2439 let place = *self.places.get(at).ok_or_else(|| invalid("part index out of range"))?;
2440 let index = place.stripe as usize;
2441 let stripe =
2442 self.table.stripes.get(index).ok_or_else(|| invalid("stripe index out of range"))?;
2443 let rows = place.rows as usize;
2444 let mut picked = Vec::with_capacity(columns.len());
2445 for &column in columns {
2446 let field = self
2447 .table
2448 .fields
2449 .get(column)
2450 .ok_or_else(|| invalid("column index out of range"))?;
2451 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2452 let held = self.held(index, stripe, column, whole)?;
2453 let span = *held
2454 .index
2455 .get(place.part as usize)
2456 .ok_or_else(|| invalid("part index out of range"))?;
2457 let owned;
2458 let bytes = match &held.page {
2459 Some(held) => part_bytes(held, span)?,
2460 None => {
2461 let offset = page
2462 .offset
2463 .checked_add(span.start as u64)
2464 .ok_or_else(|| invalid("part range overflow"))?;
2465 let mut bytes = vec![0; span.length];
2466 read_at(&self.file, offset, &mut bytes)?;
2467 owned = bytes;
2468 &owned
2469 }
2470 };
2471 if checksum(bytes) != span.hash {
2472 return Err(invalid(&format!(
2473 "column page checksum differs, column {column} part {} at {}+{} of {} bytes, \
2474 wanted {:016x} and got {:016x}",
2475 place.part,
2476 page.offset,
2477 span.start,
2478 span.length,
2479 span.hash,
2480 checksum(bytes),
2481 )));
2482 }
2483 let dictionary = self.dictionary(column)?;
2484 picked.push(decode(&field.ty, rows, bytes, dictionary)?);
2485 }
2486 Chunk::with_rows(picked, rows)
2487 }
2488
2489 #[must_use]
2499 pub fn skips(&self, part: usize, probes: &[Probe]) -> bool {
2500 let Some(place) = self.places.get(part).copied() else { return false };
2501 let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
2502 if stripe.zone.skips(probes) {
2503 return true;
2504 }
2505 probes.iter().any(|probe| self.sifted(place, probe))
2506 }
2507
2508 #[must_use]
2519 pub fn stripe_skips(&self, stripe: usize, probes: &[Probe]) -> bool {
2520 self.table.stripes.get(stripe).is_some_and(|held| held.zone.skips(probes))
2521 }
2522
2523 fn sifted(&self, place: Place, probe: &Probe) -> bool {
2529 if probe.op != Op::Equal {
2530 return false;
2531 }
2532 match self.stripe_sieves(place.stripe as usize, probe.column) {
2533 Some(sieves) => sieves
2534 .get(place.part as usize)
2535 .and_then(Option::as_ref)
2536 .is_some_and(|sieve| sieve.excludes(&probe.value)),
2537 None => false,
2538 }
2539 }
2540
2541 fn stripe_sieves(&self, stripe: usize, column: usize) -> Option<&[Option<Sieve>]> {
2548 let slot = self.sieves.get(column)?.get(stripe)?;
2549 if let Some(held) = slot.get() {
2550 return Some(held);
2551 }
2552 let page = self.table.stripes.get(stripe)?.sieves.get(column).copied().flatten()?;
2553 let mut bytes = vec![0; page.length as usize];
2554 read_at(&self.file, page.offset, &mut bytes).ok()?;
2555 if checksum(&bytes) != page.hash {
2556 return None;
2557 }
2558 let sieves = Arc::new(decode_sieves(&bytes).ok()?);
2559 let _ = slot.set(sieves);
2560 slot.get().map(|held| held.as_slice())
2561 }
2562}
2563
2564fn text_at_rank(dictionary: &Vector, rank: usize) -> Result<Value> {
2566 let code = dictionary.code_at_rank(rank)? as usize;
2567 let text = dictionary
2568 .try_text_at(code)?
2569 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
2570 Ok(Value::Varchar(text.into()))
2571}
2572
2573#[cfg(unix)]
2578fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2579 use std::os::unix::fs::FileExt;
2580 while !bytes.is_empty() {
2581 let written = file.write_at(bytes, offset).map_err(io)?;
2582 if written == 0 {
2583 return Err(invalid("a write to the native file wrote nothing"));
2584 }
2585 offset += written as u64;
2586 bytes = &bytes[written..];
2587 }
2588 Ok(())
2589}
2590
2591#[cfg(windows)]
2593fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2594 use std::os::windows::fs::FileExt;
2595 while !bytes.is_empty() {
2596 let written = file.seek_write(bytes, offset).map_err(io)?;
2597 if written == 0 {
2598 return Err(invalid("a write to the native file wrote nothing"));
2599 }
2600 offset += written as u64;
2601 bytes = &bytes[written..];
2602 }
2603 Ok(())
2604}
2605
2606#[cfg(not(any(unix, windows)))]
2608fn write_at(file: &File, offset: u64, bytes: &[u8]) -> Result<()> {
2609 use std::io::Write;
2610 let mut file = file.try_clone().map_err(io)?;
2611 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2612 file.write_all(bytes).map_err(io)
2613}
2614
2615#[cfg(unix)]
2625fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2626 use std::os::unix::fs::FileExt;
2627 while !bytes.is_empty() {
2628 let read = file.read_at(bytes, offset).map_err(io)?;
2629 if read == 0 {
2630 return Err(invalid("column page ends before its declared length"));
2631 }
2632 offset += read as u64;
2633 bytes = &mut bytes[read..];
2634 }
2635 Ok(())
2636}
2637
2638#[cfg(windows)]
2644fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2645 use std::os::windows::fs::FileExt;
2646 while !bytes.is_empty() {
2647 let read = file.seek_read(bytes, offset).map_err(io)?;
2648 if read == 0 {
2649 return Err(invalid("column page ends before its declared length"));
2650 }
2651 offset += read as u64;
2652 bytes = &mut bytes[read..];
2653 }
2654 Ok(())
2655}
2656
2657#[cfg(not(any(unix, windows)))]
2662fn read_at(file: &File, offset: u64, bytes: &mut [u8]) -> Result<()> {
2663 let mut file = file.try_clone().map_err(io)?;
2664 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2665 file.read_exact(bytes).map_err(io)
2666}
2667
2668fn type_tag(ty: &LogicalType) -> Result<u8> {
2669 match ty {
2670 LogicalType::SmallInt => Ok(1),
2671 LogicalType::Integer => Ok(2),
2672 LogicalType::BigInt => Ok(3),
2673 LogicalType::Varchar => Ok(4),
2674 LogicalType::Date => Ok(5),
2675 LogicalType::Timestamp => Ok(6),
2676 LogicalType::Boolean => Ok(7),
2677 LogicalType::TinyInt => Ok(8),
2678 LogicalType::UTinyInt => Ok(9),
2679 LogicalType::USmallInt => Ok(10),
2680 LogicalType::UInteger => Ok(11),
2681 LogicalType::UBigInt => Ok(12),
2682 _ => Err(Error::not_implemented(format!("native storage for {ty}"))),
2683 }
2684}
2685
2686fn tag_type(tag: u8) -> Result<LogicalType> {
2687 match tag {
2688 1 => Ok(LogicalType::SmallInt),
2689 2 => Ok(LogicalType::Integer),
2690 3 => Ok(LogicalType::BigInt),
2691 4 => Ok(LogicalType::Varchar),
2692 5 => Ok(LogicalType::Date),
2693 6 => Ok(LogicalType::Timestamp),
2694 7 => Ok(LogicalType::Boolean),
2695 8 => Ok(LogicalType::TinyInt),
2696 9 => Ok(LogicalType::UTinyInt),
2697 10 => Ok(LogicalType::USmallInt),
2698 11 => Ok(LogicalType::UInteger),
2699 12 => Ok(LogicalType::UBigInt),
2700 _ => Err(invalid("column type tag is unknown")),
2701 }
2702}
2703
2704fn put_u16(out: &mut Vec<u8>, value: u16) {
2705 out.extend_from_slice(&value.to_le_bytes());
2706}
2707fn put_u32(out: &mut Vec<u8>, value: u32) {
2708 out.extend_from_slice(&value.to_le_bytes());
2709}
2710fn put_u64(out: &mut Vec<u8>, value: u64) {
2711 out.extend_from_slice(&value.to_le_bytes());
2712}
2713fn put_var_u64(out: &mut Vec<u8>, mut value: u64) {
2714 while value >= 0x80 {
2715 out.push((value as u8 & 0x7f) | 0x80);
2716 value >>= 7;
2717 }
2718 out.push(value as u8);
2719}
2720
2721fn frequency_order(left: FrequencyValue, right: FrequencyValue) -> Ordering {
2722 match (left, right) {
2723 (FrequencyValue::Null, FrequencyValue::Null) => Ordering::Equal,
2724 (FrequencyValue::Null, _) => Ordering::Less,
2725 (_, FrequencyValue::Null) => Ordering::Greater,
2726 (FrequencyValue::Integer(left), FrequencyValue::Integer(right)) => left.cmp(&right),
2727 (FrequencyValue::Code(left), FrequencyValue::Code(right)) => left.cmp(&right),
2728 (FrequencyValue::Integer(_), FrequencyValue::Code(_)) => Ordering::Less,
2729 (FrequencyValue::Code(_), FrequencyValue::Integer(_)) => Ordering::Greater,
2730 }
2731}
2732
2733fn code_frequency(dictionary: &GlobalDictionary) -> FrequencySummary {
2734 let mut entries = dictionary
2735 .counts
2736 .iter()
2737 .enumerate()
2738 .filter(|(_, count)| **count != 0)
2739 .map(|(code, &count)| FrequencyEntry { value: FrequencyValue::Code(code as u32), count })
2740 .collect::<Vec<_>>();
2741 if dictionary.nulls != 0 {
2742 entries.push(FrequencyEntry { value: FrequencyValue::Null, count: dictionary.nulls });
2743 }
2744 entries.sort_unstable_by(|left, right| {
2745 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
2746 });
2747 let omitted_max = entries.get(FREQUENCY_ENTRIES).map_or(0, |entry| entry.count);
2748 entries.truncate(FREQUENCY_ENTRIES);
2749 FrequencySummary { entries, omitted_max, ordinals: Vec::new() }
2750}
2751
2752fn encode_directory(table: &Table) -> Result<Vec<u8>> {
2753 let mut out = DIRECTORY.to_vec();
2754 let name = table.name.as_bytes();
2755 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
2756 out.extend_from_slice(name);
2757 put_u16(&mut out, u16::try_from(table.fields.len()).map_err(|_| invalid("too many columns"))?);
2758 for field in &table.fields {
2759 let name = field.name.as_bytes();
2760 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?);
2761 out.extend_from_slice(name);
2762 out.push(type_tag(&field.ty)?);
2763 out.push(u8::from(field.not_null));
2764 }
2765 for dictionary in &table.dictionaries {
2766 match dictionary {
2767 None => out.push(0),
2768 Some(page) => {
2769 out.push(1);
2770 put_u64(&mut out, page.offset);
2771 put_u32(&mut out, page.length);
2772 put_u64(&mut out, page.hash);
2773 }
2774 }
2775 }
2776 put_u64(&mut out, u64::try_from(table.rows).map_err(|_| invalid("row count overflow"))?);
2777 put_u32(&mut out, u32::try_from(table.stripes.len()).map_err(|_| invalid("too many stripes"))?);
2778 for stripe in &table.stripes {
2779 put_u32(
2780 &mut out,
2781 u32::try_from(stripe.parts.len()).map_err(|_| invalid("too many parts in a stripe"))?,
2782 );
2783 for &rows in &stripe.parts {
2784 put_u32(&mut out, rows);
2785 }
2786 put_u64(&mut out, stripe.index.offset);
2787 put_u32(&mut out, stripe.index.length);
2788 for page in &stripe.pages {
2789 put_u64(&mut out, page.offset);
2790 put_u32(&mut out, page.length);
2791 }
2792 for (field, membership) in table.fields.iter().zip(&stripe.memberships) {
2793 if field.ty != LogicalType::Varchar {
2794 continue;
2795 }
2796 let page =
2797 membership.ok_or_else(|| invalid("string page has no code membership index"))?;
2798 put_u64(&mut out, page.offset);
2799 put_u32(&mut out, page.length);
2800 put_u64(&mut out, page.hash);
2801 }
2802 for sieve in &stripe.sieves {
2803 match sieve {
2804 None => out.push(0),
2805 Some(page) => {
2806 out.push(1);
2807 put_u64(&mut out, page.offset);
2808 put_u32(&mut out, page.length);
2809 put_u64(&mut out, page.hash);
2810 }
2811 }
2812 }
2813 for range in stripe.zone.columns() {
2814 put_bound(&mut out, range.low.as_ref())?;
2815 put_bound(&mut out, range.high.as_ref())?;
2816 put_u32(
2817 &mut out,
2818 u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?,
2819 );
2820 out.push(u8::from(range.exact));
2821 match range.sum {
2822 None => out.push(0),
2823 Some(total) => {
2824 out.push(1);
2825 out.extend_from_slice(&total.to_le_bytes());
2826 }
2827 }
2828 }
2829 }
2830 out.extend_from_slice(FREQUENCIES);
2831 put_u16(
2832 &mut out,
2833 u16::try_from(table.frequencies.len())
2834 .map_err(|_| invalid("too many frequency columns"))?,
2835 );
2836 for summary in &table.frequencies {
2837 let Some(summary) = summary else {
2838 out.push(0);
2839 continue;
2840 };
2841 out.push(1);
2842 put_u64(&mut out, summary.omitted_max);
2843 put_u32(
2844 &mut out,
2845 u32::try_from(summary.entries.len())
2846 .map_err(|_| invalid("too many frequency entries"))?,
2847 );
2848 for entry in &summary.entries {
2849 match entry.value {
2850 FrequencyValue::Null => out.push(0),
2851 FrequencyValue::Integer(value) => {
2852 out.push(1);
2853 out.extend_from_slice(&value.to_le_bytes());
2854 }
2855 FrequencyValue::Code(value) => {
2856 out.push(2);
2857 put_u32(&mut out, value);
2858 }
2859 }
2860 put_u64(&mut out, entry.count);
2861 }
2862 put_u32(
2863 &mut out,
2864 u32::try_from(summary.ordinals.len())
2865 .map_err(|_| invalid("too many frequency ordinals"))?,
2866 );
2867 let mut previous = 0_u64;
2868 for (at, &ordinal) in summary.ordinals.iter().enumerate() {
2869 let delta = if at == 0 {
2870 ordinal
2871 } else {
2872 ordinal
2873 .checked_sub(previous)
2874 .ok_or_else(|| invalid("frequency ordinals are not ordered"))?
2875 };
2876 if at != 0 && delta == 0 {
2877 return Err(invalid("frequency ordinals are not unique"));
2878 }
2879 put_var_u64(&mut out, delta);
2880 previous = ordinal;
2881 }
2882 }
2883 Ok(out)
2884}
2885
2886struct Cursor<'a> {
2887 bytes: &'a [u8],
2888 at: usize,
2889}
2890impl<'a> Cursor<'a> {
2891 fn take(&mut self, len: usize) -> Result<&'a [u8]> {
2892 let end = self.at.checked_add(len).ok_or_else(|| invalid("directory offset overflow"))?;
2893 let bytes =
2894 self.bytes.get(self.at..end).ok_or_else(|| invalid("directory is truncated"))?;
2895 self.at = end;
2896 Ok(bytes)
2897 }
2898 fn u8(&mut self) -> Result<u8> {
2899 Ok(self.take(1)?[0])
2900 }
2901 fn u16(&mut self) -> Result<u16> {
2902 Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("two bytes")))
2903 }
2904 fn u32(&mut self) -> Result<u32> {
2905 Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("four bytes")))
2906 }
2907 fn u64(&mut self) -> Result<u64> {
2908 Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("eight bytes")))
2909 }
2910 fn var_u64(&mut self) -> Result<u64> {
2911 let mut value = 0_u64;
2912 for shift in (0..=63).step_by(7) {
2913 let byte = self.u8()?;
2914 let part = u64::from(byte & 0x7f);
2915 if shift == 63 && part > 1 {
2916 return Err(invalid("frequency ordinal varint overflows"));
2917 }
2918 value |= part << shift;
2919 if byte & 0x80 == 0 {
2920 return Ok(value);
2921 }
2922 }
2923 Err(invalid("frequency ordinal varint is too long"))
2924 }
2925 fn bound(&mut self) -> Result<Option<Bound>> {
2926 Ok(match self.u8()? {
2927 0 => None,
2928 1 => Some(Bound::Int(i128::from_le_bytes(
2929 self.take(16)?.try_into().expect("sixteen bytes"),
2930 ))),
2931 2 => Some(Bound::Real(f64::from_le_bytes(
2932 self.take(8)?.try_into().expect("eight bytes"),
2933 ))),
2934 3 => {
2935 let length = self.u32()? as usize;
2936 Some(Bound::Bytes(self.take(length)?.to_vec()))
2937 }
2938 _ => return Err(invalid("bound tag differs")),
2939 })
2940 }
2941 fn text(&mut self) -> Result<String> {
2942 let len = self.u16()? as usize;
2943 String::from_utf8(self.take(len)?.to_vec()).map_err(|_| invalid("name is not UTF-8"))
2944 }
2945}
2946
2947fn decode_directory(bytes: &[u8], size: u64) -> Result<Table> {
2948 let mut cur = Cursor { bytes, at: 0 };
2949 if cur.take(8)? != DIRECTORY {
2950 return Err(invalid("directory magic differs"));
2951 }
2952 let name = cur.text()?;
2953 let width = cur.u16()? as usize;
2954 let mut fields = Vec::with_capacity(width);
2955 for _ in 0..width {
2956 let name = cur.text()?;
2957 let ty = tag_type(cur.u8()?)?;
2958 let not_null = match cur.u8()? {
2959 0 => false,
2960 1 => true,
2961 _ => return Err(invalid("nullability flag differs")),
2962 };
2963 fields.push(Field { name, ty, not_null });
2964 }
2965 let mut dictionaries = Vec::with_capacity(width);
2966 for _ in 0..width {
2967 dictionaries.push(match cur.u8()? {
2968 0 => None,
2969 1 => {
2970 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2971 let end = page
2972 .offset
2973 .checked_add(u64::from(page.length))
2974 .ok_or_else(|| invalid("dictionary page offset overflow"))?;
2975 if page.offset < HEADER || end > size {
2980 return Err(invalid("dictionary page range is outside the file"));
2981 }
2982 Some(page)
2983 }
2984 _ => return Err(invalid("dictionary page tag differs")),
2985 });
2986 }
2987 let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
2988 let count = cur.u32()? as usize;
2989 let mut stripes = Vec::with_capacity(count);
2990 let mut total = 0_usize;
2991 for _ in 0..count {
2992 let count = cur.u32()? as usize;
2993 if count == 0 || count > STRIPE_PARTS {
2994 return Err(invalid("stripe part count is outside its bound"));
2995 }
2996 let mut parts = Vec::with_capacity(count);
2997 let mut stripe_rows = 0_usize;
2998 for _ in 0..count {
2999 let rows = cur.u32()?;
3000 if rows == 0 {
3001 return Err(invalid("empty part"));
3002 }
3003 parts.push(rows);
3004 stripe_rows = stripe_rows
3005 .checked_add(rows as usize)
3006 .ok_or_else(|| invalid("stripe row count overflow"))?;
3007 }
3008 total =
3009 total.checked_add(stripe_rows).ok_or_else(|| invalid("stripe row count overflow"))?;
3010 let index = Span { offset: cur.u64()?, length: cur.u32()? };
3011 let section = index_section(count)?;
3012 let wanted = section
3013 .checked_mul(width)
3014 .and_then(|bytes| u32::try_from(bytes).ok())
3015 .ok_or_else(|| invalid("index page length overflow"))?;
3016 let end = index
3017 .offset
3018 .checked_add(u64::from(index.length))
3019 .ok_or_else(|| invalid("index page offset overflow"))?;
3020 if index.offset < HEADER || end > size || index.length != wanted {
3021 return Err(invalid("index page range is outside the file"));
3022 }
3023 let mut pages = Vec::with_capacity(width);
3024 for _ in 0..width {
3025 let offset = cur.u64()?;
3026 let length = cur.u32()?;
3027 let end = offset
3028 .checked_add(u64::from(length))
3029 .ok_or_else(|| invalid("page offset overflow"))?;
3030 if offset < HEADER || end > size || length as usize > MAX_PAGE {
3031 return Err(invalid("page range is outside the file"));
3032 }
3033 pages.push(Span { offset, length });
3034 }
3035 let mut memberships = vec![None; width];
3036 for (column, field) in fields.iter().enumerate() {
3037 if field.ty != LogicalType::Varchar {
3038 continue;
3039 }
3040 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3041 let end = page
3042 .offset
3043 .checked_add(u64::from(page.length))
3044 .ok_or_else(|| invalid("membership page offset overflow"))?;
3045 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3046 return Err(invalid("membership page range is outside the file"));
3047 }
3048 memberships[column] = Some(page);
3049 }
3050 let mut sieves = vec![None; width];
3051 for sieve in sieves.iter_mut().take(width) {
3052 match cur.u8()? {
3053 0 => continue,
3054 1 => {}
3055 _ => return Err(invalid("a sieve page has an unknown tag")),
3056 }
3057 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3058 let end = page
3059 .offset
3060 .checked_add(u64::from(page.length))
3061 .ok_or_else(|| invalid("sieve page offset overflow"))?;
3062 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3063 return Err(invalid("sieve page range is outside the file"));
3064 }
3065 *sieve = Some(page);
3066 }
3067 let mut ranges = Vec::with_capacity(width);
3068 for _ in 0..width {
3069 let low = cur.bound()?;
3070 let high = cur.bound()?;
3071 let nulls = cur.u32()? as usize;
3072 if nulls > stripe_rows {
3073 return Err(invalid("null count exceeds stripe rows"));
3074 }
3075 let exact = cur.u8()? != 0;
3076 let sum = match cur.u8()? {
3077 0 => None,
3078 1 => Some(i128::from_le_bytes(
3079 cur.take(16)?.try_into().map_err(|_| invalid("a stripe sum is truncated"))?,
3080 )),
3081 _ => return Err(invalid("a stripe sum has an unknown tag")),
3082 };
3083 ranges.push(Range { low, high, nulls, exact, sum });
3084 }
3085 stripes.push(Stripe {
3086 rows: stripe_rows,
3087 parts,
3088 index,
3089 pages,
3090 memberships,
3091 sieves,
3092 zone: Zone::from_ranges(ranges),
3093 });
3094 }
3095 if total != rows {
3096 return Err(invalid("table row count differs from stripes"));
3097 }
3098 let frequencies = if cur.at == bytes.len() {
3099 vec![None; width]
3100 } else {
3101 if cur.take(8)? != FREQUENCIES {
3102 return Err(invalid("directory extension magic differs"));
3103 }
3104 if cur.u16()? as usize != width {
3105 return Err(invalid("frequency column count differs"));
3106 }
3107 let mut frequencies = Vec::with_capacity(width);
3108 for field in &fields {
3109 let summary = match cur.u8()? {
3110 0 => None,
3111 1 => {
3112 let omitted_max = cur.u64()?;
3113 let count = cur.u32()? as usize;
3114 if count > FREQUENCY_ENTRIES {
3115 return Err(invalid("frequency entry count exceeds its bound"));
3116 }
3117 let mut entries = Vec::with_capacity(count);
3118 for _ in 0..count {
3120 let value = match cur.u8()? {
3121 0 => FrequencyValue::Null,
3122 1 => FrequencyValue::Integer(i128::from_le_bytes(
3123 cur.take(16)?.try_into().expect("sixteen bytes"),
3124 )),
3125 2 => FrequencyValue::Code(cur.u32()?),
3126 _ => return Err(invalid("frequency value tag differs")),
3127 };
3128 let valid = matches!(
3129 (&field.ty, value),
3130 (_, FrequencyValue::Null)
3131 | (LogicalType::Varchar, FrequencyValue::Code(_))
3132 | (
3133 LogicalType::TinyInt
3134 | LogicalType::SmallInt
3135 | LogicalType::Integer
3136 | LogicalType::BigInt
3137 | LogicalType::UTinyInt
3138 | LogicalType::USmallInt
3139 | LogicalType::UInteger
3140 | LogicalType::UBigInt
3141 | LogicalType::Date
3142 | LogicalType::Timestamp,
3143 FrequencyValue::Integer(_),
3144 )
3145 );
3146 if !valid {
3147 return Err(invalid("frequency value does not match its column"));
3148 }
3149 let count = cur.u64()?;
3150 if count == 0 || count > rows as u64 {
3151 return Err(invalid("frequency count is outside the table"));
3152 }
3153 entries.push(FrequencyEntry { value, count });
3154 }
3155 if entries.windows(2).any(|pair| pair[0].count < pair[1].count) {
3156 return Err(invalid("frequency entries are not descending"));
3157 }
3158 let ordinals = {
3159 let ordinal_count = cur.u32()? as usize;
3160 if ordinal_count > FREQUENCY_ORDINALS || ordinal_count > rows {
3161 return Err(invalid("frequency ordinal count exceeds its bound"));
3162 }
3163 let mut ordinals = Vec::with_capacity(ordinal_count);
3164 let mut previous = 0_u64;
3165 for at in 0..ordinal_count {
3166 let delta = cur.var_u64()?;
3167 if at != 0 && delta == 0 {
3168 return Err(invalid("frequency ordinals are not increasing"));
3169 }
3170 let ordinal = if at == 0 {
3171 delta
3172 } else {
3173 previous
3174 .checked_add(delta)
3175 .ok_or_else(|| invalid("frequency ordinal overflows"))?
3176 };
3177 if ordinal >= rows as u64 {
3178 return Err(invalid("frequency ordinal is outside the table"));
3179 }
3180 ordinals.push(ordinal);
3181 previous = ordinal;
3182 }
3183 ordinals
3184 };
3185 Some(FrequencySummary { entries, omitted_max, ordinals })
3186 }
3187 _ => return Err(invalid("frequency summary tag differs")),
3188 };
3189 frequencies.push(summary);
3190 }
3191 frequencies
3192 };
3193 if cur.at != bytes.len() {
3194 return Err(invalid("directory has trailing bytes"));
3195 }
3196 Ok(Table { name, fields, stripes, rows, dictionaries, frequencies })
3197}
3198
3199fn put_bound(out: &mut Vec<u8>, bound: Option<&Bound>) -> Result<()> {
3200 match bound {
3201 None => out.push(0),
3202 Some(Bound::Int(value)) => {
3203 out.push(1);
3204 out.extend_from_slice(&value.to_le_bytes());
3205 }
3206 Some(Bound::Real(value)) => {
3207 out.push(2);
3208 out.extend_from_slice(&value.to_le_bytes());
3209 }
3210 Some(Bound::Bytes(value)) => {
3211 out.push(3);
3212 put_u32(out, u32::try_from(value.len()).map_err(|_| invalid("bound length overflow"))?);
3213 out.extend_from_slice(value);
3214 }
3215 }
3216 Ok(())
3217}
3218
3219#[derive(Debug)]
3236struct Codes;
3237
3238impl chooser::Chooser for Codes {
3239 fn name(&self) -> &'static str {
3240 "codes"
3241 }
3242
3243 fn narrow_strings(
3244 &self,
3245 _values: &[&[u8]],
3246 offered: &[string::Kind],
3247 _depth: u8,
3248 ) -> Vec<string::Kind> {
3249 offered.to_vec()
3252 }
3253
3254 fn narrow_integers(
3255 &self,
3256 _values: &[i64],
3257 offered: &[integer::Kind],
3258 depth: u8,
3259 ) -> Vec<integer::Kind> {
3260 let keep: &[integer::Kind] = if depth == 0 {
3261 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Rle]
3262 } else {
3263 &[integer::Kind::Constant, integer::Kind::Packed]
3264 };
3265 let narrowed: Vec<integer::Kind> =
3266 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3267 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3270 }
3271}
3272
3273#[derive(Debug)]
3283struct Fixed;
3284
3285impl chooser::Chooser for Fixed {
3286 fn name(&self) -> &'static str {
3287 "fixed"
3288 }
3289
3290 fn narrow_strings(
3291 &self,
3292 _values: &[&[u8]],
3293 offered: &[string::Kind],
3294 _depth: u8,
3295 ) -> Vec<string::Kind> {
3296 offered.to_vec()
3297 }
3298
3299 fn narrow_integers(
3300 &self,
3301 _values: &[i64],
3302 offered: &[integer::Kind],
3303 depth: u8,
3304 ) -> Vec<integer::Kind> {
3305 let keep: &[integer::Kind] = if depth == 0 {
3306 &[
3307 integer::Kind::Constant,
3308 integer::Kind::Packed,
3309 integer::Kind::Delta,
3310 integer::Kind::Rle,
3311 integer::Kind::Sparse,
3312 ]
3313 } else {
3314 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Delta]
3315 };
3316 let narrowed: Vec<integer::Kind> =
3317 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3318 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3319 }
3320}
3321
3322fn widened(data: &Data) -> Option<Vec<i64>> {
3329 match data {
3330 Data::Int8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3331 Data::UInt8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3332 Data::Int16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3333 Data::UInt16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3334 Data::Int32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3335 Data::UInt32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3336 Data::Int64(values) => Some(values.to_vec()),
3337 _ => None,
3338 }
3339}
3340
3341fn narrowed(ty: &LogicalType, values: Vec<i64>) -> Result<Data> {
3346 fn fit<T: TryFrom<i64>>(values: &[i64]) -> Result<Vec<T>> {
3347 values
3348 .iter()
3349 .map(|value| T::try_from(*value).map_err(|_| invalid("page value is not of its type")))
3350 .collect()
3351 }
3352 Ok(match ty {
3353 LogicalType::TinyInt => Data::Int8(fit::<i8>(&values)?.into()),
3354 LogicalType::UTinyInt => Data::UInt8(fit::<u8>(&values)?.into()),
3355 LogicalType::SmallInt => Data::Int16(fit::<i16>(&values)?.into()),
3356 LogicalType::USmallInt => Data::UInt16(fit::<u16>(&values)?.into()),
3357 LogicalType::Integer | LogicalType::Date => Data::Int32(fit::<i32>(&values)?.into()),
3358 LogicalType::UInteger => Data::UInt32(fit::<u32>(&values)?.into()),
3359 LogicalType::BigInt | LogicalType::Timestamp => Data::Int64(values.into()),
3360 _ => return Err(invalid("cascade codec belongs to a page that is not integers")),
3361 })
3362}
3363
3364fn plain_width(ty: &LogicalType) -> Option<usize> {
3367 Some(match ty {
3368 LogicalType::TinyInt | LogicalType::UTinyInt => 1,
3369 LogicalType::SmallInt | LogicalType::USmallInt => 2,
3370 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date => 4,
3371 LogicalType::BigInt | LogicalType::Timestamp => 8,
3372 _ => return None,
3373 })
3374}
3375
3376fn cascaded(
3382 flat: &Vector,
3383 ty: &LogicalType,
3384 packed: Option<&Packed<'_>>,
3385) -> Result<Option<Vec<u8>>> {
3386 let (Some(width), Some(data)) = (plain_width(ty), flat.data()) else { return Ok(None) };
3387 let Some(values) = widened(data) else { return Ok(None) };
3388 let plain = values.len().saturating_mul(width);
3389 let best = match packed {
3390 Some(packed) => plain.min(21 + size_of_val(packed.words())),
3392 None => plain,
3393 };
3394 let out = integer::encode_with(&values, &Fixed)?;
3395 Ok((out.len() < best).then_some(out))
3396}
3397
3398fn encoded_codes(codes: &[u32]) -> Result<Option<Vec<u8>>> {
3410 let wide: Vec<i64> = codes.iter().map(|code| i64::from(*code)).collect();
3411 let coded = integer::encode_with(&wide, &Codes)?;
3412 let plain = codes.len().saturating_mul(size_of::<u32>());
3413 Ok((coded.len() < plain).then_some(coded))
3414}
3415
3416fn encode(
3417 vector: &Vector,
3418 global: Option<&mut GlobalDictionary>,
3419) -> Result<(Vec<u8>, Option<Vec<u32>>)> {
3420 let ty = vector.logical_type();
3421 let flat = vector.flatten()?;
3423 let mut out = Vec::new();
3424 let mut global_codes = None;
3425 if let Some(global) = global {
3426 let mut codes = Vec::with_capacity(flat.len());
3427 for row in 0..flat.len() {
3428 let text = flat.text_at(row).unwrap_or("");
3429 let code = global.code(text)?;
3430 global.observe(code, flat.is_null_at(row))?;
3431 codes.push(code);
3432 }
3433 global_codes = Some(codes);
3434 }
3435 let membership = global_codes.as_deref().map(unique_codes);
3436 let dictionary = if global_codes.is_none() && ty == &LogicalType::Varchar {
3437 string_dictionary(&flat)?
3438 } else {
3439 None
3440 };
3441 let packed_vector = if dictionary.is_none() && global_codes.is_none() {
3442 Some(flat.bit_packed()?)
3443 } else {
3444 None
3445 };
3446 let packed = packed_vector.as_ref().and_then(Vector::packed_parts);
3447 let coded = match global_codes.as_deref() {
3448 Some(codes) => encoded_codes(codes)?,
3449 None => None,
3450 };
3451 let cascade = if dictionary.is_none() && global_codes.is_none() {
3455 cascaded(&flat, ty, packed.as_ref())?
3456 } else {
3457 None
3458 };
3459 out.push(if coded.is_some() {
3460 4
3461 } else if cascade.is_some() {
3462 5
3463 } else if global_codes.is_some() {
3464 3
3465 } else if dictionary.is_some() {
3466 1
3467 } else if packed.is_some() {
3468 2
3469 } else {
3470 0
3471 });
3472 let nulls = flat.validity();
3473 let flag = match nulls {
3474 Validity::AllValid => 0,
3475 Validity::AllInvalid => 1,
3476 Validity::Mask(_) => 2,
3477 };
3478 out.push(flag);
3479 if flag == 2 {
3480 for group in (0..vector.len()).step_by(8) {
3481 let mut bits = 0_u8;
3482 for bit in 0..8 {
3483 if group + bit < vector.len() && !flat.is_null_at(group + bit) {
3484 bits |= 1 << bit;
3485 }
3486 }
3487 out.push(bits);
3488 }
3489 }
3490 if let Some(coded) = coded {
3491 out.extend_from_slice(&coded);
3492 return Ok((out, membership));
3493 }
3494 if let Some(cascade) = cascade {
3495 out.extend_from_slice(&cascade);
3496 return Ok((out, membership));
3497 }
3498 if let Some(codes) = global_codes {
3499 for code in codes {
3500 put_u32(&mut out, code);
3501 }
3502 return Ok((out, membership));
3503 }
3504 if let Some(dictionary) = dictionary {
3505 out.extend_from_slice(&dictionary);
3506 return Ok((out, membership));
3507 }
3508 if let Some(packed) = packed {
3509 if packed.offset() != 0 {
3510 return Err(invalid("writer received a sliced packed vector"));
3511 }
3512 out.push(u8::try_from(packed.width()).map_err(|_| invalid("packed width overflow"))?);
3513 out.extend_from_slice(&packed.base().to_le_bytes());
3514 put_u32(
3515 &mut out,
3516 u32::try_from(packed.words().len()).map_err(|_| invalid("too many packed words"))?,
3517 );
3518 for word in packed.words() {
3519 put_u64(&mut out, *word);
3520 }
3521 return Ok((out, membership));
3522 }
3523 let data = flat.data().ok_or_else(|| invalid("scalar column did not flatten"))?;
3524 match (ty, data) {
3525 (LogicalType::TinyInt, Data::Int8(values)) => {
3526 for value in &**values {
3527 out.extend_from_slice(&value.to_le_bytes());
3528 }
3529 }
3530 (LogicalType::UTinyInt, Data::UInt8(values)) => {
3531 for value in &**values {
3532 out.extend_from_slice(&value.to_le_bytes());
3533 }
3534 }
3535 (LogicalType::SmallInt, Data::Int16(values)) => {
3536 for value in &**values {
3537 out.extend_from_slice(&value.to_le_bytes());
3538 }
3539 }
3540 (LogicalType::USmallInt, Data::UInt16(values)) => {
3541 for value in &**values {
3542 out.extend_from_slice(&value.to_le_bytes());
3543 }
3544 }
3545 (LogicalType::UInteger, Data::UInt32(values)) => {
3546 for value in &**values {
3547 out.extend_from_slice(&value.to_le_bytes());
3548 }
3549 }
3550 (LogicalType::UBigInt, Data::UInt64(values)) => {
3551 for value in &**values {
3552 out.extend_from_slice(&value.to_le_bytes());
3553 }
3554 }
3555 (LogicalType::Integer | LogicalType::Date, Data::Int32(values)) => {
3556 for value in &**values {
3557 out.extend_from_slice(&value.to_le_bytes());
3558 }
3559 }
3560 (LogicalType::BigInt | LogicalType::Timestamp, Data::Int64(values)) => {
3561 for value in &**values {
3562 out.extend_from_slice(&value.to_le_bytes());
3563 }
3564 }
3565 (LogicalType::Boolean, Data::Bool(values)) => {
3566 for value in &**values {
3567 out.push(u8::from(*value));
3568 }
3569 }
3570 (LogicalType::Varchar, Data::Varlen(values)) => {
3571 let mut bytes = Vec::new();
3572 put_u32(&mut out, 0);
3573 for row in 0..vector.len() {
3574 let value = values.bytes(row).ok_or_else(|| invalid("string view is invalid"))?;
3575 bytes.extend_from_slice(value);
3576 put_u32(
3577 &mut out,
3578 u32::try_from(bytes.len())
3579 .map_err(|_| invalid("string payload exceeds 4GiB"))?,
3580 );
3581 }
3582 out.extend_from_slice(&bytes);
3583 }
3584 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
3585 }
3586 Ok((out, membership))
3587}
3588
3589fn put_varint(out: &mut Vec<u8>, mut value: u32) {
3590 while value >= 0x80 {
3591 out.push((value as u8 & 0x7f) | 0x80);
3592 value >>= 7;
3593 }
3594 out.push(value as u8);
3595}
3596
3597fn unique_codes(codes: &[u32]) -> Vec<u32> {
3599 let mut unique = codes.to_vec();
3600 unique.sort_unstable();
3601 unique.dedup();
3602 unique
3603}
3604
3605fn merged_codes(lists: Vec<Vec<u32>>) -> Vec<u32> {
3611 let mut lists = lists;
3612 while lists.len() > 1 {
3613 let mut next = Vec::with_capacity(lists.len().div_ceil(2));
3614 for pair in lists.chunks(2) {
3615 match pair {
3616 [left, right] => next.push(merged_pair(left, right)),
3617 [only] => next.push(only.clone()),
3618 _ => {}
3619 }
3620 }
3621 lists = next;
3622 }
3623 lists.pop().unwrap_or_default()
3624}
3625
3626fn merged_pair(left: &[u32], right: &[u32]) -> Vec<u32> {
3627 let mut out = Vec::with_capacity(left.len().saturating_add(right.len()));
3628 let mut at = 0;
3629 let mut to = 0;
3630 while at < left.len() && to < right.len() {
3631 match left[at].cmp(&right[to]) {
3632 Ordering::Less => {
3633 out.push(left[at]);
3634 at += 1;
3635 }
3636 Ordering::Greater => {
3637 out.push(right[to]);
3638 to += 1;
3639 }
3640 Ordering::Equal => {
3641 out.push(left[at]);
3642 at += 1;
3643 to += 1;
3644 }
3645 }
3646 }
3647 out.extend_from_slice(&left[at..]);
3648 out.extend_from_slice(&right[to..]);
3649 out
3650}
3651
3652fn merged_range(ranges: impl Iterator<Item = Range>) -> Range {
3657 let mut merged = Range::default();
3658 let mut first = true;
3659 for range in ranges {
3660 merged.nulls = merged.nulls.saturating_add(range.nulls);
3661 merged.sum = match (merged.sum.take(), range.sum) {
3665 (Some(held), Some(next)) if !first => held.checked_add(next),
3666 (_, next) if first => next,
3667 _ => None,
3668 };
3669 merged.exact = if first { range.exact } else { merged.exact && range.exact };
3670 if first {
3671 merged.low = range.low;
3672 merged.high = range.high;
3673 first = false;
3674 continue;
3675 }
3676 merged.low = match (merged.low.take(), range.low) {
3677 (Some(held), Some(next)) => Some(held.smaller(next)),
3678 _ => None,
3679 };
3680 merged.high = match (merged.high.take(), range.high) {
3681 (Some(held), Some(next)) => Some(held.larger(next)),
3682 _ => None,
3683 };
3684 }
3685 merged
3686}
3687
3688fn encode_sieves<'a>(sieves: impl Iterator<Item = &'a Option<Sieve>>) -> Result<Vec<u8>> {
3694 let held: Vec<&Option<Sieve>> = sieves.collect();
3695 let mut out = Vec::new();
3696 put_u32(
3697 &mut out,
3698 u32::try_from(held.len()).map_err(|_| invalid("too many parts in a stripe"))?,
3699 );
3700 for sieve in &held {
3701 let length = sieve.as_ref().map_or(0, Sieve::len);
3702 put_u32(&mut out, u32::try_from(length).map_err(|_| invalid("sieve length overflow"))?);
3703 }
3704 for sieve in held.into_iter().flatten() {
3706 out.extend_from_slice(&sieve.to_bytes());
3707 }
3708 Ok(out)
3709}
3710
3711fn decode_sieves(bytes: &[u8]) -> Result<Vec<Option<Sieve>>> {
3717 let parts = u32::from_le_bytes(
3718 bytes
3719 .get(..4)
3720 .ok_or_else(|| invalid("sieve page is truncated"))?
3721 .try_into()
3722 .map_err(|_| invalid("sieve page is truncated"))?,
3723 ) as usize;
3724 let mut lengths = Vec::with_capacity(parts);
3725 for part in 0..parts {
3726 let at = 4 + part * 4;
3727 let field = bytes.get(at..at + 4).ok_or_else(|| invalid("sieve page is truncated"))?;
3728 lengths.push(u32::from_le_bytes(
3729 field.try_into().map_err(|_| invalid("sieve page is truncated"))?,
3730 ) as usize);
3731 }
3732 let mut at = 4 + parts * 4;
3733 let mut out = Vec::with_capacity(parts);
3734 for length in lengths {
3735 if length == 0 {
3736 out.push(None);
3737 continue;
3738 }
3739 let end = at.checked_add(length).ok_or_else(|| invalid("sieve page is truncated"))?;
3740 let field = bytes.get(at..end).ok_or_else(|| invalid("sieve page is truncated"))?;
3741 out.push(Sieve::from_bytes(field));
3742 at = end;
3743 }
3744 if at != bytes.len() {
3745 return Err(invalid("sieve page has trailing bytes"));
3746 }
3747 Ok(out)
3748}
3749
3750fn encode_membership(unique: &[u32]) -> Vec<u8> {
3756 let mut out = Vec::with_capacity(unique.len().saturating_mul(2).saturating_add(5));
3757 put_varint(&mut out, u32::try_from(unique.len()).unwrap_or(u32::MAX));
3758 let mut previous = 0;
3759 for (at, &code) in unique.iter().enumerate() {
3760 put_varint(&mut out, if at == 0 { code } else { code - previous });
3761 previous = code;
3762 }
3763 out
3764}
3765
3766fn take_varint(bytes: &[u8], at: &mut usize) -> Result<u32> {
3767 let mut value = 0_u32;
3768 for shift in (0..35).step_by(7) {
3769 let byte = *bytes.get(*at).ok_or_else(|| invalid("membership varint is truncated"))?;
3770 *at += 1;
3771 let part = u32::from(byte & 0x7f);
3772 if shift == 28 && part > 0x0f {
3773 return Err(invalid("membership varint overflow"));
3774 }
3775 value = value
3776 .checked_add(
3777 part.checked_shl(shift).ok_or_else(|| invalid("membership varint overflow"))?,
3778 )
3779 .ok_or_else(|| invalid("membership varint overflow"))?;
3780 if byte & 0x80 == 0 {
3781 return Ok(value);
3782 }
3783 }
3784 Err(invalid("membership varint is too long"))
3785}
3786
3787fn decode_membership(bytes: &[u8]) -> Result<Vec<u32>> {
3788 let mut at = 0;
3789 let count = take_varint(bytes, &mut at)? as usize;
3790 let mut codes = Vec::with_capacity(count);
3791 let mut previous = 0_u32;
3792 for index in 0..count {
3793 let delta = take_varint(bytes, &mut at)?;
3794 let code = if index == 0 {
3795 delta
3796 } else {
3797 previous.checked_add(delta).ok_or_else(|| invalid("membership code overflow"))?
3798 };
3799 if index > 0 && code <= previous {
3800 return Err(invalid("membership codes are not increasing"));
3801 }
3802 codes.push(code);
3803 previous = code;
3804 }
3805 if at != bytes.len() {
3806 return Err(invalid("membership page has trailing bytes"));
3807 }
3808 Ok(codes)
3809}
3810
3811fn string_dictionary(vector: &Vector) -> Result<Option<Vec<u8>>> {
3812 let mut by_text = HashMap::new();
3813 let mut values = Vec::new();
3814 let mut codes = Vec::with_capacity(vector.len());
3815 let mut plain_bytes = 0_usize;
3816 for row in 0..vector.len() {
3817 let text = vector.text_at(row).unwrap_or("");
3818 plain_bytes = plain_bytes.saturating_add(text.len());
3819 let code = match by_text.get(text) {
3820 Some(&code) => code,
3821 None => {
3822 let code = u32::try_from(values.len())
3823 .map_err(|_| invalid("too many dictionary values"))?;
3824 by_text.insert(text, code);
3825 values.push(text);
3826 code
3827 }
3828 };
3829 codes.push(code);
3830 }
3831 let dictionary_bytes = values.iter().map(|value| value.len()).sum::<usize>();
3832 let encoded = 8_usize
3833 .saturating_add((values.len() + 1).saturating_mul(4))
3834 .saturating_add(dictionary_bytes)
3835 .saturating_add(codes.len().saturating_mul(4));
3836 let plain = (vector.len() + 1).saturating_mul(4).saturating_add(plain_bytes);
3837 if encoded >= plain {
3838 return Ok(None);
3839 }
3840 let mut out = Vec::with_capacity(encoded);
3841 put_u32(
3842 &mut out,
3843 u32::try_from(values.len()).map_err(|_| invalid("too many dictionary values"))?,
3844 );
3845 put_u32(
3846 &mut out,
3847 u32::try_from(dictionary_bytes).map_err(|_| invalid("dictionary payload exceeds 4GiB"))?,
3848 );
3849 let mut offset = 0_u32;
3850 put_u32(&mut out, offset);
3851 for value in &values {
3852 offset = offset
3853 .checked_add(
3854 u32::try_from(value.len()).map_err(|_| invalid("dictionary value is too long"))?,
3855 )
3856 .ok_or_else(|| invalid("dictionary payload exceeds 4GiB"))?;
3857 put_u32(&mut out, offset);
3858 }
3859 for value in values {
3860 out.extend_from_slice(value.as_bytes());
3861 }
3862 for code in codes {
3863 put_u32(&mut out, code);
3864 }
3865 Ok(Some(out))
3866}
3867
3868struct EncodedDictionary {
3869 index: Vec<u8>,
3870 ranks: Vec<u8>,
3871 payload: Vec<Vec<u8>>,
3874}
3875
3876fn head(bytes: &[u8]) -> u64 {
3878 let mut word = [0; 8];
3879 let take = bytes.len().min(8);
3880 word[..take].copy_from_slice(&bytes[..take]);
3881 u64::from_be_bytes(word)
3882}
3883
3884fn rankings(dictionaries: &[Option<GlobalDictionary>]) -> Result<Vec<Vec<(u64, u32)>>> {
3892 let present =
3893 dictionaries.iter().enumerate().filter(|(_, held)| held.is_some()).map(|(at, _)| at);
3894 let present = present.collect::<Vec<_>>();
3895 let mut orders = vec![Vec::new(); dictionaries.len()];
3896 let workers = std::thread::available_parallelism()
3897 .map_or(1, usize::from)
3898 .min(MAX_FREQUENCY_WORKERS)
3899 .min(present.len());
3900 if workers <= 1 {
3901 for at in present {
3902 if let Some(dictionary) = &dictionaries[at] {
3903 orders[at] = dictionary.ranked();
3904 }
3905 }
3906 return Ok(orders);
3907 }
3908 let width = present.len().div_ceil(workers);
3909 let pieces = std::thread::scope(|scope| {
3910 present
3911 .chunks(width)
3912 .map(|columns| {
3913 scope.spawn(|| {
3914 columns
3915 .iter()
3916 .filter_map(|&at| dictionaries[at].as_ref().map(|held| (at, held.ranked())))
3917 .collect::<Vec<_>>()
3918 })
3919 })
3920 .collect::<Vec<_>>()
3921 .into_iter()
3922 .map(|handle| {
3923 handle.join().map_err(|_| Error::internal("a dictionary sort worker panicked"))
3924 })
3925 .collect::<Result<Vec<_>>>()
3926 })?;
3927 for piece in pieces {
3928 for (at, order) in piece {
3929 orders[at] = order;
3930 }
3931 }
3932 Ok(orders)
3933}
3934
3935fn encode_global_dictionary(
3936 dictionary: GlobalDictionary,
3937 order: &[(u64, u32)],
3938) -> Result<EncodedDictionary> {
3939 let values = dictionary.offsets.len() - 1;
3940 if order.len() != values {
3941 return Err(invalid("global dictionary order does not cover its values"));
3942 }
3943 let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
3944 let payload = encode_payload(&dictionary)?;
3945 if payload.len() != blocks {
3946 return Err(invalid("global dictionary payload is not the blocks it says it is"));
3947 }
3948 let ranks = encode_ranks(order);
3949 let rank_blocks = values.div_ceil(TEXT_RANK_BLOCK);
3950 let mut index = Vec::with_capacity(12 + (values + 1) * 4 + (blocks * 2 + rank_blocks) * 8);
3951 put_u32(
3952 &mut index,
3953 u32::try_from(values).map_err(|_| invalid("global dictionary has too many values"))?,
3954 );
3955 put_u32(&mut index, TEXT_PAYLOAD_VALUES as u32);
3956 put_u32(
3957 &mut index,
3958 u32::try_from(blocks).map_err(|_| invalid("global dictionary has too many blocks"))?,
3959 );
3960 for offset in dictionary.offsets {
3961 put_u32(&mut index, offset);
3962 }
3963 let mut at = 0_u64;
3967 for block in &payload {
3968 at = at
3969 .checked_add(block.len() as u64)
3970 .ok_or_else(|| invalid("global dictionary payload overflow"))?;
3971 put_u64(&mut index, at);
3972 }
3973 for block in &payload {
3974 put_u64(&mut index, checksum(block));
3975 }
3976 for block in ranks.chunks(TEXT_RANK_BLOCK * RANK_ENTRY) {
3977 put_u64(&mut index, checksum(block));
3978 }
3979 Ok(EncodedDictionary { index, ranks, payload })
3980}
3981
3982const PAYLOAD_SAMPLE_BLOCKS: usize = 8;
3989
3990fn payload_shapes() -> Vec<chooser::Settled> {
4016 let integers = vec![integer::Kind::Packed];
4017 [
4018 vec![string::Kind::Front, string::Kind::Lz],
4019 vec![string::Kind::Lz, string::Kind::Fsst],
4020 vec![string::Kind::Lz, string::Kind::Plain],
4021 vec![string::Kind::Fsst],
4022 vec![string::Kind::Plain],
4023 ]
4024 .into_iter()
4025 .map(|strings| chooser::Settled::new(strings, integers.clone()))
4026 .collect()
4027}
4028
4029fn encode_payload(dictionary: &GlobalDictionary) -> Result<Vec<Vec<u8>>> {
4035 let values = dictionary.offsets.len() - 1;
4036 let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
4037 let run = |block: usize| {
4038 let first = block * TEXT_PAYLOAD_VALUES;
4039 let last = (first + TEXT_PAYLOAD_VALUES).min(values);
4040 (first..last)
4041 .map(|value| {
4042 let from = dictionary.offsets[value] as usize;
4043 let to = dictionary.offsets[value + 1] as usize;
4044 &dictionary.payload[from..to]
4045 })
4046 .collect::<Vec<_>>()
4047 };
4048 let shape = (blocks > PAYLOAD_SAMPLE_BLOCKS).then(|| settle_shape(&run, blocks)).transpose()?;
4051 let one = |block: usize| match &shape {
4052 Some(shape) => string::encode_with(&run(block), shape),
4053 None => string::encode(&run(block)),
4054 };
4055 let workers = std::thread::available_parallelism()
4056 .map_or(1, usize::from)
4057 .min(MAX_FREQUENCY_WORKERS)
4058 .min(blocks);
4059 if workers <= 1 {
4060 return (0..blocks).map(one).collect();
4061 }
4062 let next = AtomicUsize::new(0);
4063 let pieces = std::thread::scope(|scope| {
4064 (0..workers)
4065 .map(|_| {
4066 scope.spawn(|| {
4067 let mut mine = Vec::new();
4068 loop {
4069 let block = next.fetch_add(1, Atomic::Relaxed);
4070 if block >= blocks {
4071 break;
4072 }
4073 mine.push((block, one(block)?));
4074 }
4075 Ok(mine)
4076 })
4077 })
4078 .collect::<Vec<_>>()
4079 .into_iter()
4080 .map(|handle| {
4081 handle.join().map_err(|_| Error::internal("a dictionary encode worker panicked"))?
4082 })
4083 .collect::<Result<Vec<_>>>()
4084 })?;
4085 let mut payload = vec![Vec::new(); blocks];
4086 for piece in pieces {
4087 for (block, bytes) in piece {
4088 payload[block] = bytes;
4089 }
4090 }
4091 Ok(payload)
4092}
4093
4094fn settle_shape<'a>(
4102 run: &dyn Fn(usize) -> Vec<&'a [u8]>,
4103 blocks: usize,
4104) -> Result<chooser::Settled> {
4105 let last = blocks - 1;
4106 let sample = (0..PAYLOAD_SAMPLE_BLOCKS)
4107 .map(|region| run(region * last / (PAYLOAD_SAMPLE_BLOCKS - 1)))
4108 .collect::<Vec<_>>();
4109 let mut best: Option<(chooser::Settled, usize)> = None;
4110 for shape in payload_shapes() {
4111 let mut size = 0;
4112 for block in &sample {
4113 size += string::encode_with(block, &shape)?.len();
4114 }
4115 if best.as_ref().is_none_or(|(_, smallest)| size < *smallest) {
4116 best = Some((shape, size));
4117 }
4118 }
4119 best.map(|(shape, _)| shape)
4120 .ok_or_else(|| invalid("no shape applies to a global dictionary payload"))
4121}
4122
4123fn encode_ranks(order: &[(u64, u32)]) -> Vec<u8> {
4130 let mut out = Vec::with_capacity(order.len() * RANK_ENTRY);
4131 for block in order.chunks(TEXT_RANK_BLOCK) {
4132 for &(head, _) in block {
4133 put_u64(&mut out, head);
4134 }
4135 for &(_, code) in block {
4136 put_u32(&mut out, code);
4137 }
4138 }
4139 out
4140}
4141
4142fn open_global_dictionary(file: Arc<File>, page: Page, ty: &LogicalType) -> Result<Vector> {
4143 if ty != &LogicalType::Varchar {
4144 return Err(invalid("global dictionary belongs to a non-string column"));
4145 }
4146 let mut header = [0; 12];
4147 read_at(&file, page.offset, &mut header)?;
4148 let count = u32::from_le_bytes(header[0..4].try_into().expect("four bytes")) as usize;
4149 let per_block = u32::from_le_bytes(header[4..8].try_into().expect("four bytes")) as usize;
4150 let blocks = u32::from_le_bytes(header[8..12].try_into().expect("four bytes")) as usize;
4151 if per_block != TEXT_PAYLOAD_VALUES {
4152 return Err(invalid("global dictionary block width differs"));
4153 }
4154 if blocks != count.div_ceil(TEXT_PAYLOAD_VALUES) {
4155 return Err(invalid("global dictionary block count differs from its value count"));
4156 }
4157 let offset_len = (count + 1)
4158 .checked_mul(4)
4159 .ok_or_else(|| invalid("global dictionary offset count overflow"))?;
4160 let ranks = count;
4165 let rank_blocks = ranks.div_ceil(TEXT_RANK_BLOCK);
4166 let rank_len =
4167 ranks.checked_mul(RANK_ENTRY).ok_or_else(|| invalid("global dictionary rank overflow"))?;
4168 let hash_len = blocks
4171 .checked_mul(2)
4172 .and_then(|words| words.checked_add(rank_blocks))
4173 .and_then(|words| words.checked_mul(8))
4174 .ok_or_else(|| invalid("global dictionary block count overflow"))?;
4175 let index_len = 12usize
4176 .checked_add(offset_len)
4177 .and_then(|len| len.checked_add(hash_len))
4178 .ok_or_else(|| invalid("global dictionary header overflow"))?;
4179 let body_len = index_len
4180 .checked_add(rank_len)
4181 .ok_or_else(|| invalid("global dictionary header overflow"))?;
4182 if body_len > page.length as usize {
4183 return Err(invalid("global dictionary offset index exceeds its page"));
4184 }
4185 let mut index = vec![0; index_len];
4186 index[..12].copy_from_slice(&header);
4187 read_at(&file, page.offset + 12, &mut index[12..])?;
4188 if checksum(&index) != page.hash {
4189 return Err(invalid("global dictionary index checksum differs"));
4190 }
4191 let offsets = index[12..12 + offset_len]
4192 .chunks_exact(4)
4193 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4194 .collect::<Vec<_>>();
4195 let mut words = index[12 + offset_len..]
4196 .chunks_exact(8)
4197 .map(|part| u64::from_le_bytes(part.try_into().expect("eight bytes")))
4198 .collect::<Vec<_>>();
4199 let mut hashes = words.split_off(blocks);
4200 let rank_hashes = hashes.split_off(blocks);
4201 let ends = words;
4202 let stored_len = page.length as usize - body_len;
4205 if ends.last().copied().unwrap_or_default() as usize != stored_len
4206 || ends.windows(2).any(|pair| pair[0] > pair[1])
4207 {
4208 return Err(invalid("global dictionary blocks do not bound the payload"));
4209 }
4210 if offsets.first() != Some(&0) || offsets.windows(2).any(|pair| pair[0] > pair[1]) {
4211 return Err(invalid("global dictionary offsets do not bound the payload"));
4212 }
4213 Vector::external_text(
4214 LogicalType::Varchar,
4215 Arc::new(NativeText {
4216 file,
4217 offsets,
4218 ranks,
4219 rank_at: page.offset + index_len as u64,
4220 rank_hashes,
4221 rank_blocks: (0..rank_blocks).map(|_| OnceLock::new()).collect(),
4222 code_ranks: OnceLock::new(),
4223 payload: page.offset + body_len as u64,
4224 ends,
4225 hashes,
4226 blocks: (0..blocks).map(|_| OnceLock::new()).collect(),
4227 }),
4228 )
4229}
4230
4231fn decode(
4232 ty: &LogicalType,
4233 rows: usize,
4234 bytes: &[u8],
4235 global: Option<Arc<Vector>>,
4236) -> Result<Vector> {
4237 let mut cur = Cursor { bytes, at: 0 };
4238 let codec = cur.u8()?;
4239 let flag = cur.u8()?;
4240 let validity = match flag {
4241 0 => Validity::AllValid,
4242 1 => Validity::AllInvalid,
4243 2 => {
4244 let mask = cur.take(rows.div_ceil(8))?;
4245 Validity::from_iter(rows, |row| mask[row / 8] >> (row % 8) & 1 == 1)
4246 }
4247 _ => return Err(invalid("page validity tag differs")),
4248 };
4249 if codec == 1 {
4250 if ty != &LogicalType::Varchar {
4251 return Err(invalid("dictionary codec belongs to a non-string page"));
4252 }
4253 let count = cur.u32()? as usize;
4254 let payload_len = cur.u32()? as usize;
4255 let offset_bytes = cur.take(
4256 (count + 1)
4257 .checked_mul(4)
4258 .ok_or_else(|| invalid("dictionary offset count overflow"))?,
4259 )?;
4260 let offsets = offset_bytes
4261 .chunks_exact(4)
4262 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4263 .collect::<Vec<_>>();
4264 let payload = cur.take(payload_len)?.to_vec();
4265 if offsets.first() != Some(&0)
4266 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4267 || offsets.windows(2).any(|pair| pair[0] > pair[1])
4268 {
4269 return Err(invalid("dictionary offsets do not bound the payload"));
4270 }
4271 let mut strings = StringColumn::over(Buffer::from_vec(payload));
4272 for pair in offsets.windows(2) {
4273 strings.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4274 }
4275 let mut codes = Vec::with_capacity(rows);
4276 for _ in 0..rows {
4277 codes.push(cur.u32()?);
4278 }
4279 if codes.iter().any(|code| *code as usize >= count) {
4280 return Err(invalid("dictionary code is out of range"));
4281 }
4282 if cur.at != bytes.len() {
4283 return Err(invalid("dictionary page has trailing bytes"));
4284 }
4285 let dictionary = Vector::flat(LogicalType::Varchar, Data::Varlen(strings))?;
4286 return Ok(Vector::dictionary(codes, dictionary)?.with_validity(validity));
4287 }
4288 if codec == 3 || codec == 4 {
4289 let dictionary = global.ok_or_else(|| invalid("global code page has no dictionary"))?;
4290 let codes = if codec == 4 {
4291 let wide = integer::decode(&bytes[cur.at..])?;
4294 if wide.len() != rows {
4295 return Err(invalid("encoded code page holds the wrong number of rows"));
4296 }
4297 wide.into_iter()
4298 .map(|code| u32::try_from(code).map_err(|_| invalid("code is not a code")))
4299 .collect::<Result<Vec<u32>>>()?
4300 } else {
4301 let mut codes = Vec::with_capacity(rows);
4302 for _ in 0..rows {
4303 codes.push(cur.u32()?);
4304 }
4305 if cur.at != bytes.len() {
4306 return Err(invalid("global code page has trailing bytes"));
4307 }
4308 codes
4309 };
4310 let highest = codes.iter().copied().max();
4311 return Ok(Vector::stable_dictionary_validated(codes, dictionary, highest)?
4312 .with_validity(validity));
4313 }
4314 if codec == 5 {
4315 let values = integer::decode(&bytes[cur.at..])?;
4317 if values.len() != rows {
4318 return Err(invalid("cascade page holds the wrong number of rows"));
4319 }
4320 let data = narrowed(ty, values)?;
4321 return Ok(Vector::flat(ty.clone(), data)?.with_validity(validity));
4322 }
4323 if codec == 2 {
4324 let width = u32::from(cur.u8()?);
4325 let base = i128::from_le_bytes(cur.take(16)?.try_into().expect("sixteen bytes"));
4326 let count = cur.u32()? as usize;
4327 let mut words = Vec::with_capacity(count);
4328 for _ in 0..count {
4329 words.push(cur.u64()?);
4330 }
4331 if cur.at != bytes.len() {
4332 return Err(invalid("packed page has trailing bytes"));
4333 }
4334 return Ok(Vector::packed(ty.clone(), words, width, base, rows)?.with_validity(validity));
4335 }
4336 if codec != 0 {
4337 return Err(invalid("page codec is unknown"));
4338 }
4339 let data = match ty {
4340 LogicalType::TinyInt => {
4341 let values = cur.take(rows)?;
4342 Data::Int8(values.iter().map(|item| *item as i8).collect::<Vec<_>>().into())
4343 }
4344 LogicalType::UTinyInt => Data::UInt8(cur.take(rows)?.to_vec().into()),
4345 LogicalType::SmallInt => {
4346 let values =
4347 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4348 Data::Int16(
4349 values
4350 .chunks_exact(2)
4351 .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
4352 .collect::<Vec<_>>()
4353 .into(),
4354 )
4355 }
4356 LogicalType::USmallInt => {
4357 let values =
4358 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4359 Data::UInt16(
4360 values
4361 .chunks_exact(2)
4362 .map(|item| u16::from_le_bytes(item.try_into().expect("two bytes")))
4363 .collect::<Vec<_>>()
4364 .into(),
4365 )
4366 }
4367 LogicalType::UInteger => {
4368 let values =
4369 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4370 Data::UInt32(
4371 values
4372 .chunks_exact(4)
4373 .map(|item| u32::from_le_bytes(item.try_into().expect("four bytes")))
4374 .collect::<Vec<_>>()
4375 .into(),
4376 )
4377 }
4378 LogicalType::UBigInt => {
4379 let values =
4380 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4381 Data::UInt64(
4382 values
4383 .chunks_exact(8)
4384 .map(|item| u64::from_le_bytes(item.try_into().expect("eight bytes")))
4385 .collect::<Vec<_>>()
4386 .into(),
4387 )
4388 }
4389 LogicalType::Integer | LogicalType::Date => {
4390 let values =
4391 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4392 Data::Int32(
4393 values
4394 .chunks_exact(4)
4395 .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
4396 .collect::<Vec<_>>()
4397 .into(),
4398 )
4399 }
4400 LogicalType::BigInt | LogicalType::Timestamp => {
4401 let values =
4402 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4403 Data::Int64(
4404 values
4405 .chunks_exact(8)
4406 .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
4407 .collect::<Vec<_>>()
4408 .into(),
4409 )
4410 }
4411 LogicalType::Boolean => {
4412 let values = cur.take(rows)?;
4413 if values.iter().any(|value| *value > 1) {
4414 return Err(invalid("boolean page has another value"));
4415 }
4416 Data::Bool(values.iter().map(|value| *value == 1).collect::<Vec<_>>().into())
4417 }
4418 LogicalType::Varchar => {
4419 let offset_bytes = cur
4420 .take((rows + 1).checked_mul(4).ok_or_else(|| invalid("offset count overflow"))?)?;
4421 let offsets = offset_bytes
4422 .chunks_exact(4)
4423 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4424 .collect::<Vec<_>>();
4425 let payload = cur.take(bytes.len() - cur.at)?.to_vec();
4426 if offsets.first() != Some(&0)
4427 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4428 || offsets.windows(2).any(|pair| pair[0] > pair[1])
4429 {
4430 return Err(invalid("string offsets do not bound the payload"));
4431 }
4432 let mut values = StringColumn::over(Buffer::from_vec(payload));
4433 for pair in offsets.windows(2) {
4434 values.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4435 }
4436 Data::Varlen(values)
4437 }
4438 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
4439 };
4440 if cur.at != bytes.len() {
4441 return Err(invalid("page has trailing bytes"));
4442 }
4443 Ok(Vector::flat(ty.clone(), data)?.with_validity(validity))
4444}
4445
4446#[cfg(test)]
4447mod tests {
4448 use std::fs;
4449 use std::io::{Seek, SeekFrom, Write};
4450 use std::path::PathBuf;
4451 use std::time::{SystemTime, UNIX_EPOCH};
4452
4453 use rudb_common::Value;
4454 use rudb_common::bounds::Op;
4455
4456 use super::*;
4457
4458 #[test]
4459 fn checksum_matches_fixed_vectors() {
4460 assert_eq!(checksum(b""), 0xef46_db37_51d8_e999);
4461 assert_eq!(checksum(b"a"), 0xd24e_c4f1_a98c_6e5b);
4462 assert_eq!(checksum(b"abc"), 0x44bc_2cf5_ad77_0999);
4463 }
4464
4465 fn path(label: &str) -> PathBuf {
4466 let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
4467 std::env::temp_dir().join(format!("rudb-native-{label}-{}-{stamp}.rdb", std::process::id()))
4468 }
4469
4470 #[test]
4472 fn a_read_at_an_offset_ignores_where_another_thread_left_the_cursor() {
4473 const SPANS: usize = 64;
4474 const SPAN: usize = 512;
4475 let path = path("positional");
4476 let content: Vec<u8> =
4477 (0..SPANS).flat_map(|span| std::iter::repeat_n(span as u8, SPAN)).collect();
4478 fs::write(&path, &content).expect("the file is written");
4479 let file = Arc::new(File::open(&path).expect("the file opens"));
4480 std::thread::scope(|scope| {
4481 for _ in 0..8 {
4482 let file = Arc::clone(&file);
4483 scope.spawn(move || {
4484 for _ in 0..64 {
4485 for span in 0..SPANS {
4486 let mut bytes = [0_u8; SPAN];
4487 read_at(&file, (span * SPAN) as u64, &mut bytes)
4488 .expect("the span reads");
4489 assert!(
4490 bytes.iter().all(|byte| *byte == span as u8),
4491 "span {span} came back as {}",
4492 bytes[0],
4493 );
4494 }
4495 }
4496 });
4497 }
4498 });
4499 let mut past = [0_u8; SPAN];
4500 let end = (SPANS * SPAN) as u64;
4501 let error = read_at(&file, end, &mut past).expect_err("a read past the end is refused");
4502 assert!(error.message().contains("ends before its declared length"), "{error}");
4503 drop(file);
4504 let _ = fs::remove_file(&path);
4505 }
4506
4507 #[test]
4513 fn a_writer_puts_a_page_where_it_said_it_did_wherever_the_cursor_has_got_to() {
4514 let path = path("cursor");
4515 let mut writer = Writer::create(
4516 &path,
4517 "items",
4518 vec![
4519 Field::required("id", LogicalType::Integer),
4520 Field::new("text", LogicalType::Varchar),
4521 ],
4522 )
4523 .expect("new file");
4524 writer.append(&sample()).expect("first part");
4525 writer.file.seek(SeekFrom::Start(0)).expect("the cursor goes back to the header");
4526 writer.append(&sample()).expect("second part");
4527 writer.file.seek(SeekFrom::Start(1)).expect("and somewhere useless again");
4528 writer.finish().expect("commit");
4529 let reader = Reader::open(&path).expect("reopen from disk");
4530 assert_eq!(reader.table().rows(), 6);
4531 let ids = reader.read(0, &[0]).expect("the integer page reads back");
4532 assert_eq!(ids.value_at(0, 0), Value::Integer(4));
4533 assert_eq!(ids.value_at(2, 0), Value::Integer(-2));
4534 let text = reader.read(1, &[1]).expect("the text page reads back");
4535 assert_eq!(text.value_at(1, 0), Value::Null);
4536 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4537 let end = reader.table().stripes().iter().flat_map(|stripe| {
4540 stripe
4541 .pages
4542 .iter()
4543 .map(|page| page.offset + u64::from(page.length))
4544 .chain(std::iter::once(stripe.index.offset + u64::from(stripe.index.length)))
4545 });
4546 let last = end.fold(HEADER, u64::max);
4547 let directory = fs::metadata(&path).expect("the file is there").len();
4548 assert!(last <= directory, "a page runs to {last} in a file of {directory} bytes");
4549 fs::remove_file(path).expect("remove scratch file");
4550 }
4551
4552 fn sample() -> Chunk {
4553 Chunk::new(vec![
4554 Vector::from_values(
4555 LogicalType::Integer,
4556 &[Value::Integer(4), Value::Integer(9), Value::Integer(-2)],
4557 )
4558 .expect("integers"),
4559 Vector::from_values(
4560 LogicalType::Varchar,
4561 &[
4562 Value::Varchar("alpha".into()),
4563 Value::Null,
4564 Value::Varchar("long text after a slash".into()),
4565 ],
4566 )
4567 .expect("strings"),
4568 ])
4569 .expect("matching rows")
4570 }
4571
4572 fn sample_ids() -> Chunk {
4573 Chunk::new(vec![
4574 Vector::flat(LogicalType::Integer, Data::Int32(vec![7, 8, 9].into()))
4575 .expect("integers"),
4576 ])
4577 .expect("one column")
4578 }
4579
4580 #[test]
4581 fn committed_file_reopens_and_reads_only_requested_columns() {
4582 let path = path("reopen");
4583 let mut writer = Writer::create(
4584 &path,
4585 "items",
4586 vec![
4587 Field::required("id", LogicalType::Integer),
4588 Field::new("text", LogicalType::Varchar),
4589 ],
4590 )
4591 .expect("new file");
4592 writer.append(&sample()).expect("first part");
4593 writer.append(&sample()).expect("second part");
4594 writer.finish().expect("commit");
4595 let reader = Reader::open(&path).expect("reopen from disk");
4596 assert_eq!(reader.table().rows(), 6);
4597 assert_eq!(reader.table().stripes().len(), 1);
4600 assert_eq!(reader.parts(), 2);
4601 assert_eq!(reader.part_rows(0), 3);
4602 assert_eq!(reader.part_rows(1), 3);
4603 let text = reader.read(1, &[1]).expect("only text page");
4604 assert_eq!(text.width(), 1);
4605 assert_eq!(text.value_at(1, 0), Value::Null);
4606 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4607 let sparse = reader.read_sparse(1, &[1]).expect("one part without its whole page");
4608 assert_eq!(sparse.width(), 1);
4609 assert_eq!(sparse.value_at(1, 0), Value::Null);
4610 assert_eq!(sparse.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4611 assert!(!reader.skips_codes(0, 1, &[0]).expect("alpha is in the stripe"));
4612 assert!(!reader.skips_codes(0, 1, &[2]).expect("long text is in the stripe"));
4613 assert!(reader.skips_codes(0, 1, &[3]).expect("unknown code is absent"));
4614 let count = reader.read(0, &[]).expect("no page is needed for count");
4615 assert_eq!(count.len(), 3);
4616 assert!(reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }]));
4617 assert!(!reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(0) }]));
4618 let integers = reader.top_frequencies(0, 1).expect("valid integer synopsis").expect("kept");
4619 assert_eq!(
4620 integers,
4621 vec![(Value::Integer(-2), 2), (Value::Integer(4), 2), (Value::Integer(9), 2),]
4622 );
4623 let strings = reader.top_frequencies(1, 1).expect("valid string synopsis").expect("kept");
4624 assert_eq!(strings.len(), 3);
4625 assert!(strings.contains(&(Value::Null, 2)));
4626 assert!(strings.contains(&(Value::Varchar("alpha".into()), 2)));
4627 assert!(strings.contains(&(Value::Varchar("long text after a slash".into()), 2)));
4628 fs::remove_file(path).expect("remove scratch file");
4629 }
4630
4631 #[test]
4639 fn runs_handed_over_out_of_order_still_read_back_in_source_order() {
4640 let path = path("interleaved-runs");
4641 let mut writer =
4642 Writer::create(&path, "interleaved", vec![Field::new("v", LogicalType::BigInt)])
4643 .expect("new file");
4644 for morsel in [2_u64, 0, 3, 1] {
4645 let parts = (0..4_u64)
4646 .map(|chunk| {
4647 let first = i64::try_from(morsel * 32 + chunk * 8).expect("small");
4648 let values =
4649 (0..8_i64).map(|row| Value::BigInt(first + row)).collect::<Vec<_>>();
4650 let column =
4651 Vector::from_values(LogicalType::BigInt, &values).expect("a column");
4652 ((morsel, chunk), Chunk::new(vec![column]).expect("one column"))
4653 })
4654 .collect::<Vec<_>>();
4655 writer.append_stripe(parts).expect("a stripe");
4656 }
4657 writer.finish().expect("commit");
4658
4659 let reader = Reader::open(&path).expect("valid directory");
4660 assert_eq!(reader.table().stripes().len(), 4, "a run is a stripe of its own");
4661 assert_eq!(reader.table().rows(), 128);
4662 for part in 0..16_usize {
4663 let read = reader.read(part, &[0]).expect("a part back");
4664 for row in 0..8_usize {
4665 let want = i64::try_from(part * 8 + row).expect("small");
4666 assert_eq!(read.value_at(row, 0), Value::BigInt(want), "part {part} row {row}");
4667 }
4668 }
4669 fs::remove_file(path).expect("remove scratch file");
4670 }
4671
4672 #[test]
4675 fn runs_that_overlap_each_other_are_refused_at_commit() {
4676 let path = path("overlapping-runs");
4677 let mut writer =
4678 Writer::create(&path, "overlapping", vec![Field::new("v", LogicalType::BigInt)])
4679 .expect("new file");
4680 let one = |order: (u64, u64)| {
4681 let column =
4682 Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)]).expect("a column");
4683 (order, Chunk::new(vec![column]).expect("one column"))
4684 };
4685 writer.append_stripe(vec![one((0, 0)), one((0, 2))]).expect("a stripe");
4688 writer.append_stripe(vec![one((0, 1))]).expect("a stripe");
4689 let error = writer.finish().expect_err("the runs overlap");
4690 assert!(error.message().contains("source order"), "{error}");
4691 fs::remove_file(path).expect("remove scratch file");
4692 }
4693
4694 #[test]
4697 fn a_run_longer_than_a_stripe_is_refused() {
4698 let path = path("overlong-run");
4699 let mut writer =
4700 Writer::create(&path, "overlong", vec![Field::new("v", LogicalType::BigInt)])
4701 .expect("new file");
4702 let parts = (0..=STRIPE_PARTS)
4703 .map(|at| {
4704 let column = Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)])
4705 .expect("a column");
4706 let chunk = Chunk::new(vec![column]).expect("one column");
4707 ((0, u64::try_from(at).expect("small")), chunk)
4708 })
4709 .collect::<Vec<_>>();
4710 let error = writer.append_stripe(parts).expect_err("one part too many");
4711 assert!(error.message().contains("more parts than it holds"), "{error}");
4712 fs::remove_file(path).expect("remove scratch file");
4713 }
4714
4715 #[test]
4721 fn parts_past_the_stripe_bound_start_a_new_stripe() {
4722 let path = path("stripe-bound");
4723 let mut writer = Writer::create(
4724 &path,
4725 "items",
4726 vec![
4727 Field::required("id", LogicalType::Integer),
4728 Field::new("text", LogicalType::Varchar),
4729 ],
4730 )
4731 .expect("new file");
4732 let parts = STRIPE_PARTS * 2 + 3;
4733 for part in 0..parts {
4734 let id = part as i32;
4735 let chunk = Chunk::new(vec![
4736 Vector::from_values(
4737 LogicalType::Integer,
4738 &[Value::Integer(id), Value::Integer(-id)],
4739 )
4740 .expect("integers"),
4741 Vector::from_values(
4742 LogicalType::Varchar,
4743 &[Value::Varchar(format!("value {part}")), Value::Null],
4744 )
4745 .expect("strings"),
4746 ])
4747 .expect("matching rows");
4748 writer.append(&chunk).expect("one part");
4749 }
4750 writer.finish().expect("commit");
4751
4752 let reader = Reader::open(&path).expect("reopen from disk");
4753 assert_eq!(reader.parts(), parts);
4754 assert_eq!(reader.table().rows(), parts * 2);
4755 assert_eq!(reader.table().stripes().len(), parts.div_ceil(STRIPE_PARTS));
4756 assert_eq!(reader.table().stripes()[0].parts(), STRIPE_PARTS);
4757 assert_eq!(reader.table().stripes()[0].rows(), STRIPE_PARTS * 2);
4758 assert_eq!(reader.table().stripes()[2].parts(), 3);
4759 for part in (0..parts).rev() {
4762 let dense = reader.read(part, &[0, 1]).expect("a whole page read");
4763 let sparse = reader.read_sparse(part, &[0, 1]).expect("one part read");
4764 for chunk in [&dense, &sparse] {
4765 assert_eq!(chunk.len(), 2, "part {part} has its own row count");
4766 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4767 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
4768 assert_eq!(chunk.value_at(0, 1), Value::Varchar(format!("value {part}")));
4769 assert_eq!(chunk.value_at(1, 1), Value::Null);
4770 }
4771 }
4772 let above = [Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }];
4775 assert!(reader.skips(0, &above), "the first stripe stops at 63");
4776 assert!(!reader.skips(STRIPE_PARTS * 2, &above), "the third stripe reaches 130");
4777 fs::remove_file(path).expect("remove scratch file");
4778 }
4779
4780 fn scattered(n: i64) -> i64 {
4782 n.wrapping_mul(-7_046_029_254_386_353_131)
4783 }
4784
4785 #[test]
4791 fn a_part_is_skipped_when_its_sieve_does_not_hold_the_constant() {
4792 let path = path("sieve-skip");
4793 let mut writer =
4794 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
4795 .expect("new file");
4796 let parts = STRIPE_PARTS + 3;
4797 let per_part = 8;
4798 for part in 0..parts {
4799 let held: Vec<Value> = (0..per_part)
4800 .map(|row| Value::BigInt(scattered((part * per_part + row) as i64)))
4801 .collect();
4802 let chunk =
4803 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
4804 .expect("one column");
4805 writer.append(&chunk).expect("one part");
4806 }
4807 writer.finish().expect("commit");
4808
4809 let reader = Reader::open(&path).expect("reopen from disk");
4810 let probe = |value: i64| Probe {
4811 column: 0,
4812 op: Op::Equal,
4813 value: Bound::Int(i128::from(scattered(value))),
4814 };
4815 for wanted in [0_i64, 9, (parts * per_part - 1) as i64] {
4816 let tests = [probe(wanted)];
4817 let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &tests)).collect();
4818 let home = wanted as usize / per_part;
4819 assert_eq!(kept, vec![home], "only the part holding {wanted} is read");
4820 }
4821 let absent = [probe((parts * per_part) as i64 + 1)];
4822 assert!((0..parts).all(|part| reader.skips(part, &absent)), "no part holds it");
4823 let tests = [probe(0)];
4826 assert!(
4827 reader.table().stripes().iter().all(|stripe| !stripe.zone.skips(&tests)),
4828 "the bounds rule out no stripe at all"
4829 );
4830 fs::remove_file(path).expect("remove scratch file");
4831 }
4832
4833 #[test]
4839 fn a_damaged_sieve_page_is_read_through_rather_than_refused() {
4840 let path = path("sieve-damaged");
4841 let mut writer =
4842 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
4843 .expect("new file");
4844 let held: Vec<Value> = (0..8).map(|row| Value::BigInt(scattered(row))).collect();
4845 let chunk =
4846 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
4847 .expect("one column");
4848 writer.append(&chunk).expect("one part");
4849 writer.finish().expect("commit");
4850
4851 let page =
4852 Reader::open(&path).expect("reopen").table.stripes[0].sieves[0].expect("a sieve page");
4853 let mut file = OpenOptions::new().write(true).open(&path).expect("open the sieve page");
4854 file.seek(SeekFrom::Start(page.offset + u64::from(page.length) - 1)).expect("seek");
4855 file.write_all(&[0xff]).expect("damage one byte");
4856 drop(file);
4857
4858 let reader = Reader::open(&path).expect("reopen the damaged file");
4859 let absent =
4860 [Probe { column: 0, op: Op::Equal, value: Bound::Int(i128::from(scattered(99))) }];
4861 assert!(!reader.skips(0, &absent), "a sieve that cannot be read skips nothing");
4862 assert_eq!(reader.read(0, &[0]).expect("the rows are untouched").len(), 8);
4863 fs::remove_file(path).expect("remove scratch file");
4864 }
4865
4866 #[test]
4877 fn workers_that_want_the_same_stripe_read_it_once() {
4878 let path = path("single-flight");
4879 let mut writer =
4880 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4881 .expect("new file");
4882 for part in 0..STRIPE_PARTS {
4883 let id = part as i32;
4884 let chunk = Chunk::new(vec![
4885 Vector::from_values(
4886 LogicalType::Integer,
4887 &[Value::Integer(id), Value::Integer(-id)],
4888 )
4889 .expect("integers"),
4890 ])
4891 .expect("matching rows");
4892 writer.append(&chunk).expect("one part");
4893 }
4894 writer.finish().expect("commit");
4895
4896 let reader = Reader::open(&path).expect("reopen from disk");
4897 assert_eq!(reader.table().stripes().len(), 1, "one stripe is the point of the test");
4898 let barrier = std::sync::Barrier::new(8);
4899 std::thread::scope(|scope| {
4900 for worker in 0..8 {
4901 let reader = &reader;
4902 let barrier = &barrier;
4903 scope.spawn(move || {
4904 barrier.wait();
4905 for part in (worker..STRIPE_PARTS).step_by(8) {
4906 let chunk = reader.read(part, &[0]).expect("a whole page read");
4907 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4908 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
4909 }
4910 });
4911 }
4912 });
4913 assert_eq!(reader.pages.load(Atomic::Relaxed), 1, "one stripe, one page read, whoever won");
4914 fs::remove_file(path).expect("remove scratch file");
4915 }
4916
4917 #[test]
4930 fn opening_costs_the_same_over_a_thousand_times_the_rows() {
4931 let opened = |label: &str, rows_per_part: i32| {
4932 let path = path(label);
4933 let mut writer =
4934 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4935 .expect("new file");
4936 for part in 0..STRIPE_PARTS * 3 {
4937 let values = (0..rows_per_part)
4941 .map(|row| {
4942 Value::Integer((part as i32 * rows_per_part + row).wrapping_mul(2_654_435))
4943 })
4944 .collect::<Vec<_>>();
4945 let chunk = Chunk::new(vec![
4946 Vector::from_values(LogicalType::Integer, &values).expect("integers"),
4947 ])
4948 .expect("matching rows");
4949 writer.append(&chunk).expect("one part");
4950 }
4951 writer.finish().expect("commit");
4952 let reader = Reader::open(&path).expect("reopen from disk");
4953 let size = fs::metadata(&path).expect("the file is there").len();
4954 let out = (reader.reads(), reader.table().stripes().len(), size);
4955 fs::remove_file(path).expect("remove scratch file");
4956 out
4957 };
4958
4959 let (thin, thin_stripes, thin_size) = opened("open-thin", 1);
4960 let (fat, fat_stripes, fat_size) = opened("open-fat", 1000);
4961 assert_eq!(
4962 thin_stripes, fat_stripes,
4963 "the same stripe count is what makes this a fair ask"
4964 );
4965 assert!(
4966 fat_size > thin_size * 50,
4967 "the fat file has to actually be larger, and it is {fat_size} against {thin_size}"
4968 );
4969
4970 assert_eq!(thin.opening.reads, fat.opening.reads, "the same reads either way");
4971 assert_eq!(thin.pages, 0, "opening read a page");
4972 assert_eq!(fat.pages, 0, "opening read a page");
4973 assert_eq!(thin.indexes, 0, "opening read an index");
4974 assert_eq!(fat.indexes, 0, "opening read an index");
4975 assert!(
4978 fat.opening.bytes < thin.opening.bytes * 2,
4979 "opening the thin file read {} bytes and the fat one read {}",
4980 thin.opening.bytes,
4981 fat.opening.bytes
4982 );
4983 }
4984
4985 #[test]
4993 fn two_opens_of_one_file_cost_the_same_and_the_second_is_not_cheaper() {
4994 let path = path("open-twice");
4995 let mut writer =
4996 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4997 .expect("new file");
4998 for part in 0..STRIPE_PARTS * 3 {
4999 let chunk = Chunk::new(vec![
5000 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
5001 .expect("integers"),
5002 ])
5003 .expect("matching rows");
5004 writer.append(&chunk).expect("one part");
5005 }
5006 writer.finish().expect("commit");
5007
5008 let first = Reader::open(&path).expect("open");
5009 for part in 0..first.parts() {
5012 first.read(part, &[0]).expect("a part");
5013 }
5014 assert!(first.reads().pages > 0, "the scan has to have read something");
5015 let second = Reader::open(&path).expect("open again");
5016
5017 assert_eq!(first.reads().opening, second.reads().opening);
5018 assert_eq!(
5019 second.reads().pages,
5020 0,
5021 "the second open read a page off the back of the first"
5022 );
5023 assert_eq!(second.reads().indexes, 0, "the second open read an index it inherited");
5024 fs::remove_file(path).expect("remove scratch file");
5025 }
5026
5027 #[test]
5035 fn an_index_is_read_once_per_stripe_however_often_the_page_is_evicted() {
5036 let path = path("index-cache");
5037 let mut writer =
5038 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5039 .expect("new file");
5040 let parts = STRIPE_PARTS * (CACHED_STRIPES_PER_COLUMN + 2);
5041 for part in 0..parts {
5042 let id = part as i32;
5043 let chunk = Chunk::new(vec![
5044 Vector::from_values(LogicalType::Integer, &[Value::Integer(id)]).expect("integers"),
5045 ])
5046 .expect("matching rows");
5047 writer.append(&chunk).expect("one part");
5048 }
5049 writer.finish().expect("commit");
5050
5051 let reader = Reader::open(&path).expect("reopen from disk");
5052 let stripes = reader.table().stripes().len();
5053 assert!(stripes > CACHED_STRIPES_PER_COLUMN, "the page cache has to be too small for this");
5054 for _ in 0..2 {
5056 for part in 0..parts {
5057 let chunk = reader.read(part, &[0]).expect("a part");
5058 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5059 }
5060 }
5061 assert_eq!(reader.indexes.load(Atomic::Relaxed), stripes, "one index read per stripe");
5062 assert!(
5063 reader.pages.load(Atomic::Relaxed) > stripes,
5064 "the pages are the ones that get read again, which is what makes the index count mean \
5065 something"
5066 );
5067 fs::remove_file(path).expect("remove scratch file");
5068 }
5069
5070 #[test]
5079 fn a_worker_per_stripe_reads_its_page_once_when_the_cache_was_told_to_expect_it() {
5080 let workers = CACHED_STRIPES_PER_COLUMN + 4;
5081 let path = path("stripe-per-worker");
5082 let mut writer =
5083 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5084 .expect("new file");
5085 for part in 0..STRIPE_PARTS * workers {
5086 let chunk = Chunk::new(vec![
5087 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
5088 .expect("integers"),
5089 ])
5090 .expect("matching rows");
5091 writer.append(&chunk).expect("one part");
5092 }
5093 writer.finish().expect("commit");
5094
5095 let read = |told: bool| {
5096 let reader = Reader::open(&path).expect("reopen from disk");
5097 assert_eq!(reader.table().stripes().len(), workers, "a stripe per worker");
5098 if told {
5099 reader.keep_stripes(workers);
5100 }
5101 let barrier = std::sync::Barrier::new(workers);
5102 std::thread::scope(|scope| {
5103 for (worker, run) in reader.stripe_parts().into_iter().enumerate() {
5104 let reader = &reader;
5105 let barrier = &barrier;
5106 scope.spawn(move || {
5107 for part in run {
5108 barrier.wait();
5109 let chunk = reader.read(part, &[0]).expect("a part of my own stripe");
5110 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5111 }
5112 assert!(worker < workers);
5113 });
5114 }
5115 });
5116 reader.pages.load(Atomic::Relaxed)
5117 };
5118
5119 assert_eq!(read(true), workers, "one page read per stripe and no more");
5120 assert!(read(false) > workers, "a cache that small is read again on every part");
5121 fs::remove_file(path).expect("remove scratch file");
5122 }
5123
5124 #[test]
5129 fn a_damaged_index_page_is_an_error() {
5130 let path = path("damaged-index");
5131 let mut writer =
5132 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5133 .expect("new file");
5134 writer.append(&sample_ids()).expect("first part");
5135 writer.append(&sample_ids()).expect("second part");
5136 writer.finish().expect("commit");
5137
5138 let reader = Reader::open(&path).expect("valid directory");
5139 let index = reader.table.stripes[0].index;
5140 let mut byte = [0; 1];
5141 read_at(&reader.file, index.offset, &mut byte).expect("the first part length");
5142 let mut file = OpenOptions::new().write(true).open(&path).expect("open index page");
5143 file.seek(SeekFrom::Start(index.offset)).expect("index start");
5144 file.write_all(&[!byte[0]]).expect("damage the first part length");
5145 let error = reader.read(1, &[0]).expect_err("a damaged index must not be used");
5146 assert!(error.message().contains("index page section checksum differs"), "{error}");
5147 fs::remove_file(path).expect("remove scratch file");
5148 }
5149
5150 #[test]
5157 fn every_integer_width_round_trips_through_a_page() {
5158 let path = path("integer-widths");
5159 let columns = [
5160 (LogicalType::TinyInt, vec![Value::TinyInt(i8::MIN), Value::TinyInt(i8::MAX)]),
5161 (LogicalType::UTinyInt, vec![Value::UTinyInt(0), Value::UTinyInt(u8::MAX)]),
5162 (LogicalType::SmallInt, vec![Value::SmallInt(i16::MIN), Value::SmallInt(i16::MAX)]),
5163 (LogicalType::USmallInt, vec![Value::USmallInt(0), Value::USmallInt(u16::MAX)]),
5164 (LogicalType::Integer, vec![Value::Integer(i32::MIN), Value::Integer(i32::MAX)]),
5165 (LogicalType::UInteger, vec![Value::UInteger(0), Value::UInteger(u32::MAX)]),
5166 (LogicalType::BigInt, vec![Value::BigInt(i64::MIN), Value::BigInt(i64::MAX)]),
5167 (LogicalType::UBigInt, vec![Value::UBigInt(0), Value::UBigInt(u64::MAX)]),
5168 ];
5169 let fields = columns
5170 .iter()
5171 .enumerate()
5172 .map(|(at, (ty, _))| Field::required(format!("c{at}"), ty.clone()))
5173 .collect::<Vec<_>>();
5174 let vectors = columns
5175 .iter()
5176 .map(|(ty, values)| Vector::from_values(ty.clone(), values).expect("a vector"))
5177 .collect::<Vec<_>>();
5178 let mut writer = Writer::create(&path, "widths", fields).expect("new file");
5179 writer.append(&Chunk::new(vectors).expect("matching rows")).expect("one stripe");
5180 writer.finish().expect("commit");
5181
5182 let reader = Reader::open(&path).expect("reopen from disk");
5183 let wanted = (0..columns.len()).collect::<Vec<_>>();
5184 let read = reader.read(0, &wanted).expect("every column");
5185 assert_eq!(read.len(), 2);
5186 for (at, (ty, values)) in columns.iter().enumerate() {
5188 assert_eq!(read.value_at(0, at), values[0], "the low end of {ty}");
5189 assert_eq!(read.value_at(1, at), values[1], "the high end of {ty}");
5190 }
5191 fs::remove_file(path).expect("remove scratch file");
5192 }
5193
5194 #[test]
5195 fn numeric_frequency_candidates_keep_bounded_row_ordinals() {
5196 let path = path("frequency-ordinals");
5197 let mut writer =
5198 Writer::create(&path, "items", vec![Field::required("id", LogicalType::BigInt)])
5199 .expect("new file");
5200 let mut values = Vec::new();
5201 for leader in 0..10_i64 {
5202 values.extend(std::iter::repeat_n(leader, 100));
5203 }
5204 values.extend(1_000_i64..41_000);
5205 for part in values.chunks(1_024) {
5206 let vector = Vector::flat(LogicalType::BigInt, Data::Int64(part.to_vec().into()))
5207 .expect("big integers");
5208 writer.append(&Chunk::new(vec![vector]).expect("one column")).expect("one stripe");
5209 }
5210 writer.finish().expect("commit");
5211
5212 let reader = Reader::open(&path).expect("reopen from disk");
5213 let occurrences =
5214 reader.frequency_occurrences(0).expect("valid metadata").expect("bounded ordinals");
5215 assert!(occurrences.omitted_max < 100);
5216 assert!(occurrences.ordinals.len() <= FREQUENCY_ORDINALS);
5217 assert!(occurrences.ordinals.windows(2).all(|pair| pair[0] < pair[1]));
5218 assert_eq!(&occurrences.ordinals[..1_000], &(0_u64..1_000).collect::<Vec<_>>());
5219 fs::remove_file(path).expect("remove scratch file");
5220 }
5221
5222 #[test]
5228 fn a_file_from_another_format_says_which_format_it_is() {
5229 let older = path("older-format");
5230 let mut writer =
5231 Writer::create(&older, "items", vec![Field::new("id", LogicalType::Integer)])
5232 .expect("new file");
5233 let chunk = Chunk::new(vec![
5234 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
5235 .expect("integers"),
5236 ])
5237 .expect("chunk");
5238 writer.append(&chunk).expect("page written");
5239 writer.finish().expect("commit");
5240
5241 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
5242 file.seek(SeekFrom::Start(8)).expect("the version follows the magic");
5243 file.write_all(&(FORMAT - 1).to_le_bytes()).expect("write an older version");
5244 drop(file);
5245 let complaint = Reader::open(&older).expect_err("an older format is refused").to_string();
5246 assert!(complaint.contains(&format!("format {}", FORMAT - 1)), "{complaint}");
5247 assert!(complaint.contains(&format!("format {FORMAT}")), "{complaint}");
5248
5249 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
5250 file.seek(SeekFrom::Start(0)).expect("the magic is first");
5251 file.write_all(b"NOTRUDB!").expect("write another engine's magic");
5252 drop(file);
5253 let complaint = Reader::open(&older).expect_err("a foreign file is refused").to_string();
5254 assert!(complaint.contains("magic"), "{complaint}");
5255 assert!(!complaint.contains("format"), "a version has nothing to do with it: {complaint}");
5256 fs::remove_file(older).expect("remove scratch file");
5257 }
5258
5259 #[test]
5260 fn an_unfinished_or_damaged_file_does_not_answer_with_partial_rows() {
5261 let unfinished = path("unfinished");
5262 let mut writer =
5263 Writer::create(&unfinished, "items", vec![Field::new("id", LogicalType::Integer)])
5264 .expect("new file");
5265 let chunk = Chunk::new(vec![
5266 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
5267 .expect("integers"),
5268 ])
5269 .expect("chunk");
5270 writer.append(&chunk).expect("page written");
5271 drop(writer);
5272 assert!(Reader::open(&unfinished).is_err(), "no directory was committed");
5273 fs::remove_file(unfinished).expect("remove scratch file");
5274
5275 let damaged = path("damaged");
5276 let mut writer =
5277 Writer::create(&damaged, "items", vec![Field::new("id", LogicalType::Integer)])
5278 .expect("new file");
5279 writer.append(&chunk).expect("page written");
5280 writer.finish().expect("commit");
5281 let reader = Reader::open(&damaged).expect("valid directory");
5282 let mut file =
5283 OpenOptions::new().write(true).open(&damaged).expect("open for a damaged page");
5284 file.seek(SeekFrom::Start(HEADER + 1)).expect("inside first page");
5285 file.write_all(&[255]).expect("damage one byte");
5286 assert!(reader.read(0, &[0]).is_err(), "page checksum rejects corruption");
5287 fs::remove_file(damaged).expect("remove scratch file");
5288 }
5289
5290 #[test]
5291 fn damaged_lazy_dictionary_payload_is_an_error() {
5292 let path = path("damaged-dictionary");
5293 let mut writer = Writer::create(
5294 &path,
5295 "items",
5296 vec![
5297 Field::required("id", LogicalType::Integer),
5298 Field::new("text", LogicalType::Varchar),
5299 ],
5300 )
5301 .expect("new file");
5302 writer.append(&sample()).expect("stripe written");
5303 writer.finish().expect("commit");
5304
5305 let reader = Reader::open(&path).expect("valid directory");
5306 let dictionary = reader.table.dictionaries[1].expect("string dictionary page");
5307 let mut header = [0; 12];
5310 read_at(&reader.file, dictionary.offset, &mut header).expect("dictionary header");
5311 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
5312 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
5313 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
5314 let index_len =
5315 12 + (count + 1) * 4 + (blocks * 2 + rank_blocks) * 8 + count * RANK_ENTRY as u64;
5316 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5317 file.seek(SeekFrom::Start(dictionary.offset + index_len))
5318 .expect("inside dictionary payload");
5319 file.write_all(&[255]).expect("damage dictionary payload");
5320
5321 let chunk = reader.read(0, &[1]).expect("code page and dictionary index remain valid");
5322 let error =
5323 chunk.validate_external().expect_err("payload corruption must reach the caller");
5324 assert!(error.message().contains("payload checksum differs"), "{error}");
5325 fs::remove_file(path).expect("remove scratch file");
5326 }
5327
5328 #[test]
5335 fn a_dictionary_over_many_blocks_checks_every_block_of_it() {
5336 let path = path("dictionary-blocks");
5337 let value =
5338 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5339 let parts = 30;
5340 let per_part = 1000;
5341 let mut writer =
5342 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5343 .expect("new file");
5344 for part in 0..parts {
5345 let values = (0..per_part)
5346 .map(|row| Value::Varchar(value(part * per_part + row)))
5347 .collect::<Vec<_>>();
5348 let chunk = Chunk::new(vec![
5349 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5350 ])
5351 .expect("matching rows");
5352 writer.append(&chunk).expect("a part");
5353 }
5354 writer.finish().expect("commit");
5355
5356 let reader = Reader::open(&path).expect("reopen from disk");
5357 let dictionary = reader.table.dictionaries[0].expect("string dictionary page");
5358 assert!(
5359 parts * per_part > TEXT_PAYLOAD_VALUES * 4,
5360 "the dictionary has to be several blocks for this to be testing anything"
5361 );
5362 for part in [0, parts - 1] {
5363 let chunk = reader.read(part, &[0]).expect("a part");
5364 chunk.validate_external().expect("every payload block checks out");
5365 assert_eq!(chunk.value_at(0, 0), Value::Varchar(value(part * per_part)));
5366 }
5367
5368 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5369 file.seek(SeekFrom::Start(dictionary.offset + u64::from(dictionary.length) - 4))
5370 .expect("the last bytes of the page are payload");
5371 file.write_all(&[255]).expect("damage the last payload block");
5372 let reader = Reader::open(&path).expect("the directory and the index are untouched");
5373 let chunk = reader.read(parts - 1, &[0]).expect("the code page remains valid");
5374 let error = chunk.validate_external().expect_err("the damage must reach the caller");
5375 assert!(error.message().contains("payload checksum differs"), "{error}");
5376 fs::remove_file(path).expect("remove scratch file");
5377 }
5378
5379 #[test]
5391 fn a_global_dictionary_is_opened_once_however_many_workers_ask_at_once() {
5392 let path = path("dictionary-once");
5393 let parts = 8;
5394 let per_part = 500;
5395 let value =
5396 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5397 let mut writer =
5398 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5399 .expect("new file");
5400 for part in 0..parts {
5401 let values = (0..per_part)
5402 .map(|row| Value::Varchar(value(part * per_part + row)))
5403 .collect::<Vec<_>>();
5404 let chunk = Chunk::new(vec![
5405 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5406 ])
5407 .expect("matching rows");
5408 writer.append(&chunk).expect("a part");
5409 }
5410 writer.finish().expect("commit");
5411
5412 let reader = Reader::open(&path).expect("reopen from disk");
5413 assert!(reader.table.dictionaries[0].is_some(), "the column has to have one to share");
5414 assert_eq!(reader.reads().dictionaries, 0, "opening the file does not open a dictionary");
5415
5416 let workers = 16;
5417 let gate = std::sync::Barrier::new(workers);
5418 std::thread::scope(|scope| {
5419 for worker in 0..workers {
5420 let reader = reader.clone();
5421 let gate = &gate;
5422 scope.spawn(move || {
5423 gate.wait();
5424 let chunk = reader.read(worker % parts, &[0]).expect("a part");
5425 assert_eq!(
5426 chunk.value_at(0, 0),
5427 Value::Varchar(value((worker % parts) * per_part))
5428 );
5429 });
5430 }
5431 });
5432
5433 assert_eq!(reader.reads().dictionaries, 1, "sixteen workers, one dictionary, one open");
5434 fs::remove_file(path).expect("remove scratch file");
5435 }
5436
5437 #[test]
5442 fn a_damaged_sorted_order_is_an_error() {
5443 let path = path("damaged-order");
5444 let mut writer = Writer::create(
5445 &path,
5446 "items",
5447 vec![
5448 Field::required("id", LogicalType::Integer),
5449 Field::new("text", LogicalType::Varchar),
5450 ],
5451 )
5452 .expect("new file");
5453 writer.append(&sample()).expect("stripe written");
5454 writer.finish().expect("commit");
5455
5456 let reader = Reader::open(&path).expect("valid directory");
5457 let page = reader.table.dictionaries[1].expect("string dictionary page");
5458 let mut header = [0; 12];
5459 read_at(&reader.file, page.offset, &mut header).expect("dictionary header");
5460 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
5461 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
5462 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
5463 let index_len = 12 + (count + 1) * 4 + (blocks * 2 + rank_blocks) * 8;
5464 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5465 file.seek(SeekFrom::Start(page.offset + index_len)).expect("the first head");
5466 file.write_all(&[255]).expect("damage the order");
5467
5468 let dictionary = reader.dictionary(1).expect("read").expect("a string column has one");
5469 let error = dictionary.compare_rank(0, b"anything").expect_err("a damaged order is caught");
5470 assert!(error.message().contains("rank checksum differs"), "{error}");
5471 fs::remove_file(path).expect("remove scratch file");
5472 }
5473
5474 #[test]
5478 fn a_global_dictionary_carries_the_sorted_order_of_its_values() {
5479 let spellings = ["overlong1z", "b", "", "overlong1a", "overlong", "ab", "a", "overlong1"];
5482 let path = path("dictionary-order");
5483 let mut writer =
5484 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
5485 .expect("new file");
5486 writer
5487 .append(
5488 &Chunk::new(vec![
5489 Vector::from_values(
5490 LogicalType::Varchar,
5491 &spellings.map(|text| Value::Varchar(text.into())),
5492 )
5493 .expect("strings"),
5494 ])
5495 .expect("one column"),
5496 )
5497 .expect("stripe written");
5498 writer.finish().expect("commit");
5499
5500 let reader = Reader::open(&path).expect("valid directory");
5501 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
5502 let count = dictionary.ranks().expect("a v10 file stores one");
5503 assert_eq!(count, spellings.len(), "every distinct value has a rank");
5504 let order = (0..count)
5505 .map(|rank| dictionary.code_at_rank(rank).expect("a code"))
5506 .collect::<Vec<_>>();
5507 let mut seen = order.clone();
5508 seen.sort_unstable();
5509 assert_eq!(seen, (0..spellings.len() as u32).collect::<Vec<_>>(), "a permutation of codes");
5510
5511 let ranked = order
5512 .iter()
5513 .map(|&code| {
5514 dictionary.try_bytes_at(code as usize).expect("read").expect("a value").to_vec()
5515 })
5516 .collect::<Vec<_>>();
5517 let mut expected = spellings.map(|text| text.as_bytes().to_vec()).to_vec();
5518 expected.sort();
5519 assert_eq!(ranked, expected, "rank order is value order");
5520
5521 for (rank, value) in expected.iter().enumerate() {
5524 assert_eq!(
5525 dictionary.compare_rank(rank, value).expect("compare"),
5526 Ordering::Equal,
5527 "rank {rank} is its own value"
5528 );
5529 if rank > 0 {
5530 assert_eq!(
5531 dictionary.compare_rank(rank - 1, value).expect("compare"),
5532 Ordering::Less,
5533 "rank {rank} follows the one before it"
5534 );
5535 }
5536 }
5537 fs::remove_file(path).expect("remove scratch file");
5538 }
5539
5540 #[test]
5541 fn damaged_membership_cannot_skip_a_string_page() {
5542 let path = path("damaged-membership");
5543 let mut writer = Writer::create(
5544 &path,
5545 "items",
5546 vec![
5547 Field::required("id", LogicalType::Integer),
5548 Field::new("text", LogicalType::Varchar),
5549 ],
5550 )
5551 .expect("new file");
5552 writer.append(&sample()).expect("stripe written");
5553 writer.finish().expect("commit");
5554
5555 let reader = Reader::open(&path).expect("valid directory");
5556 let membership = reader.table.stripes[0].memberships[1].expect("string membership");
5557 let mut file = OpenOptions::new().write(true).open(&path).expect("open membership page");
5558 file.seek(SeekFrom::Start(membership.offset)).expect("membership start");
5559 file.write_all(&[255]).expect("damage membership");
5560 let error = reader.skips_codes(0, 1, &[3]).expect_err("corruption must not skip rows");
5561 assert!(error.message().contains("membership page checksum differs"), "{error}");
5562 fs::remove_file(path).expect("remove scratch file");
5563 }
5564
5565 #[test]
5566 fn membership_delta_stream_is_sorted_exact_and_bounded() {
5567 let unique = unique_codes(&[900, 4, 4, 72, 9, u32::MAX]);
5568 assert_eq!(unique, [4, 9, 72, 900, u32::MAX]);
5569 let encoded = encode_membership(&unique);
5570 assert_eq!(
5571 decode_membership(&encoded).expect("valid membership"),
5572 [4, 9, 72, 900, u32::MAX]
5573 );
5574 let merged = merged_codes(vec![vec![4, 900], vec![9, 900, u32::MAX], vec![72]]);
5577 assert_eq!(merged, [4, 9, 72, 900, u32::MAX]);
5578 assert_eq!(
5579 decode_membership(&encode_membership(&merged)).expect("valid membership"),
5580 unique
5581 );
5582 assert!(decode_membership(&[1, 0x80]).is_err(), "a truncated varint is invalid");
5583 assert!(
5584 decode_membership(&[1, 0xff, 0xff, 0xff, 0xff, 0x10]).is_err(),
5585 "a value past u32 is invalid"
5586 );
5587 }
5588
5589 #[test]
5590 fn a_global_dictionary_may_be_larger_than_one_column_page() {
5591 let dictionary = Page {
5592 offset: HEADER,
5593 length: u32::try_from(MAX_PAGE + 1).expect("the page bound fits on disk"),
5594 hash: 0,
5595 };
5596 let table = Table {
5597 name: "items".to_owned(),
5598 fields: vec![Field::new("text", LogicalType::Varchar)],
5599 stripes: Vec::new(),
5600 rows: 0,
5601 dictionaries: vec![Some(dictionary)],
5602 frequencies: vec![None],
5603 };
5604 let directory = encode_directory(&table).expect("directory");
5605 let file_size = dictionary.offset + u64::from(dictionary.length) + 1;
5606
5607 let decoded = decode_directory(&directory, file_size).expect("large lazy dictionary");
5608 assert_eq!(decoded.dictionaries[0].expect("dictionary").length, dictionary.length);
5609 }
5610
5611 #[test]
5612 fn a_column_with_one_value_everywhere_costs_almost_nothing_a_row() {
5613 let path = path("constant-codes");
5614 let mut writer =
5615 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
5616 .expect("new file");
5617 let empty = vec![Value::Varchar(String::new()); 1024];
5618 for _ in 0..4 {
5619 let column = Vector::from_values(LogicalType::Varchar, &empty).expect("strings");
5620 writer.append(&Chunk::new(vec![column]).expect("one column")).expect("a part");
5621 }
5622 writer.finish().expect("commit");
5623
5624 let reader = Reader::open(&path).expect("valid directory");
5625 let pages = reader.layout().columns.first().expect("one column").pages;
5626 assert!(pages < 256, "{pages} bytes of pages for 4,096 rows of one value");
5630 let read = reader.read(3, &[0]).expect("the last part back");
5631 assert_eq!(read.value_at(0, 0), Value::Varchar(String::new()));
5632 assert_eq!(read.value_at(1023, 0), Value::Varchar(String::new()));
5633 fs::remove_file(path).expect("remove scratch file");
5634 }
5635
5636 #[test]
5637 fn a_cascade_value_too_wide_for_its_column_is_refused_rather_than_cut() {
5638 let over = vec![i64::from(i32::MAX) + 1];
5641 let error = narrowed(&LogicalType::Integer, over).expect_err("a page that disagrees");
5642 assert!(format!("{error}").contains("not of its type"), "{error}");
5643 assert!(narrowed(&LogicalType::BigInt, vec![i64::MIN]).is_ok(), "bigint holds all of i64");
5644 assert!(narrowed(&LogicalType::Varchar, vec![0]).is_err(), "strings are not integers");
5645 }
5646
5647 #[test]
5648 fn a_code_stream_the_cascade_cannot_shrink_is_left_alone() {
5649 let mut state: u32 = 0x9e37_79b9;
5653 let spread: Vec<u32> = (0..1024)
5654 .map(|_| {
5655 state ^= state << 13;
5656 state ^= state >> 17;
5657 state ^= state << 5;
5658 state
5659 })
5660 .collect();
5661 assert_eq!(encoded_codes(&spread).expect("no failure"), None);
5662 let near: Vec<u32> = (0..1024).collect();
5663 let coded = encoded_codes(&near).expect("no failure").expect("counting up is packable");
5664 assert!(coded.len() < near.len() * 4, "{} bytes for a run of 1,024", coded.len());
5665 }
5666
5667 #[test]
5673 fn two_writes_of_the_same_rows_give_the_same_bytes() {
5674 fn written(path: &PathBuf) {
5675 let fields = (0..40)
5676 .map(|column| {
5677 let ty =
5678 if column % 4 == 0 { LogicalType::Varchar } else { LogicalType::BigInt };
5679 Field::new(format!("c{column}"), ty)
5680 })
5681 .collect::<Vec<_>>();
5682 let mut writer = Writer::create(path, "wide", fields).expect("new file");
5683 for part in 0..70_u64 {
5684 let columns = (0..40)
5685 .map(|column| {
5686 let values = (0..64_u64)
5687 .map(|row| {
5688 let seed = part.wrapping_mul(31).wrapping_add(row);
5689 if column % 4 == 0 {
5690 Value::Varchar(format!("v{}", seed % 17))
5691 } else {
5692 Value::BigInt(i64::try_from(seed % 97).expect("small"))
5693 }
5694 })
5695 .collect::<Vec<_>>();
5696 let ty = if column % 4 == 0 {
5697 LogicalType::Varchar
5698 } else {
5699 LogicalType::BigInt
5700 };
5701 Vector::from_values(ty, &values).expect("a column")
5702 })
5703 .collect::<Vec<_>>();
5704 writer.append(&Chunk::new(columns).expect("forty columns")).expect("a part");
5705 }
5706 writer.finish().expect("commit");
5707 }
5708
5709 let first = path("repeatable-one");
5710 let second = path("repeatable-two");
5711 written(&first);
5712 written(&second);
5713 let left = fs::read(&first).expect("the first file");
5714 let right = fs::read(&second).expect("the second file");
5715 assert_eq!(left.len(), right.len(), "two writes of the same rows differ in length");
5716 assert!(left == right, "two writes of the same rows differ in their bytes");
5717
5718 let reader = Reader::open(&first).expect("valid directory");
5721 assert_eq!(reader.table().rows(), 70 * 64);
5722 let read = reader.read(0, &[0, 1]).expect("the first part back");
5723 assert_eq!(read.value_at(0, 0), Value::Varchar("v0".to_owned()));
5724 assert_eq!(read.value_at(0, 1), Value::BigInt(0));
5725 fs::remove_file(first).expect("remove scratch file");
5726 fs::remove_file(second).expect("remove scratch file");
5727 }
5728}