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 = 16;
51const HEADER: u64 = 80;
52const SLOT_BYTES: usize = 28;
53const MAX_PAGE: usize = 256 * 1024 * 1024;
54const MAX_DIRECTORY: usize = 128 * 1024 * 1024;
55const FREQUENCIES: &[u8; 8] = b"RUDBFQ2\0";
56const FREQUENCY_CANDIDATES: usize = 32_768;
57const FREQUENCY_ENTRIES: usize = 512;
58const FREQUENCY_BUILD_RANK: usize = 10;
59const FREQUENCY_ORDINALS: usize = 65_536;
60const MAX_FREQUENCY_WORKERS: usize = 32;
67
68const MAX_ENCODE_WORKERS: usize = 32;
75
76const SIEVE_BUDGET: usize = 8 * 1024;
84
85fn io(error: std::io::Error) -> Error {
86 Error::io(error.to_string())
87}
88
89fn invalid(message: &str) -> Error {
90 Error::invalid_input(format!("invalid rudb native file: {message}"))
91}
92
93fn sum(counts: impl Iterator<Item = u64>) -> u64 {
95 counts.fold(0, u64::saturating_add)
96}
97
98fn span_bytes(spans: &[Span], at: usize) -> u64 {
100 spans.get(at).map_or(0, |span| u64::from(span.length))
101}
102
103fn page_bytes(pages: &[Option<Page>], at: usize) -> u64 {
105 pages.get(at).and_then(Option::as_ref).map_or(0, Page::bytes)
106}
107
108fn checksum(bytes: &[u8]) -> u64 {
109 const P1: u64 = 11_400_714_785_074_694_791;
110 const P2: u64 = 14_029_467_366_897_019_727;
111 const P3: u64 = 1_609_587_929_392_839_161;
112 const P4: u64 = 9_650_029_242_287_828_579;
113 const P5: u64 = 2_870_177_450_012_600_261;
114 let round = |state: u64, word: u64| {
115 state.wrapping_add(word.wrapping_mul(P2)).rotate_left(31).wrapping_mul(P1)
116 };
117 let merge = |state: u64, lane: u64| (state ^ round(0, lane)).wrapping_mul(P1).wrapping_add(P4);
118 let word =
119 |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("eight checksum bytes"));
120
121 let mut at = 0;
122 let mut hash = if bytes.len() >= 32 {
123 let mut one = P1.wrapping_add(P2);
124 let mut two = P2;
125 let mut three = 0;
126 let mut four = 0_u64.wrapping_sub(P1);
127 while at + 32 <= bytes.len() {
128 one = round(one, word(at));
129 two = round(two, word(at + 8));
130 three = round(three, word(at + 16));
131 four = round(four, word(at + 24));
132 at += 32;
133 }
134 let combined = one
135 .rotate_left(1)
136 .wrapping_add(two.rotate_left(7))
137 .wrapping_add(three.rotate_left(12))
138 .wrapping_add(four.rotate_left(18));
139 merge(merge(merge(merge(combined, one), two), three), four)
140 } else {
141 P5
142 };
143 hash = hash.wrapping_add(bytes.len() as u64);
144 while at + 8 <= bytes.len() {
145 hash ^= round(0, word(at));
146 hash = hash.rotate_left(27).wrapping_mul(P1).wrapping_add(P4);
147 at += 8;
148 }
149 if at + 4 <= bytes.len() {
150 let tail = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four checksum bytes"));
151 hash ^= u64::from(tail).wrapping_mul(P1);
152 hash = hash.rotate_left(23).wrapping_mul(P2).wrapping_add(P3);
153 at += 4;
154 }
155 while at < bytes.len() {
156 hash ^= u64::from(bytes[at]).wrapping_mul(P5);
157 hash = hash.rotate_left(11).wrapping_mul(P1);
158 at += 1;
159 }
160 hash ^= hash >> 33;
161 hash = hash.wrapping_mul(P2);
162 hash ^= hash >> 29;
163 hash = hash.wrapping_mul(P3);
164 hash ^ (hash >> 32)
165}
166
167#[derive(Debug, Clone, Copy)]
168struct Slot {
169 offset: u64,
170 length: u32,
171 generation: u64,
172 hash: u64,
173}
174
175impl Slot {
176 fn bytes(self) -> [u8; SLOT_BYTES] {
177 let mut result = [0; SLOT_BYTES];
178 result[..8].copy_from_slice(&self.offset.to_le_bytes());
179 result[8..12].copy_from_slice(&self.length.to_le_bytes());
180 result[12..20].copy_from_slice(&self.generation.to_le_bytes());
181 result[20..28].copy_from_slice(&self.hash.to_le_bytes());
182 result
183 }
184
185 fn read(bytes: &[u8]) -> Self {
186 Self {
187 offset: u64::from_le_bytes(bytes[..8].try_into().expect("eight bytes")),
188 length: u32::from_le_bytes(bytes[8..12].try_into().expect("four bytes")),
189 generation: u64::from_le_bytes(bytes[12..20].try_into().expect("eight bytes")),
190 hash: u64::from_le_bytes(bytes[20..28].try_into().expect("eight bytes")),
191 }
192 }
193}
194
195#[derive(Debug, Clone, Copy)]
196struct Page {
197 offset: u64,
198 length: u32,
199 hash: u64,
200}
201
202impl Page {
203 fn bytes(&self) -> u64 {
205 u64::from(self.length)
206 }
207}
208
209#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
210enum FrequencyValue {
211 Null,
212 Integer(i128),
213 Code(u32),
214}
215
216#[derive(Debug, Clone)]
217struct FrequencyEntry {
218 value: FrequencyValue,
219 count: u64,
220}
221
222#[derive(Debug, Clone)]
227struct FrequencySummary {
228 entries: Vec<FrequencyEntry>,
229 omitted_max: u64,
230 ordinals: Vec<u64>,
231}
232
233#[derive(Debug, Clone, PartialEq, Eq)]
235pub struct FrequencyOccurrences {
236 pub omitted_max: u64,
238 pub ordinals: Vec<u64>,
240}
241
242#[derive(Debug, Clone, Copy, Default)]
249struct Span {
250 offset: u64,
251 length: u32,
252}
253
254#[derive(Debug, Clone)]
256pub struct Stripe {
257 rows: usize,
258 parts: Vec<u32>,
261 index: Span,
265 pages: Vec<Span>,
266 memberships: Vec<Option<Page>>,
267 sieves: Vec<Option<Page>>,
270 zone: Zone,
271}
272
273impl Stripe {
274 #[must_use]
276 pub fn rows(&self) -> usize {
277 self.rows
278 }
279
280 #[must_use]
282 pub fn parts(&self) -> usize {
283 self.parts.len()
284 }
285}
286
287#[derive(Debug, Clone)]
289pub struct Table {
290 name: String,
291 fields: Vec<Field>,
292 stripes: Vec<Stripe>,
293 rows: usize,
294 dictionaries: Vec<Option<Page>>,
295 frequencies: Vec<Option<FrequencySummary>>,
296}
297
298impl Table {
299 #[must_use]
301 pub fn name(&self) -> &str {
302 &self.name
303 }
304
305 #[must_use]
307 pub fn fields(&self) -> &[Field] {
308 &self.fields
309 }
310
311 #[must_use]
313 pub fn rows(&self) -> usize {
314 self.rows
315 }
316
317 #[must_use]
319 pub fn stripes(&self) -> &[Stripe] {
320 &self.stripes
321 }
322}
323
324#[derive(Debug, Clone)]
326pub struct ColumnLayout {
327 pub name: String,
329 pub kind: String,
331 pub pages: u64,
333 pub memberships: u64,
335 pub sieves: u64,
337 pub dictionary: u64,
339}
340
341impl ColumnLayout {
342 #[must_use]
344 pub fn total(&self) -> u64 {
345 self.pages
346 .saturating_add(self.memberships)
347 .saturating_add(self.sieves)
348 .saturating_add(self.dictionary)
349 }
350}
351
352#[derive(Debug, Clone)]
363pub struct Layout {
364 pub file: u64,
366 pub rows: usize,
368 pub stripes: usize,
370 pub parts: usize,
372 pub columns: Vec<ColumnLayout>,
374 pub indexes: u64,
377 pub directory: u64,
379 pub header: u64,
381}
382
383impl Layout {
384 #[must_use]
386 pub fn columns_total(&self) -> u64 {
387 self.columns.iter().map(ColumnLayout::total).fold(0, u64::saturating_add)
388 }
389
390 #[must_use]
396 pub fn unaccounted(&self) -> u64 {
397 self.file
398 .saturating_sub(self.columns_total())
399 .saturating_sub(self.indexes)
400 .saturating_sub(self.directory)
401 .saturating_sub(self.header)
402 }
403}
404
405#[derive(Debug)]
407struct GlobalDictionary {
408 primary: HashMap<u64, u32>,
409 collisions: HashMap<u64, Vec<u32>>,
410 offsets: Vec<u32>,
411 payload: Vec<u8>,
412 counts: Vec<u64>,
413 nulls: u64,
414}
415
416impl GlobalDictionary {
417 fn new() -> Self {
418 Self {
419 primary: HashMap::new(),
420 collisions: HashMap::new(),
421 offsets: vec![0],
422 payload: Vec::new(),
423 counts: Vec::new(),
424 nulls: 0,
425 }
426 }
427
428 fn bytes(&self, code: u32) -> Option<&[u8]> {
429 let start = *self.offsets.get(code as usize)? as usize;
430 let end = *self.offsets.get(code as usize + 1)? as usize;
431 self.payload.get(start..end)
432 }
433
434 fn code(&mut self, text: &str) -> Result<u32> {
435 let hash = checksum(text.as_bytes());
436 if let Some(&code) = self.primary.get(&hash) {
437 if self.bytes(code) == Some(text.as_bytes()) {
438 return Ok(code);
439 }
440 if let Some(codes) = self.collisions.get(&hash) {
441 if let Some(code) =
442 codes.iter().copied().find(|&code| self.bytes(code) == Some(text.as_bytes()))
443 {
444 return Ok(code);
445 }
446 }
447 let code = self.insert(text)?;
448 self.collisions.entry(hash).or_default().push(code);
449 return Ok(code);
450 }
451 let code = self.insert(text)?;
452 self.primary.insert(hash, code);
453 Ok(code)
454 }
455
456 fn insert(&mut self, text: &str) -> Result<u32> {
457 let code = u32::try_from(self.offsets.len() - 1)
458 .map_err(|_| invalid("global dictionary has too many values"))?;
459 self.payload.extend_from_slice(text.as_bytes());
460 self.offsets.push(
461 u32::try_from(self.payload.len())
462 .map_err(|_| invalid("global dictionary payload exceeds 4 GiB"))?,
463 );
464 self.counts.push(0);
465 Ok(code)
466 }
467
468 fn ranked(&self) -> Vec<(u64, u32)> {
488 let count = self.offsets.len() - 1;
489 let mut ranked = (0..count)
490 .map(|code| {
491 let code = code as u32;
492 (head(self.bytes(code).unwrap_or_default()), code)
493 })
494 .collect::<Vec<_>>();
495 ranked.sort_unstable_by(|left, right| {
496 left.0.cmp(&right.0).then_with(|| self.bytes(left.1).cmp(&self.bytes(right.1)))
497 });
498 ranked
499 }
500
501 fn observe(&mut self, code: u32, null: bool) -> Result<()> {
502 if null {
503 self.nulls = self.nulls.saturating_add(1);
504 return Ok(());
505 }
506 let count = self
507 .counts
508 .get_mut(code as usize)
509 .ok_or_else(|| invalid("global dictionary count code is out of range"))?;
510 *count = count.saturating_add(1);
511 Ok(())
512 }
513}
514
515#[derive(Debug)]
517pub struct Writer {
518 file: File,
519 at: u64,
527 table: Table,
528 generation: u64,
529 order: Vec<((u64, u64), (u64, u64))>,
532 next_order: u64,
533 dictionaries: Vec<Option<GlobalDictionary>>,
534 pending: Vec<PendingChunk>,
535}
536
537#[derive(Debug)]
545struct PendingChunk {
546 order: (u64, u64),
547 chunk: Chunk,
548}
549
550#[derive(Debug)]
556struct ColumnStripe {
557 pages: Vec<Vec<u8>>,
558 codes: Vec<Option<Vec<u32>>>,
559 sieves: Vec<Option<Sieve>>,
560 ranges: Vec<Range>,
561}
562
563fn weight(ty: &LogicalType) -> usize {
571 match ty {
572 LogicalType::Varchar | LogicalType::Blob => 64,
573 LogicalType::BigInt
574 | LogicalType::UBigInt
575 | LogicalType::Timestamp
576 | LogicalType::Double
577 | LogicalType::Decimal { .. } => 8,
578 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date | LogicalType::Float => 4,
579 LogicalType::SmallInt | LogicalType::USmallInt => 2,
580 _ => 1,
581 }
582}
583
584pub const STRIPE_PARTS: usize = 64;
591
592const INDEX_ENTRY: usize = size_of::<u32>() + size_of::<u64>();
594
595fn index_section(parts: usize) -> Result<usize> {
597 parts
598 .checked_mul(INDEX_ENTRY)
599 .and_then(|bytes| bytes.checked_add(size_of::<u64>()))
600 .ok_or_else(|| invalid("index page length overflow"))
601}
602
603impl Writer {
604 pub fn create(
610 path: impl AsRef<Path>,
611 name: impl Into<String>,
612 fields: Vec<Field>,
613 ) -> Result<Self> {
614 for field in &fields {
615 type_tag(&field.ty)?;
616 }
617 let file =
618 OpenOptions::new().write(true).read(true).create_new(true).open(path).map_err(io)?;
619 let mut header = [0; HEADER as usize];
620 header[..8].copy_from_slice(MAGIC);
621 header[8..12].copy_from_slice(&FORMAT.to_le_bytes());
622 write_at(&file, 0, &header)?;
623 Ok(Self {
624 file,
625 at: HEADER,
626 dictionaries: fields
627 .iter()
628 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
629 .collect(),
630 table: Table {
631 name: name.into(),
632 dictionaries: vec![None; fields.len()],
633 fields,
634 stripes: Vec::new(),
635 rows: 0,
636 frequencies: Vec::new(),
637 },
638 generation: 1,
639 order: Vec::new(),
640 next_order: 0,
641 pending: Vec::with_capacity(STRIPE_PARTS),
642 })
643 }
644
645 fn put(&mut self, bytes: &[u8]) -> Result<()> {
650 write_at(&self.file, self.at, bytes)?;
651 self.at = self
652 .at
653 .checked_add(bytes.len() as u64)
654 .ok_or_else(|| invalid("native file length overflow"))?;
655 Ok(())
656 }
657
658 pub fn append(&mut self, chunk: &Chunk) -> Result<()> {
664 let order = (self.next_order, 0);
665 self.next_order = self.next_order.saturating_add(1);
666 self.append_at(order, chunk)
667 }
668
669 pub fn append_at(&mut self, order: (u64, u64), chunk: &Chunk) -> Result<()> {
680 if chunk.is_empty() {
681 return Ok(());
682 }
683 self.admit(chunk)?;
684 if self.pending.last().is_some_and(|last| last.order > order) {
685 self.flush_pending()?;
686 }
687 self.pending.push(PendingChunk { order, chunk: chunk.clone() });
692 if self.pending.len() == STRIPE_PARTS {
693 self.flush_pending()?;
694 }
695 Ok(())
696 }
697
698 pub fn append_stripe(&mut self, parts: Vec<((u64, u64), Chunk)>) -> Result<()> {
714 if parts.len() > STRIPE_PARTS {
715 return Err(invalid("a stripe was handed more parts than it holds"));
716 }
717 self.flush_pending()?;
720 for (order, chunk) in parts {
721 if chunk.is_empty() {
722 continue;
723 }
724 self.admit(&chunk)?;
725 self.pending.push(PendingChunk { order, chunk });
726 }
727 self.flush_pending()
728 }
729
730 fn admit(&mut self, chunk: &Chunk) -> Result<()> {
732 if chunk.width() != self.table.fields.len() {
733 return Err(invalid("chunk width differs from table schema"));
734 }
735 for (index, field) in self.table.fields.iter().enumerate() {
736 if chunk.column(index)?.logical_type() != &field.ty {
737 return Err(invalid("chunk type differs from table schema"));
738 }
739 }
740 self.table.rows = self
741 .table
742 .rows
743 .checked_add(chunk.len())
744 .ok_or_else(|| invalid("row count overflow"))?;
745 Ok(())
746 }
747
748 fn encode_column(
756 index: usize,
757 held: &[PendingChunk],
758 mut dictionary: Option<&mut GlobalDictionary>,
759 ) -> Result<ColumnStripe> {
760 let mut stripe = ColumnStripe {
761 pages: Vec::with_capacity(held.len()),
762 codes: Vec::with_capacity(held.len()),
763 sieves: Vec::with_capacity(held.len()),
764 ranges: Vec::with_capacity(held.len()),
765 };
766 for pending in held {
767 let column = pending.chunk.column(index)?;
768 let (bytes, unique) = encode(column, dictionary.as_deref_mut())?;
769 if bytes.len() > MAX_PAGE {
770 return Err(invalid("column page exceeds the configured bound"));
771 }
772 let range = Range::of(column);
775 let sieve = match dictionary {
788 Some(_) => None,
789 None => Sieve::of(column, &range, SIEVE_BUDGET)
790 .filter(|sieve| sieve.len() < bytes.len()),
791 };
792 stripe.pages.push(bytes);
793 stripe.codes.push(unique);
794 stripe.sieves.push(sieve);
795 stripe.ranges.push(range);
796 }
797 Ok(stripe)
798 }
799
800 fn encode_columns(&mut self, held: &[PendingChunk]) -> Result<Vec<ColumnStripe>> {
809 let width = self.table.fields.len();
810 let workers = std::thread::available_parallelism()
811 .map_or(1, usize::from)
812 .min(MAX_ENCODE_WORKERS)
813 .min(width);
814 if workers <= 1 || held.len() <= 1 {
815 return self
816 .dictionaries
817 .iter_mut()
818 .enumerate()
819 .map(|(index, dictionary)| Self::encode_column(index, held, dictionary.as_mut()))
820 .collect();
821 }
822 let mut jobs: Vec<(usize, Option<GlobalDictionary>)> =
825 std::mem::take(&mut self.dictionaries).into_iter().enumerate().collect();
826 jobs.sort_by_key(|(index, _)| weight(&self.table.fields[*index].ty));
828 let queue = Mutex::new(jobs);
829 let pieces = std::thread::scope(|scope| {
830 (0..workers)
831 .map(|_| {
832 scope.spawn(|| {
833 let mut mine = Vec::new();
834 loop {
835 let taken = queue
836 .lock()
837 .map_err(|_| Error::internal("a native encode worker panicked"))?
838 .pop();
839 let Some((index, mut dictionary)) = taken else { break };
840 let encoded = Self::encode_column(index, held, dictionary.as_mut())?;
841 mine.push((index, dictionary, encoded));
842 }
843 Ok(mine)
844 })
845 })
846 .collect::<Vec<_>>()
847 .into_iter()
848 .map(|handle| {
849 handle.join().map_err(|_| Error::internal("a native encode worker panicked"))?
850 })
851 .collect::<Result<Vec<_>>>()
852 })?;
853 let mut dictionaries: Vec<Option<GlobalDictionary>> = (0..width).map(|_| None).collect();
854 let mut encoded: Vec<Option<ColumnStripe>> = (0..width).map(|_| None).collect();
855 for piece in pieces {
856 for (index, dictionary, stripe) in piece {
857 dictionaries[index] = dictionary;
858 encoded[index] = Some(stripe);
859 }
860 }
861 self.dictionaries = dictionaries;
862 encoded
863 .into_iter()
864 .map(|stripe| stripe.ok_or_else(|| Error::internal("a column was never encoded")))
865 .collect()
866 }
867
868 fn flush_pending(&mut self) -> Result<()> {
870 if self.pending.is_empty() {
871 return Ok(());
872 }
873 let width = self.table.fields.len();
874 let mut held = std::mem::take(&mut self.pending);
877 let parts = held.len();
878 let encoded = self.encode_columns(&held)?;
879 let mut pages = Vec::with_capacity(width);
880 let mut memberships = vec![None; width];
881 let mut ranges = Vec::with_capacity(width);
882 let mut index = Vec::with_capacity(width.saturating_mul(index_section(parts)?));
883 for stripe in &encoded {
884 let offset = self.at;
885 let section = index.len();
886 let mut length = 0_usize;
887 for bytes in &stripe.pages {
888 write_at(&self.file, self.at + length as u64, bytes)?;
889 put_u32(
890 &mut index,
891 u32::try_from(bytes.len()).map_err(|_| invalid("part length overflow"))?,
892 );
893 put_u64(&mut index, checksum(bytes));
894 length = length
895 .checked_add(bytes.len())
896 .ok_or_else(|| invalid("column page length overflow"))?;
897 }
898 let hash = checksum(&index[section..]);
899 put_u64(&mut index, hash);
900 if length > MAX_PAGE {
901 return Err(invalid("column page exceeds the configured bound"));
902 }
903 self.at = self
904 .at
905 .checked_add(length as u64)
906 .ok_or_else(|| invalid("native file length overflow"))?;
907 pages.push(Span {
908 offset,
909 length: u32::try_from(length).map_err(|_| invalid("page length overflow"))?,
910 });
911 ranges.push(merged_range(stripe.ranges.iter().cloned()));
912 }
913 for (membership, stripe) in memberships.iter_mut().zip(&encoded) {
914 if stripe.codes.iter().all(Option::is_none) {
915 continue;
916 }
917 let lists = stripe
918 .codes
919 .iter()
920 .map(|codes| codes.clone().unwrap_or_default())
921 .collect::<Vec<_>>();
922 let bytes = encode_membership(&merged_codes(lists));
923 let offset = self.at;
924 self.put(&bytes)?;
925 *membership = Some(Page {
926 offset,
927 length: u32::try_from(bytes.len())
928 .map_err(|_| invalid("membership page length overflow"))?,
929 hash: checksum(&bytes),
930 });
931 }
932 let mut sieves = vec![None; width];
933 for (page, stripe) in sieves.iter_mut().zip(&encoded) {
934 if stripe.sieves.iter().all(Option::is_none) {
935 continue;
936 }
937 let bytes = encode_sieves(stripe.sieves.iter())?;
938 let offset = self.at;
939 self.put(&bytes)?;
940 *page = Some(Page {
941 offset,
942 length: u32::try_from(bytes.len())
943 .map_err(|_| invalid("sieve page length overflow"))?,
944 hash: checksum(&bytes),
945 });
946 }
947 let offset = self.at;
948 self.put(&index)?;
949 let index = Span {
950 offset,
951 length: u32::try_from(index.len())
952 .map_err(|_| invalid("index page length overflow"))?,
953 };
954 let mut rows = 0_usize;
955 let mut lengths = Vec::with_capacity(parts);
956 let mut span = None;
957 for pending in held.drain(..) {
958 let part = pending.chunk.len();
959 rows = rows.checked_add(part).ok_or_else(|| invalid("row count overflow"))?;
960 lengths.push(u32::try_from(part).map_err(|_| invalid("part row count overflow"))?);
961 span = Some(
962 span.map_or((pending.order, pending.order), |(first, _)| (first, pending.order)),
963 );
964 }
965 self.order.push(span.ok_or_else(|| invalid("a stripe was flushed with no parts"))?);
966 self.table.stripes.push(Stripe {
967 rows,
968 parts: lengths,
969 index,
970 pages,
971 memberships,
972 sieves,
973 zone: Zone::from_ranges(ranges),
974 });
975 self.pending = held;
977 Ok(())
978 }
979
980 fn numeric_frequency(&self, column: usize) -> Result<Option<FrequencySummary>> {
984 let ty = &self.table.fields[column].ty;
985 if !matches!(
986 ty,
987 LogicalType::TinyInt
988 | LogicalType::SmallInt
989 | LogicalType::Integer
990 | LogicalType::BigInt
991 | LogicalType::UTinyInt
992 | LogicalType::USmallInt
993 | LogicalType::UInteger
994 | LogicalType::UBigInt
995 | LogicalType::Date
996 | LogicalType::Timestamp
997 ) {
998 return Ok(None);
999 }
1000 let mut candidates: HashMap<FrequencyValue, u32> = HashMap::new();
1001 let mut decrements = 0_u64;
1002 self.visit_numeric(column, |_, value| {
1003 if let Some(count) = candidates.get_mut(&value) {
1004 *count = count.saturating_add(1);
1005 } else if candidates.len() < FREQUENCY_CANDIDATES {
1006 candidates.insert(value, 1);
1007 } else {
1008 candidates.retain(|_, count| {
1009 *count -= 1;
1010 *count != 0
1011 });
1012 decrements = decrements.saturating_add(1);
1013 }
1014 })?;
1015 let (exact, ordinals) = if decrements == 0 {
1016 (
1017 candidates
1018 .into_iter()
1019 .map(|(value, count)| (value, u64::from(count)))
1020 .collect::<HashMap<_, _>>(),
1021 Vec::new(),
1022 )
1023 } else {
1024 let mut lower = candidates.values().copied().collect::<Vec<_>>();
1025 lower.sort_unstable_by(|left, right| right.cmp(left));
1026 if lower.len() < FREQUENCY_BUILD_RANK
1027 || u64::from(lower[FREQUENCY_BUILD_RANK - 1]) <= decrements
1028 {
1029 return Ok(None);
1030 }
1031 let mut exact =
1032 candidates.into_keys().map(|value| (value, 0_u64)).collect::<HashMap<_, _>>();
1033 let mut ordinals = Vec::new();
1034 let mut exceeded = false;
1035 self.visit_numeric(column, |ordinal, value| {
1036 if let Some(count) = exact.get_mut(&value) {
1037 *count = count.saturating_add(1);
1038 if !exceeded {
1039 if ordinals.len() < FREQUENCY_ORDINALS {
1040 ordinals.push(ordinal);
1041 } else {
1042 ordinals.clear();
1043 exceeded = true;
1044 }
1045 }
1046 }
1047 })?;
1048 (exact, ordinals)
1049 };
1050 let mut entries = exact
1051 .into_iter()
1052 .map(|(value, count)| FrequencyEntry { value, count })
1053 .collect::<Vec<_>>();
1054 entries.sort_unstable_by(|left, right| {
1055 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
1056 });
1057 let omitted_max =
1058 entries.get(FREQUENCY_ENTRIES).map_or(decrements, |entry| decrements.max(entry.count));
1059 entries.truncate(FREQUENCY_ENTRIES);
1060 Ok(Some(FrequencySummary { entries, omitted_max, ordinals }))
1061 }
1062
1063 fn visit_numeric(
1064 &self,
1065 column: usize,
1066 mut visit: impl FnMut(u64, FrequencyValue),
1067 ) -> Result<()> {
1068 let ty = &self.table.fields[column].ty;
1069 let mut start = 0_u64;
1070 for stripe in &self.table.stripes {
1071 let spans = read_index(&self.file, stripe, column)?;
1072 let page = stripe.pages[column];
1073 let mut bytes = vec![0; page.length as usize];
1074 read_at(&self.file, page.offset, &mut bytes)?;
1075 for (span, &rows) in spans.iter().zip(&stripe.parts) {
1076 let part = part_bytes(&bytes, *span)?;
1077 if checksum(part) != span.hash {
1078 return Err(invalid("column page checksum differs while building frequencies"));
1079 }
1080 let rows = rows as usize;
1081 let vector = decode(ty, rows, part, None)?;
1082 for row in 0..rows {
1084 let value = if vector.is_null_at(row) {
1085 FrequencyValue::Null
1086 } else {
1087 let widened = match vector.signed_at(row) {
1091 Some(value) => Some(value),
1092 None => match vector.value_at(row) {
1093 Value::UTinyInt(value) => Some(i128::from(value)),
1094 Value::USmallInt(value) => Some(i128::from(value)),
1095 Value::UInteger(value) => Some(i128::from(value)),
1096 Value::UBigInt(value) => Some(i128::from(value)),
1097 _ => None,
1098 },
1099 };
1100 FrequencyValue::Integer(widened.ok_or_else(|| {
1101 invalid("numeric frequency page did not contain an integer value")
1102 })?)
1103 };
1104 visit(start.saturating_add(row as u64), value);
1105 }
1106 start = start.saturating_add(rows as u64);
1107 }
1108 }
1109 Ok(())
1110 }
1111
1112 fn numeric_frequencies(&self) -> Result<Vec<Option<FrequencySummary>>> {
1120 let mut columns = self
1121 .table
1122 .fields
1123 .iter()
1124 .enumerate()
1125 .filter_map(|(column, field)| {
1126 matches!(
1127 field.ty,
1128 LogicalType::TinyInt
1129 | LogicalType::SmallInt
1130 | LogicalType::Integer
1131 | LogicalType::BigInt
1132 | LogicalType::UTinyInt
1133 | LogicalType::USmallInt
1134 | LogicalType::UInteger
1135 | LogicalType::UBigInt
1136 | LogicalType::Date
1137 | LogicalType::Timestamp
1138 )
1139 .then_some(column)
1140 })
1141 .collect::<Vec<_>>();
1142 let workers = std::thread::available_parallelism()
1143 .map_or(1, usize::from)
1144 .min(MAX_FREQUENCY_WORKERS)
1145 .min(columns.len());
1146 if workers <= 1 {
1147 let mut frequencies = vec![None; self.table.fields.len()];
1148 for column in columns {
1149 frequencies[column] = self.numeric_frequency(column)?;
1150 }
1151 return Ok(frequencies);
1152 }
1153 columns.sort_by_key(|&column| weight(&self.table.fields[column].ty));
1156 let queue = Mutex::new(columns);
1157 let pieces = std::thread::scope(|scope| {
1158 (0..workers)
1159 .map(|_| {
1160 scope.spawn(|| {
1161 let mut mine = Vec::new();
1162 loop {
1163 let taken = queue
1164 .lock()
1165 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1166 .pop();
1167 let Some(column) = taken else { break };
1168 mine.push((column, self.numeric_frequency(column)?));
1169 }
1170 Ok(mine)
1171 })
1172 })
1173 .collect::<Vec<_>>()
1174 .into_iter()
1175 .map(|handle| {
1176 handle
1177 .join()
1178 .map_err(|_| Error::internal("a native frequency worker panicked"))?
1179 })
1180 .collect::<Result<Vec<_>>>()
1181 })?;
1182 let mut frequencies = vec![None; self.table.fields.len()];
1183 for piece in pieces {
1184 for (column, summary) in piece {
1185 frequencies[column] = summary;
1186 }
1187 }
1188 Ok(frequencies)
1189 }
1190
1191 pub fn finish(mut self) -> Result<Table> {
1197 self.flush_pending()?;
1198 let mut stripes = std::mem::take(&mut self.order)
1199 .into_iter()
1200 .zip(std::mem::take(&mut self.table.stripes))
1201 .collect::<Vec<_>>();
1202 stripes.sort_by_key(|(order, _)| order.0);
1203 let mut previous: Option<(u64, u64)> = None;
1204 for ((first, last), _) in &stripes {
1205 if previous.is_some_and(|previous| previous >= *first) {
1206 return Err(invalid("chunks did not arrive in source order"));
1207 }
1208 previous = Some(*last);
1209 }
1210 self.table.stripes = stripes.into_iter().map(|(_, stripe)| stripe).collect();
1211 self.table.frequencies = self.numeric_frequencies()?;
1212 let dictionaries = std::mem::take(&mut self.dictionaries);
1213 let orders = rankings(&dictionaries)?;
1214 for (index, (dictionary, order)) in dictionaries.into_iter().zip(orders).enumerate() {
1215 let Some(dictionary) = dictionary else { continue };
1216 self.table.frequencies[index] = Some(code_frequency(&dictionary));
1217 let encoded = encode_global_dictionary(dictionary, &order)?;
1218 let offset = self.at;
1219 self.put(&encoded.index)?;
1220 self.put(&encoded.ranks)?;
1221 for block in &encoded.payload {
1222 self.put(block)?;
1223 }
1224 let payload_len =
1225 encoded.payload.iter().try_fold(0_usize, |len, block| len.checked_add(block.len()));
1226 let length = payload_len
1227 .and_then(|len| len.checked_add(encoded.index.len()))
1228 .and_then(|len| len.checked_add(encoded.ranks.len()))
1229 .ok_or_else(|| invalid("dictionary page length overflow"))?;
1230 self.table.dictionaries[index] = Some(Page {
1231 offset,
1232 length: u32::try_from(length)
1233 .map_err(|_| invalid("dictionary page length overflow"))?,
1234 hash: checksum(&encoded.index),
1235 });
1236 }
1237 let directory = encode_directory(&self.table)?;
1238 if directory.len() > MAX_DIRECTORY {
1239 return Err(invalid("directory exceeds the configured bound"));
1240 }
1241 let offset = self.at;
1242 self.put(&directory)?;
1243 self.file.sync_all().map_err(io)?;
1244 let slot = Slot {
1245 offset,
1246 length: u32::try_from(directory.len())
1247 .map_err(|_| invalid("directory length overflow"))?,
1248 generation: self.generation,
1249 hash: checksum(&directory),
1250 };
1251 write_at(&self.file, 16, &slot.bytes())?;
1254 self.file.sync_all().map_err(io)?;
1255 Ok(self.table)
1256 }
1257}
1258
1259#[derive(Debug, Clone)]
1261pub struct Reader {
1262 file: Arc<File>,
1263 table: Arc<Table>,
1264 dictionaries: Arc<Vec<OnceLock<Arc<Vector>>>>,
1265 loading: Arc<Vec<Mutex<()>>>,
1274 opened: Arc<AtomicUsize>,
1278 sieves: Arc<Vec<Vec<SieveSlot>>>,
1282 places: Arc<Vec<Place>>,
1284 cache: Arc<Vec<Mutex<Cached>>>,
1285 pages: Arc<AtomicUsize>,
1288 indexes: Arc<AtomicUsize>,
1291 kept: Arc<AtomicUsize>,
1294 size: u64,
1296 directory: u64,
1298 opening: Opening,
1300}
1301
1302#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1314pub struct Opening {
1315 pub reads: u32,
1318 pub bytes: u64,
1320}
1321
1322#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
1324pub struct Reads {
1325 pub opening: Opening,
1327 pub pages: usize,
1329 pub indexes: usize,
1331 pub dictionaries: usize,
1334}
1335
1336#[derive(Debug, Clone, Copy)]
1338struct Place {
1339 stripe: u32,
1340 part: u32,
1341 rows: u32,
1342}
1343
1344#[derive(Debug, Clone, Copy)]
1346struct PartSpan {
1347 start: usize,
1348 length: usize,
1349 hash: u64,
1350}
1351
1352#[derive(Debug, Clone)]
1358struct CachedColumn {
1359 stripe: usize,
1360 index: Arc<Vec<PartSpan>>,
1361 page: Option<Arc<Vec<u8>>>,
1362}
1363
1364#[derive(Debug, Default)]
1384struct Cached {
1385 pages: Vec<Option<Arc<Vec<u8>>>>,
1386 order: VecDeque<usize>,
1387 loading: Vec<usize>,
1388 index: Vec<Option<Arc<Vec<PartSpan>>>>,
1389}
1390
1391const CACHED_STRIPES_PER_COLUMN: usize = 4;
1403
1404type SieveSlot = OnceLock<Arc<Vec<Option<Sieve>>>>;
1406
1407#[derive(Debug)]
1408struct NativeText {
1409 file: Arc<File>,
1410 offsets: Vec<u32>,
1411 ranks: usize,
1413 rank_at: u64,
1417 rank_hashes: Vec<u64>,
1418 rank_blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1419 code_ranks: OnceLock<Option<Vec<u32>>>,
1426 payload: u64,
1427 ends: Vec<u64>,
1430 hashes: Vec<u64>,
1431 blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1433}
1434
1435const TEXT_PAYLOAD_VALUES: usize = 1024;
1451
1452const TEXT_RANK_BLOCK: usize = 512;
1462
1463const RANK_ENTRY: usize = size_of::<u64>() + size_of::<u32>();
1465
1466impl NativeText {
1467 fn payload_block(&self, block: usize) -> Result<Option<&[u8]>> {
1474 let Some(slot) = self.blocks.get(block) else { return Ok(None) };
1475 let bytes = slot
1476 .get_or_init(|| {
1477 let start = if block == 0 { 0 } else { self.ends[block - 1] };
1478 let end = self.ends[block];
1479 let len = end
1480 .checked_sub(start)
1481 .ok_or_else(|| invalid("global dictionary block ends before it starts"))?;
1482 let mut stored = vec![
1483 0;
1484 usize::try_from(len).map_err(|_| invalid(
1485 "global dictionary block does not fit in memory"
1486 ))?
1487 ];
1488 read_at(&self.file, self.payload + start, &mut stored)?;
1489 if checksum(&stored) != self.hashes[block] {
1490 return Err(invalid("global dictionary payload checksum differs"));
1491 }
1492 let first = block * TEXT_PAYLOAD_VALUES;
1493 let last = (first + TEXT_PAYLOAD_VALUES).min(self.offsets.len() - 1);
1494 let want = (self.offsets[last] - self.offsets[first]) as usize;
1495 let values = string::decode_flat(&stored)?;
1496 if values.len() != last - first {
1497 return Err(invalid("global dictionary block holds the wrong value count"));
1498 }
1499 let bytes = values.into_bytes();
1500 if bytes.len() != want {
1501 return Err(invalid("global dictionary block decodes to the wrong length"));
1502 }
1503 Ok(bytes)
1504 })
1505 .as_ref()
1506 .map_err(Clone::clone)?;
1507 Ok(Some(bytes.as_slice()))
1508 }
1509
1510 fn rank_parts(&self, rank: usize) -> Result<(&[u8], usize)> {
1517 let slot = self
1518 .rank_blocks
1519 .get(rank / TEXT_RANK_BLOCK)
1520 .ok_or_else(|| invalid("global dictionary rank is past the order"))?;
1521 let block = slot
1522 .get_or_init(|| {
1523 let first = rank / TEXT_RANK_BLOCK * TEXT_RANK_BLOCK;
1524 let len = TEXT_RANK_BLOCK.min(self.ranks - first) * RANK_ENTRY;
1525 let mut bytes = vec![0; len];
1526 read_at(&self.file, self.rank_at + (first * RANK_ENTRY) as u64, &mut bytes)?;
1527 if checksum(&bytes)
1528 != *self
1529 .rank_hashes
1530 .get(rank / TEXT_RANK_BLOCK)
1531 .ok_or_else(|| invalid("global dictionary rank block has no checksum"))?
1532 {
1533 return Err(invalid("global dictionary rank checksum differs"));
1534 }
1535 Ok(bytes)
1536 })
1537 .as_ref()
1538 .map_err(Clone::clone)?;
1539 Ok((block.as_slice(), rank % TEXT_RANK_BLOCK))
1540 }
1541
1542 fn head_at(&self, rank: usize) -> Result<u64> {
1544 let (block, within) = self.rank_parts(rank)?;
1545 let at = within * size_of::<u64>();
1546 let bytes = block
1547 .get(at..at + size_of::<u64>())
1548 .ok_or_else(|| invalid("global dictionary rank block is short of heads"))?;
1549 Ok(u64::from_le_bytes(bytes.try_into().expect("eight bytes")))
1550 }
1551}
1552
1553impl TextSource for NativeText {
1554 fn len(&self) -> usize {
1555 self.offsets.len().saturating_sub(1)
1556 }
1557
1558 fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
1559 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1560 else {
1561 return Ok(None);
1562 };
1563 if start == end {
1564 return Ok(Some(&[]));
1565 }
1566 let block = index / TEXT_PAYLOAD_VALUES;
1569 let Some(bytes) = self.payload_block(block)? else { return Ok(None) };
1570 let base = self.offsets[block * TEXT_PAYLOAD_VALUES];
1571 let within = (start - base) as usize;
1572 Ok(bytes.get(within..within + (end - start) as usize))
1573 }
1574
1575 fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
1576 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1577 else {
1578 return Ok(None);
1579 };
1580 Ok(Some((end - start) as usize))
1581 }
1582
1583 fn ranks(&self) -> Option<usize> {
1584 (self.ranks > 0).then_some(self.ranks)
1585 }
1586
1587 fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
1588 let settled = self.head_at(rank)?.cmp(&head(wanted));
1592 if settled != Ordering::Equal {
1593 return Ok(settled);
1594 }
1595 let code = self.code_at_rank(rank)?;
1596 let bytes = self
1597 .bytes_at(code as usize)?
1598 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
1599 Ok(bytes.cmp(wanted))
1600 }
1601
1602 fn code_at_rank(&self, rank: usize) -> Result<u32> {
1603 let (block, within) = self.rank_parts(rank)?;
1604 let heads = block.len() / RANK_ENTRY * size_of::<u64>();
1605 let at = heads + within * size_of::<u32>();
1606 let bytes = block
1607 .get(at..at + size_of::<u32>())
1608 .ok_or_else(|| invalid("global dictionary rank block is short of codes"))?;
1609 let code = u32::from_le_bytes(bytes.try_into().expect("four bytes"));
1610 if code as usize >= self.len() {
1611 return Err(invalid("global dictionary order names a code it does not have"));
1612 }
1613 Ok(code)
1614 }
1615
1616 fn code_ranks(&self) -> Option<&[u32]> {
1617 if self.ranks == 0 || self.ranks != self.len() {
1621 return None;
1622 }
1623 self.code_ranks
1624 .get_or_init(|| {
1625 let mut ranks = vec![u32::MAX; self.ranks];
1626 for first in (0..self.ranks).step_by(TEXT_RANK_BLOCK) {
1629 let (block, _) = self.rank_parts(first).ok()?;
1630 let heads = block.len() / RANK_ENTRY * size_of::<u64>();
1631 let codes = block.get(heads..)?;
1632 for (within, entry) in codes.chunks_exact(size_of::<u32>()).enumerate() {
1633 let code = u32::from_le_bytes(entry.try_into().ok()?) as usize;
1634 *ranks.get_mut(code)? = u32::try_from(first + within).ok()?;
1635 }
1636 }
1637 if ranks.contains(&u32::MAX) {
1638 return None;
1639 }
1640 Some(ranks)
1641 })
1642 .as_deref()
1643 }
1644
1645 fn footprint(&self) -> usize {
1646 self.offsets.capacity() * size_of::<u32>()
1647 + self
1648 .code_ranks
1649 .get()
1650 .and_then(Option::as_ref)
1651 .map_or(0, |ranks| ranks.capacity() * size_of::<u32>())
1652 + self.rank_hashes.capacity() * size_of::<u64>()
1653 + self.rank_blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1654 + self
1655 .rank_blocks
1656 .iter()
1657 .filter_map(OnceLock::get)
1658 .filter_map(|result| result.as_ref().ok())
1659 .map(Vec::capacity)
1660 .sum::<usize>()
1661 + self.blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1662 + self.hashes.capacity() * size_of::<u64>()
1663 + self.ends.capacity() * size_of::<u64>()
1664 + self
1665 .blocks
1666 .iter()
1667 .filter_map(OnceLock::get)
1668 .filter_map(|result| result.as_ref().ok())
1669 .map(Vec::capacity)
1670 .sum::<usize>()
1671 }
1672}
1673
1674fn places(table: &Table) -> Result<Vec<Place>> {
1676 let mut places = Vec::with_capacity(table.stripes.len().saturating_mul(STRIPE_PARTS));
1677 for (at, stripe) in table.stripes.iter().enumerate() {
1678 let index = u32::try_from(at).map_err(|_| invalid("too many stripes"))?;
1679 for (part, &rows) in stripe.parts.iter().enumerate() {
1680 places.push(Place {
1681 stripe: index,
1682 part: u32::try_from(part).map_err(|_| invalid("too many parts in a stripe"))?,
1683 rows,
1684 });
1685 }
1686 }
1687 Ok(places)
1688}
1689
1690fn read_index(file: &File, stripe: &Stripe, column: usize) -> Result<Vec<PartSpan>> {
1695 let parts = stripe.parts.len();
1696 let section = index_section(parts)?;
1697 let at = column.checked_mul(section).ok_or_else(|| invalid("index page offset overflow"))?;
1698 let end = at.checked_add(section).ok_or_else(|| invalid("index page offset overflow"))?;
1699 if end > stripe.index.length as usize {
1700 return Err(invalid("index page is shorter than its columns"));
1701 }
1702 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1703 let mut bytes = vec![0; section];
1704 let offset = stripe
1705 .index
1706 .offset
1707 .checked_add(at as u64)
1708 .ok_or_else(|| invalid("index page offset overflow"))?;
1709 read_at(file, offset, &mut bytes)?;
1710 let entries = section - size_of::<u64>();
1711 let stored = u64::from_le_bytes(bytes[entries..].try_into().expect("eight bytes"));
1712 if checksum(&bytes[..entries]) != stored {
1713 return Err(invalid(&format!(
1716 "index page section checksum differs, column {column} of {parts} parts at {offset}, \
1717 wanted {stored:016x} and got {:016x}",
1718 checksum(&bytes[..entries]),
1719 )));
1720 }
1721 let mut spans = Vec::with_capacity(parts);
1722 let mut start = 0_usize;
1723 for part in 0..parts {
1724 let at = part * INDEX_ENTRY;
1725 let length = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four bytes")) as usize;
1726 let hash = u64::from_le_bytes(bytes[at + 4..at + 12].try_into().expect("eight bytes"));
1727 spans.push(PartSpan { start, length, hash });
1728 start = start.checked_add(length).ok_or_else(|| invalid("column page length overflow"))?;
1729 }
1730 if start != page.length as usize {
1731 return Err(invalid("column page length differs from its index"));
1732 }
1733 Ok(spans)
1734}
1735
1736fn part_bytes(page: &[u8], span: PartSpan) -> Result<&[u8]> {
1738 let end = span.start.checked_add(span.length).ok_or_else(|| invalid("part range overflow"))?;
1739 page.get(span.start..end).ok_or_else(|| invalid("part exceeds its column page"))
1740}
1741
1742fn remember(cached: &mut Cached, held: &CachedColumn, kept: usize) {
1747 if let Some(slot) = cached.index.get_mut(held.stripe) {
1748 if slot.is_none() {
1749 *slot = Some(Arc::clone(&held.index));
1750 }
1751 }
1752 let Some(page) = held.page.clone() else { return };
1753 let Some(slot) = cached.pages.get_mut(held.stripe) else { return };
1754 if slot.is_none() {
1755 cached.order.push_back(held.stripe);
1756 }
1757 *slot = Some(page);
1758 while cached.order.len() > kept.max(1) {
1759 let Some(oldest) = cached.order.pop_front() else { break };
1760 if let Some(slot) = cached.pages.get_mut(oldest) {
1761 *slot = None;
1762 }
1763 }
1764}
1765
1766impl Reader {
1767 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1773 let mut file = File::open(path).map_err(io)?;
1774 let size = file.metadata().map_err(io)?.len();
1775 if size < HEADER {
1776 return Err(invalid("file is shorter than its header"));
1777 }
1778 let mut header = [0; HEADER as usize];
1779 file.read_exact(&mut header).map_err(io)?;
1780 let mut opening = Opening { reads: 1, bytes: HEADER };
1781 let version = u32::from_le_bytes([header[8], header[9], header[10], header[11]]);
1782 if &header[..8] != MAGIC {
1787 return Err(invalid("the header does not begin with a rudb native magic"));
1788 }
1789 if version != FORMAT {
1790 return Err(invalid(&format!(
1791 "the file is format {version} and this build reads format {FORMAT}, so it has to \
1792 be written again"
1793 )));
1794 }
1795 let mut selected = None;
1796 for start in [16, 16 + SLOT_BYTES] {
1797 let slot = Slot::read(&header[start..start + SLOT_BYTES]);
1798 if slot.generation == 0 || slot.length == 0 || slot.length as usize > MAX_DIRECTORY {
1799 continue;
1800 }
1801 let Some(end) = slot.offset.checked_add(u64::from(slot.length)) else { continue };
1802 if slot.offset < HEADER || end > size {
1803 continue;
1804 }
1805 let mut bytes = vec![0; slot.length as usize];
1806 file.seek(SeekFrom::Start(slot.offset)).map_err(io)?;
1807 file.read_exact(&mut bytes).map_err(io)?;
1808 opening.reads += 1;
1809 opening.bytes += u64::from(slot.length);
1810 if checksum(&bytes) == slot.hash
1811 && selected
1812 .as_ref()
1813 .is_none_or(|(old, _): &(Slot, Vec<u8>)| old.generation < slot.generation)
1814 {
1815 selected = Some((slot, bytes));
1816 }
1817 }
1818 let (slot, bytes) =
1819 selected.ok_or_else(|| invalid("no committed directory slot is valid"))?;
1820 let table = decode_directory(&bytes, size)?;
1821 let places = places(&table)?;
1822 let dictionaries = (0..table.fields.len()).map(|_| OnceLock::new()).collect();
1823 let table_fields = table.fields.len();
1824 let stripes = table.stripes.len();
1825 let cache = (0..table.fields.len())
1826 .map(|_| {
1827 Mutex::new(Cached {
1828 pages: (0..stripes).map(|_| None).collect(),
1829 index: (0..stripes).map(|_| None).collect(),
1830 ..Cached::default()
1831 })
1832 })
1833 .collect::<Vec<_>>();
1834 let sieves = (0..table.fields.len())
1835 .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
1836 .collect();
1837 Ok(Self {
1838 file: Arc::new(file),
1839 table: Arc::new(table),
1840 dictionaries: Arc::new(dictionaries),
1841 loading: Arc::new((0..table_fields).map(|_| Mutex::new(())).collect()),
1842 opened: Arc::new(AtomicUsize::new(0)),
1843 sieves: Arc::new(sieves),
1844 places: Arc::new(places),
1845 cache: Arc::new(cache),
1846 pages: Arc::new(AtomicUsize::new(0)),
1847 indexes: Arc::new(AtomicUsize::new(0)),
1848 kept: Arc::new(AtomicUsize::new(CACHED_STRIPES_PER_COLUMN)),
1849 size,
1850 directory: u64::from(slot.length),
1851 opening,
1852 })
1853 }
1854
1855 #[must_use]
1862 pub fn reads(&self) -> Reads {
1863 Reads {
1864 opening: self.opening,
1865 pages: self.pages.load(Atomic::Relaxed),
1866 indexes: self.indexes.load(Atomic::Relaxed),
1867 dictionaries: self.opened.load(Atomic::Relaxed),
1868 }
1869 }
1870
1871 #[must_use]
1876 pub fn layout(&self) -> Layout {
1877 let table = &self.table;
1878 let stripes = table.stripes.as_slice();
1879 let columns = table
1880 .fields
1881 .iter()
1882 .enumerate()
1883 .map(|(at, field)| ColumnLayout {
1884 name: field.name.clone(),
1885 kind: field.ty.to_string(),
1886 pages: sum(stripes.iter().map(|stripe| span_bytes(&stripe.pages, at))),
1887 memberships: sum(stripes.iter().map(|stripe| page_bytes(&stripe.memberships, at))),
1888 sieves: sum(stripes.iter().map(|stripe| page_bytes(&stripe.sieves, at))),
1889 dictionary: page_bytes(&table.dictionaries, at),
1890 })
1891 .collect();
1892 Layout {
1893 file: self.size,
1894 rows: table.rows,
1895 stripes: stripes.len(),
1896 parts: self.places.len(),
1897 columns,
1898 indexes: sum(stripes.iter().map(|stripe| u64::from(stripe.index.length))),
1899 directory: self.directory,
1900 header: HEADER,
1901 }
1902 }
1903
1904 #[must_use]
1906 pub fn parts(&self) -> usize {
1907 self.places.len()
1908 }
1909
1910 #[must_use]
1917 pub fn stripe_parts(&self) -> Vec<std::ops::Range<usize>> {
1918 let mut runs = Vec::with_capacity(self.table.stripes.len());
1919 let mut start = 0;
1920 for stripe in &self.table.stripes {
1921 let end = start + stripe.parts.len();
1922 runs.push(start..end);
1923 start = end;
1924 }
1925 runs
1926 }
1927
1928 pub fn keep_stripes(&self, stripes: usize) {
1935 self.kept.fetch_max(stripes, Atomic::Relaxed);
1936 }
1937
1938 #[must_use]
1940 pub fn part_rows(&self, at: usize) -> usize {
1941 self.places.get(at).map_or(0, |place| place.rows as usize)
1942 }
1943
1944 #[must_use]
1946 pub fn table(&self) -> &Table {
1947 &self.table
1948 }
1949
1950 pub fn top_frequencies(&self, column: usize, top: usize) -> Result<Option<Vec<(Value, u64)>>> {
1959 let field = self
1960 .table
1961 .fields
1962 .get(column)
1963 .ok_or_else(|| invalid("frequency column index out of range"))?;
1964 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1965 return Ok(None);
1966 };
1967 if top == 0 || summary.entries.len() < top {
1968 return Ok(None);
1969 }
1970 let boundary = summary.entries[top - 1].count;
1971 if boundary <= summary.omitted_max {
1972 return Ok(None);
1973 }
1974 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
1975 }
1976
1977 pub fn exact_frequencies(&self, column: usize) -> Result<Option<Vec<(Value, u64)>>> {
1997 let field = self
1998 .table
1999 .fields
2000 .get(column)
2001 .ok_or_else(|| invalid("frequency column index out of range"))?;
2002 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2003 return Ok(None);
2004 };
2005 if summary.omitted_max > 0 {
2006 return Ok(None);
2007 }
2008 self.decode_frequencies(column, &field.ty, &summary.entries).map(Some)
2009 }
2010
2011 fn decode_frequencies(
2013 &self,
2014 column: usize,
2015 ty: &LogicalType,
2016 entries: &[FrequencyEntry],
2017 ) -> Result<Vec<(Value, u64)>> {
2018 let dictionary = if *ty == LogicalType::Varchar { self.dictionary(column)? } else { None };
2019 let mut out = Vec::with_capacity(entries.len());
2020 for entry in entries {
2021 let value = match entry.value {
2022 FrequencyValue::Null => Value::Null,
2023 FrequencyValue::Integer(value) => match *ty {
2024 LogicalType::TinyInt => Value::TinyInt(
2025 i8::try_from(value)
2026 .map_err(|_| invalid("frequency TINYINT is out of range"))?,
2027 ),
2028 LogicalType::UTinyInt => Value::UTinyInt(
2029 u8::try_from(value)
2030 .map_err(|_| invalid("frequency UTINYINT is out of range"))?,
2031 ),
2032 LogicalType::USmallInt => Value::USmallInt(
2033 u16::try_from(value)
2034 .map_err(|_| invalid("frequency USMALLINT is out of range"))?,
2035 ),
2036 LogicalType::UInteger => Value::UInteger(
2037 u32::try_from(value)
2038 .map_err(|_| invalid("frequency UINTEGER is out of range"))?,
2039 ),
2040 LogicalType::UBigInt => Value::UBigInt(
2041 u64::try_from(value)
2042 .map_err(|_| invalid("frequency UBIGINT is out of range"))?,
2043 ),
2044 LogicalType::SmallInt => Value::SmallInt(
2045 i16::try_from(value)
2046 .map_err(|_| invalid("frequency SMALLINT is out of range"))?,
2047 ),
2048 LogicalType::Integer => Value::Integer(
2049 i32::try_from(value)
2050 .map_err(|_| invalid("frequency INTEGER is out of range"))?,
2051 ),
2052 LogicalType::BigInt => Value::BigInt(
2053 i64::try_from(value)
2054 .map_err(|_| invalid("frequency BIGINT is out of range"))?,
2055 ),
2056 LogicalType::Date => Value::Date(
2057 i32::try_from(value)
2058 .map_err(|_| invalid("frequency DATE is out of range"))?,
2059 ),
2060 LogicalType::Timestamp => Value::Timestamp(
2061 i64::try_from(value)
2062 .map_err(|_| invalid("frequency TIMESTAMP is out of range"))?,
2063 ),
2064 _ => return Err(invalid("integer frequency belongs to another type")),
2065 },
2066 FrequencyValue::Code(code) => dictionary
2067 .as_ref()
2068 .ok_or_else(|| invalid("frequency code has no dictionary"))?
2069 .try_value_at(code as usize)?,
2070 };
2071 out.push((value, entry.count));
2072 }
2073 Ok(out)
2074 }
2075
2076 pub fn frequency_occurrences(&self, column: usize) -> Result<Option<FrequencyOccurrences>> {
2086 self.table
2087 .fields
2088 .get(column)
2089 .ok_or_else(|| invalid("frequency column index out of range"))?;
2090 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
2091 return Ok(None);
2092 };
2093 if summary.ordinals.is_empty() {
2094 return Ok(None);
2095 }
2096 Ok(Some(FrequencyOccurrences {
2097 omitted_max: summary.omitted_max,
2098 ordinals: summary.ordinals.clone(),
2099 }))
2100 }
2101
2102 pub fn distinct_values(&self, column: usize) -> Result<Option<u64>> {
2122 if self.null_count(column)? > 0 {
2123 return Ok(None);
2124 }
2125 Ok(self.dictionary(column)?.map(|dictionary| dictionary.len() as u64))
2126 }
2127
2128 pub fn null_count(&self, column: usize) -> Result<u64> {
2139 if column >= self.table.fields.len() {
2140 return Err(invalid("null count column index out of range"));
2141 }
2142 let mut nulls = 0_u64;
2143 for stripe in &self.table.stripes {
2144 let range = stripe
2145 .zone
2146 .column(column)
2147 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2148 nulls = nulls
2149 .checked_add(range.nulls as u64)
2150 .ok_or_else(|| invalid("null count overflow"))?;
2151 }
2152 Ok(nulls)
2153 }
2154
2155 pub fn text_extremes(&self, column: usize) -> Result<Option<(Value, Value)>> {
2170 if self.null_count(column)? > 0 {
2171 return Ok(None);
2172 }
2173 let Some(dictionary) = self.dictionary(column)? else { return Ok(None) };
2174 let Some(ranks) = dictionary.ranks() else { return Ok(None) };
2175 if ranks == 0 {
2176 return Ok(None);
2177 }
2178 let low = text_at_rank(&dictionary, 0)?;
2179 let high = text_at_rank(&dictionary, ranks - 1)?;
2180 Ok(Some((low, high)))
2181 }
2182
2183 pub fn exact_extremes(&self, column: usize) -> Result<Option<(Bound, Bound)>> {
2206 if column >= self.table.fields.len() {
2207 return Err(invalid("extremes column index out of range"));
2208 }
2209 let mut low: Option<Bound> = None;
2210 let mut high: Option<Bound> = None;
2211 for stripe in &self.table.stripes {
2212 let range = stripe
2213 .zone
2214 .column(column)
2215 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2216 if !range.exact {
2217 return Ok(None);
2218 }
2219 let (Some(small), Some(large)) = (range.low.as_ref(), range.high.as_ref()) else {
2224 if stripe.rows > range.nulls {
2225 return Ok(None);
2226 }
2227 continue;
2228 };
2229 low = Some(low.map_or_else(|| small.clone(), |held| held.smaller(small.clone())));
2230 high = Some(high.map_or_else(|| large.clone(), |held| held.larger(large.clone())));
2231 }
2232 Ok(low.zip(high))
2233 }
2234
2235 pub fn exact_sum(&self, column: usize) -> Result<Option<(i128, u64)>> {
2248 if column >= self.table.fields.len() {
2249 return Err(invalid("sum column index out of range"));
2250 }
2251 let mut total = 0_i128;
2252 let mut rows = 0_u64;
2253 for stripe in &self.table.stripes {
2254 let range = stripe
2255 .zone
2256 .column(column)
2257 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
2258 let Some(part) = range.sum else { return Ok(None) };
2259 let Some(sum) = total.checked_add(part) else { return Ok(None) };
2260 total = sum;
2261 rows = rows.saturating_add(stripe.rows as u64 - range.nulls as u64);
2262 }
2263 Ok(Some((total, rows)))
2264 }
2265
2266 fn dictionary(&self, column: usize) -> Result<Option<Arc<Vector>>> {
2275 let Some(page) = self.table.dictionaries[column] else { return Ok(None) };
2276 if let Some(dictionary) = self.dictionaries[column].get() {
2277 return Ok(Some(Arc::clone(dictionary)));
2278 }
2279 let _queued = self.loading[column].lock().map_err(|_| invalid("a poisoned dictionary"))?;
2280 if let Some(dictionary) = self.dictionaries[column].get() {
2281 return Ok(Some(Arc::clone(dictionary)));
2282 }
2283 self.opened.fetch_add(1, Atomic::Relaxed);
2284 let dictionary = Arc::new(open_global_dictionary(
2285 Arc::clone(&self.file),
2286 page,
2287 &self.table.fields[column].ty,
2288 )?);
2289 let _ = self.dictionaries[column].set(Arc::clone(&dictionary));
2290 Ok(Some(dictionary))
2291 }
2292
2293 pub fn read(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2302 self.read_impl(part, columns, true)
2303 }
2304
2305 pub fn read_sparse(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
2315 self.read_impl(part, columns, false)
2316 }
2317
2318 pub fn skips_codes(&self, part: usize, column: usize, candidates: &[u32]) -> Result<bool> {
2325 if candidates.is_empty() {
2326 return Ok(true);
2327 }
2328 if candidates.windows(2).any(|pair| pair[0] >= pair[1]) {
2329 return Err(Error::internal("native code candidates are not sorted and unique"));
2330 }
2331 let stripe = self.stripe_of(part)?;
2332 let Some(page) = stripe.memberships.get(column).copied().flatten() else {
2333 return Ok(false);
2334 };
2335 let mut bytes = vec![0; page.length as usize];
2336 read_at(&self.file, page.offset, &mut bytes)?;
2337 if checksum(&bytes) != page.hash {
2338 return Err(invalid("membership page checksum differs"));
2339 }
2340 let codes = decode_membership(&bytes)?;
2341 let mut left = 0;
2342 let mut right = 0;
2343 while left < codes.len() && right < candidates.len() {
2344 match codes[left].cmp(&candidates[right]) {
2345 Ordering::Less => left += 1,
2346 Ordering::Greater => right += 1,
2347 Ordering::Equal => return Ok(false),
2348 }
2349 }
2350 Ok(true)
2351 }
2352
2353 fn stripe_of(&self, part: usize) -> Result<&Stripe> {
2354 let place = self.places.get(part).ok_or_else(|| invalid("part index out of range"))?;
2355 self.table
2356 .stripes
2357 .get(place.stripe as usize)
2358 .ok_or_else(|| invalid("stripe index out of range"))
2359 }
2360
2361 fn held(&self, at: usize, stripe: &Stripe, column: usize, whole: bool) -> Result<CachedColumn> {
2378 let cache = self.cache.get(column).ok_or_else(|| invalid("column index out of range"))?;
2379 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2380 let known = cached.index.get(at).and_then(Clone::clone);
2381 let page = cached.pages.get(at).and_then(Clone::clone);
2382 if let Some(index) = known.clone() {
2383 if !whole || page.is_some() {
2384 return Ok(CachedColumn { stripe: at, index, page });
2385 }
2386 }
2387 if cached.loading.contains(&at) {
2388 drop(cached);
2389 if let Some(index) = known {
2393 return Ok(CachedColumn { stripe: at, index, page: None });
2394 }
2395 let held = self.page_of(stripe, column, at, false, None)?;
2396 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2397 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2398 return Ok(held);
2399 }
2400 cached.loading.push(at);
2401 drop(cached);
2402
2403 let read = self.page_of(stripe, column, at, whole, known);
2404
2405 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
2409 if let Some(position) = cached.loading.iter().position(|loading| *loading == at) {
2410 cached.loading.remove(position);
2411 }
2412 let held = read?;
2413 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
2414 Ok(held)
2415 }
2416
2417 fn page_of(
2423 &self,
2424 stripe: &Stripe,
2425 column: usize,
2426 at: usize,
2427 whole: bool,
2428 known: Option<Arc<Vec<PartSpan>>>,
2429 ) -> Result<CachedColumn> {
2430 let index = match known {
2431 Some(index) => index,
2432 None => {
2433 self.indexes.fetch_add(1, Atomic::Relaxed);
2434 Arc::new(read_index(&self.file, stripe, column)?)
2435 }
2436 };
2437 let page = if whole {
2438 self.pages.fetch_add(1, Atomic::Relaxed);
2439 let span = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2440 let mut bytes = vec![0; span.length as usize];
2441 read_at(&self.file, span.offset, &mut bytes)?;
2442 Some(Arc::new(bytes))
2443 } else {
2444 None
2445 };
2446 Ok(CachedColumn { stripe: at, index, page })
2447 }
2448
2449 fn read_impl(&self, at: usize, columns: &[usize], whole: bool) -> Result<Chunk> {
2450 let place = *self.places.get(at).ok_or_else(|| invalid("part index out of range"))?;
2451 let index = place.stripe as usize;
2452 let stripe =
2453 self.table.stripes.get(index).ok_or_else(|| invalid("stripe index out of range"))?;
2454 let rows = place.rows as usize;
2455 let mut picked = Vec::with_capacity(columns.len());
2456 for &column in columns {
2457 let field = self
2458 .table
2459 .fields
2460 .get(column)
2461 .ok_or_else(|| invalid("column index out of range"))?;
2462 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
2463 let held = self.held(index, stripe, column, whole)?;
2464 let span = *held
2465 .index
2466 .get(place.part as usize)
2467 .ok_or_else(|| invalid("part index out of range"))?;
2468 let owned;
2469 let bytes = match &held.page {
2470 Some(held) => part_bytes(held, span)?,
2471 None => {
2472 let offset = page
2473 .offset
2474 .checked_add(span.start as u64)
2475 .ok_or_else(|| invalid("part range overflow"))?;
2476 let mut bytes = vec![0; span.length];
2477 read_at(&self.file, offset, &mut bytes)?;
2478 owned = bytes;
2479 &owned
2480 }
2481 };
2482 if checksum(bytes) != span.hash {
2483 return Err(invalid(&format!(
2484 "column page checksum differs, column {column} part {} at {}+{} of {} bytes, \
2485 wanted {:016x} and got {:016x}",
2486 place.part,
2487 page.offset,
2488 span.start,
2489 span.length,
2490 span.hash,
2491 checksum(bytes),
2492 )));
2493 }
2494 let dictionary = self.dictionary(column)?;
2495 picked.push(decode(&field.ty, rows, bytes, dictionary)?);
2496 }
2497 Chunk::with_rows(picked, rows)
2498 }
2499
2500 #[must_use]
2510 pub fn skips(&self, part: usize, probes: &[Probe]) -> bool {
2511 let Some(place) = self.places.get(part).copied() else { return false };
2512 let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
2513 if stripe.zone.skips(probes) {
2514 return true;
2515 }
2516 probes.iter().any(|probe| self.sifted(place, probe))
2517 }
2518
2519 #[must_use]
2530 pub fn stripe_skips(&self, stripe: usize, probes: &[Probe]) -> bool {
2531 self.table.stripes.get(stripe).is_some_and(|held| held.zone.skips(probes))
2532 }
2533
2534 fn sifted(&self, place: Place, probe: &Probe) -> bool {
2540 if probe.op != Op::Equal {
2541 return false;
2542 }
2543 match self.stripe_sieves(place.stripe as usize, probe.column) {
2544 Some(sieves) => sieves
2545 .get(place.part as usize)
2546 .and_then(Option::as_ref)
2547 .is_some_and(|sieve| sieve.excludes(&probe.value)),
2548 None => false,
2549 }
2550 }
2551
2552 fn stripe_sieves(&self, stripe: usize, column: usize) -> Option<&[Option<Sieve>]> {
2559 let slot = self.sieves.get(column)?.get(stripe)?;
2560 if let Some(held) = slot.get() {
2561 return Some(held);
2562 }
2563 let page = self.table.stripes.get(stripe)?.sieves.get(column).copied().flatten()?;
2564 let mut bytes = vec![0; page.length as usize];
2565 read_at(&self.file, page.offset, &mut bytes).ok()?;
2566 if checksum(&bytes) != page.hash {
2567 return None;
2568 }
2569 let sieves = Arc::new(decode_sieves(&bytes).ok()?);
2570 let _ = slot.set(sieves);
2571 slot.get().map(|held| held.as_slice())
2572 }
2573}
2574
2575fn text_at_rank(dictionary: &Vector, rank: usize) -> Result<Value> {
2577 let code = dictionary.code_at_rank(rank)? as usize;
2578 let text = dictionary
2579 .try_text_at(code)?
2580 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
2581 Ok(Value::Varchar(text.into()))
2582}
2583
2584#[cfg(unix)]
2589fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2590 use std::os::unix::fs::FileExt;
2591 while !bytes.is_empty() {
2592 let written = file.write_at(bytes, offset).map_err(io)?;
2593 if written == 0 {
2594 return Err(invalid("a write to the native file wrote nothing"));
2595 }
2596 offset += written as u64;
2597 bytes = &bytes[written..];
2598 }
2599 Ok(())
2600}
2601
2602#[cfg(windows)]
2604fn write_at(file: &File, mut offset: u64, mut bytes: &[u8]) -> Result<()> {
2605 use std::os::windows::fs::FileExt;
2606 while !bytes.is_empty() {
2607 let written = file.seek_write(bytes, offset).map_err(io)?;
2608 if written == 0 {
2609 return Err(invalid("a write to the native file wrote nothing"));
2610 }
2611 offset += written as u64;
2612 bytes = &bytes[written..];
2613 }
2614 Ok(())
2615}
2616
2617#[cfg(not(any(unix, windows)))]
2619fn write_at(file: &File, offset: u64, bytes: &[u8]) -> Result<()> {
2620 use std::io::Write;
2621 let mut file = file.try_clone().map_err(io)?;
2622 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2623 file.write_all(bytes).map_err(io)
2624}
2625
2626#[cfg(unix)]
2636fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2637 use std::os::unix::fs::FileExt;
2638 while !bytes.is_empty() {
2639 let read = file.read_at(bytes, offset).map_err(io)?;
2640 if read == 0 {
2641 return Err(invalid("column page ends before its declared length"));
2642 }
2643 offset += read as u64;
2644 bytes = &mut bytes[read..];
2645 }
2646 Ok(())
2647}
2648
2649#[cfg(windows)]
2655fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2656 use std::os::windows::fs::FileExt;
2657 while !bytes.is_empty() {
2658 let read = file.seek_read(bytes, offset).map_err(io)?;
2659 if read == 0 {
2660 return Err(invalid("column page ends before its declared length"));
2661 }
2662 offset += read as u64;
2663 bytes = &mut bytes[read..];
2664 }
2665 Ok(())
2666}
2667
2668#[cfg(not(any(unix, windows)))]
2673fn read_at(file: &File, offset: u64, bytes: &mut [u8]) -> Result<()> {
2674 let mut file = file.try_clone().map_err(io)?;
2675 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2676 file.read_exact(bytes).map_err(io)
2677}
2678
2679fn type_tag(ty: &LogicalType) -> Result<u8> {
2680 match ty {
2681 LogicalType::SmallInt => Ok(1),
2682 LogicalType::Integer => Ok(2),
2683 LogicalType::BigInt => Ok(3),
2684 LogicalType::Varchar => Ok(4),
2685 LogicalType::Date => Ok(5),
2686 LogicalType::Timestamp => Ok(6),
2687 LogicalType::Boolean => Ok(7),
2688 LogicalType::TinyInt => Ok(8),
2689 LogicalType::UTinyInt => Ok(9),
2690 LogicalType::USmallInt => Ok(10),
2691 LogicalType::UInteger => Ok(11),
2692 LogicalType::UBigInt => Ok(12),
2693 _ => Err(Error::not_implemented(format!("native storage for {ty}"))),
2694 }
2695}
2696
2697fn tag_type(tag: u8) -> Result<LogicalType> {
2698 match tag {
2699 1 => Ok(LogicalType::SmallInt),
2700 2 => Ok(LogicalType::Integer),
2701 3 => Ok(LogicalType::BigInt),
2702 4 => Ok(LogicalType::Varchar),
2703 5 => Ok(LogicalType::Date),
2704 6 => Ok(LogicalType::Timestamp),
2705 7 => Ok(LogicalType::Boolean),
2706 8 => Ok(LogicalType::TinyInt),
2707 9 => Ok(LogicalType::UTinyInt),
2708 10 => Ok(LogicalType::USmallInt),
2709 11 => Ok(LogicalType::UInteger),
2710 12 => Ok(LogicalType::UBigInt),
2711 _ => Err(invalid("column type tag is unknown")),
2712 }
2713}
2714
2715fn put_u16(out: &mut Vec<u8>, value: u16) {
2716 out.extend_from_slice(&value.to_le_bytes());
2717}
2718fn put_u32(out: &mut Vec<u8>, value: u32) {
2719 out.extend_from_slice(&value.to_le_bytes());
2720}
2721fn put_u64(out: &mut Vec<u8>, value: u64) {
2722 out.extend_from_slice(&value.to_le_bytes());
2723}
2724fn put_var_u64(out: &mut Vec<u8>, mut value: u64) {
2725 while value >= 0x80 {
2726 out.push((value as u8 & 0x7f) | 0x80);
2727 value >>= 7;
2728 }
2729 out.push(value as u8);
2730}
2731
2732fn frequency_order(left: FrequencyValue, right: FrequencyValue) -> Ordering {
2733 match (left, right) {
2734 (FrequencyValue::Null, FrequencyValue::Null) => Ordering::Equal,
2735 (FrequencyValue::Null, _) => Ordering::Less,
2736 (_, FrequencyValue::Null) => Ordering::Greater,
2737 (FrequencyValue::Integer(left), FrequencyValue::Integer(right)) => left.cmp(&right),
2738 (FrequencyValue::Code(left), FrequencyValue::Code(right)) => left.cmp(&right),
2739 (FrequencyValue::Integer(_), FrequencyValue::Code(_)) => Ordering::Less,
2740 (FrequencyValue::Code(_), FrequencyValue::Integer(_)) => Ordering::Greater,
2741 }
2742}
2743
2744fn code_frequency(dictionary: &GlobalDictionary) -> FrequencySummary {
2745 let mut entries = dictionary
2746 .counts
2747 .iter()
2748 .enumerate()
2749 .filter(|(_, count)| **count != 0)
2750 .map(|(code, &count)| FrequencyEntry { value: FrequencyValue::Code(code as u32), count })
2751 .collect::<Vec<_>>();
2752 if dictionary.nulls != 0 {
2753 entries.push(FrequencyEntry { value: FrequencyValue::Null, count: dictionary.nulls });
2754 }
2755 entries.sort_unstable_by(|left, right| {
2756 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
2757 });
2758 let omitted_max = entries.get(FREQUENCY_ENTRIES).map_or(0, |entry| entry.count);
2759 entries.truncate(FREQUENCY_ENTRIES);
2760 FrequencySummary { entries, omitted_max, ordinals: Vec::new() }
2761}
2762
2763fn encode_directory(table: &Table) -> Result<Vec<u8>> {
2764 let mut out = DIRECTORY.to_vec();
2765 let name = table.name.as_bytes();
2766 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
2767 out.extend_from_slice(name);
2768 put_u16(&mut out, u16::try_from(table.fields.len()).map_err(|_| invalid("too many columns"))?);
2769 for field in &table.fields {
2770 let name = field.name.as_bytes();
2771 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?);
2772 out.extend_from_slice(name);
2773 out.push(type_tag(&field.ty)?);
2774 out.push(u8::from(field.not_null));
2775 }
2776 for dictionary in &table.dictionaries {
2777 match dictionary {
2778 None => out.push(0),
2779 Some(page) => {
2780 out.push(1);
2781 put_u64(&mut out, page.offset);
2782 put_u32(&mut out, page.length);
2783 put_u64(&mut out, page.hash);
2784 }
2785 }
2786 }
2787 put_u64(&mut out, u64::try_from(table.rows).map_err(|_| invalid("row count overflow"))?);
2788 put_u32(&mut out, u32::try_from(table.stripes.len()).map_err(|_| invalid("too many stripes"))?);
2789 for stripe in &table.stripes {
2790 put_u32(
2791 &mut out,
2792 u32::try_from(stripe.parts.len()).map_err(|_| invalid("too many parts in a stripe"))?,
2793 );
2794 for &rows in &stripe.parts {
2795 put_u32(&mut out, rows);
2796 }
2797 put_u64(&mut out, stripe.index.offset);
2798 put_u32(&mut out, stripe.index.length);
2799 for page in &stripe.pages {
2800 put_u64(&mut out, page.offset);
2801 put_u32(&mut out, page.length);
2802 }
2803 for (field, membership) in table.fields.iter().zip(&stripe.memberships) {
2804 if field.ty != LogicalType::Varchar {
2805 continue;
2806 }
2807 let page =
2808 membership.ok_or_else(|| invalid("string page has no code membership index"))?;
2809 put_u64(&mut out, page.offset);
2810 put_u32(&mut out, page.length);
2811 put_u64(&mut out, page.hash);
2812 }
2813 for sieve in &stripe.sieves {
2814 match sieve {
2815 None => out.push(0),
2816 Some(page) => {
2817 out.push(1);
2818 put_u64(&mut out, page.offset);
2819 put_u32(&mut out, page.length);
2820 put_u64(&mut out, page.hash);
2821 }
2822 }
2823 }
2824 for range in stripe.zone.columns() {
2825 put_bound(&mut out, range.low.as_ref())?;
2826 put_bound(&mut out, range.high.as_ref())?;
2827 put_u32(
2828 &mut out,
2829 u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?,
2830 );
2831 out.push(u8::from(range.exact));
2832 match range.sum {
2833 None => out.push(0),
2834 Some(total) => {
2835 out.push(1);
2836 out.extend_from_slice(&total.to_le_bytes());
2837 }
2838 }
2839 }
2840 }
2841 out.extend_from_slice(FREQUENCIES);
2842 put_u16(
2843 &mut out,
2844 u16::try_from(table.frequencies.len())
2845 .map_err(|_| invalid("too many frequency columns"))?,
2846 );
2847 for summary in &table.frequencies {
2848 let Some(summary) = summary else {
2849 out.push(0);
2850 continue;
2851 };
2852 out.push(1);
2853 put_u64(&mut out, summary.omitted_max);
2854 put_u32(
2855 &mut out,
2856 u32::try_from(summary.entries.len())
2857 .map_err(|_| invalid("too many frequency entries"))?,
2858 );
2859 for entry in &summary.entries {
2860 match entry.value {
2861 FrequencyValue::Null => out.push(0),
2862 FrequencyValue::Integer(value) => {
2863 out.push(1);
2864 out.extend_from_slice(&value.to_le_bytes());
2865 }
2866 FrequencyValue::Code(value) => {
2867 out.push(2);
2868 put_u32(&mut out, value);
2869 }
2870 }
2871 put_u64(&mut out, entry.count);
2872 }
2873 put_u32(
2874 &mut out,
2875 u32::try_from(summary.ordinals.len())
2876 .map_err(|_| invalid("too many frequency ordinals"))?,
2877 );
2878 let mut previous = 0_u64;
2879 for (at, &ordinal) in summary.ordinals.iter().enumerate() {
2880 let delta = if at == 0 {
2881 ordinal
2882 } else {
2883 ordinal
2884 .checked_sub(previous)
2885 .ok_or_else(|| invalid("frequency ordinals are not ordered"))?
2886 };
2887 if at != 0 && delta == 0 {
2888 return Err(invalid("frequency ordinals are not unique"));
2889 }
2890 put_var_u64(&mut out, delta);
2891 previous = ordinal;
2892 }
2893 }
2894 Ok(out)
2895}
2896
2897struct Cursor<'a> {
2898 bytes: &'a [u8],
2899 at: usize,
2900}
2901impl<'a> Cursor<'a> {
2902 fn take(&mut self, len: usize) -> Result<&'a [u8]> {
2903 let end = self.at.checked_add(len).ok_or_else(|| invalid("directory offset overflow"))?;
2904 let bytes =
2905 self.bytes.get(self.at..end).ok_or_else(|| invalid("directory is truncated"))?;
2906 self.at = end;
2907 Ok(bytes)
2908 }
2909 fn u8(&mut self) -> Result<u8> {
2910 Ok(self.take(1)?[0])
2911 }
2912 fn u16(&mut self) -> Result<u16> {
2913 Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("two bytes")))
2914 }
2915 fn u32(&mut self) -> Result<u32> {
2916 Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("four bytes")))
2917 }
2918 fn u64(&mut self) -> Result<u64> {
2919 Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("eight bytes")))
2920 }
2921 fn var_u64(&mut self) -> Result<u64> {
2922 let mut value = 0_u64;
2923 for shift in (0..=63).step_by(7) {
2924 let byte = self.u8()?;
2925 let part = u64::from(byte & 0x7f);
2926 if shift == 63 && part > 1 {
2927 return Err(invalid("frequency ordinal varint overflows"));
2928 }
2929 value |= part << shift;
2930 if byte & 0x80 == 0 {
2931 return Ok(value);
2932 }
2933 }
2934 Err(invalid("frequency ordinal varint is too long"))
2935 }
2936 fn bound(&mut self) -> Result<Option<Bound>> {
2937 Ok(match self.u8()? {
2938 0 => None,
2939 1 => Some(Bound::Int(i128::from_le_bytes(
2940 self.take(16)?.try_into().expect("sixteen bytes"),
2941 ))),
2942 2 => Some(Bound::Real(f64::from_le_bytes(
2943 self.take(8)?.try_into().expect("eight bytes"),
2944 ))),
2945 3 => {
2946 let length = self.u32()? as usize;
2947 Some(Bound::Bytes(self.take(length)?.to_vec()))
2948 }
2949 _ => return Err(invalid("bound tag differs")),
2950 })
2951 }
2952 fn text(&mut self) -> Result<String> {
2953 let len = self.u16()? as usize;
2954 String::from_utf8(self.take(len)?.to_vec()).map_err(|_| invalid("name is not UTF-8"))
2955 }
2956}
2957
2958fn decode_directory(bytes: &[u8], size: u64) -> Result<Table> {
2959 let mut cur = Cursor { bytes, at: 0 };
2960 if cur.take(8)? != DIRECTORY {
2961 return Err(invalid("directory magic differs"));
2962 }
2963 let name = cur.text()?;
2964 let width = cur.u16()? as usize;
2965 let mut fields = Vec::with_capacity(width);
2966 for _ in 0..width {
2967 let name = cur.text()?;
2968 let ty = tag_type(cur.u8()?)?;
2969 let not_null = match cur.u8()? {
2970 0 => false,
2971 1 => true,
2972 _ => return Err(invalid("nullability flag differs")),
2973 };
2974 fields.push(Field { name, ty, not_null });
2975 }
2976 let mut dictionaries = Vec::with_capacity(width);
2977 for _ in 0..width {
2978 dictionaries.push(match cur.u8()? {
2979 0 => None,
2980 1 => {
2981 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2982 let end = page
2983 .offset
2984 .checked_add(u64::from(page.length))
2985 .ok_or_else(|| invalid("dictionary page offset overflow"))?;
2986 if page.offset < HEADER || end > size {
2991 return Err(invalid("dictionary page range is outside the file"));
2992 }
2993 Some(page)
2994 }
2995 _ => return Err(invalid("dictionary page tag differs")),
2996 });
2997 }
2998 let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
2999 let count = cur.u32()? as usize;
3000 let mut stripes = Vec::with_capacity(count);
3001 let mut total = 0_usize;
3002 for _ in 0..count {
3003 let count = cur.u32()? as usize;
3004 if count == 0 || count > STRIPE_PARTS {
3005 return Err(invalid("stripe part count is outside its bound"));
3006 }
3007 let mut parts = Vec::with_capacity(count);
3008 let mut stripe_rows = 0_usize;
3009 for _ in 0..count {
3010 let rows = cur.u32()?;
3011 if rows == 0 {
3012 return Err(invalid("empty part"));
3013 }
3014 parts.push(rows);
3015 stripe_rows = stripe_rows
3016 .checked_add(rows as usize)
3017 .ok_or_else(|| invalid("stripe row count overflow"))?;
3018 }
3019 total =
3020 total.checked_add(stripe_rows).ok_or_else(|| invalid("stripe row count overflow"))?;
3021 let index = Span { offset: cur.u64()?, length: cur.u32()? };
3022 let section = index_section(count)?;
3023 let wanted = section
3024 .checked_mul(width)
3025 .and_then(|bytes| u32::try_from(bytes).ok())
3026 .ok_or_else(|| invalid("index page length overflow"))?;
3027 let end = index
3028 .offset
3029 .checked_add(u64::from(index.length))
3030 .ok_or_else(|| invalid("index page offset overflow"))?;
3031 if index.offset < HEADER || end > size || index.length != wanted {
3032 return Err(invalid("index page range is outside the file"));
3033 }
3034 let mut pages = Vec::with_capacity(width);
3035 for _ in 0..width {
3036 let offset = cur.u64()?;
3037 let length = cur.u32()?;
3038 let end = offset
3039 .checked_add(u64::from(length))
3040 .ok_or_else(|| invalid("page offset overflow"))?;
3041 if offset < HEADER || end > size || length as usize > MAX_PAGE {
3042 return Err(invalid("page range is outside the file"));
3043 }
3044 pages.push(Span { offset, length });
3045 }
3046 let mut memberships = vec![None; width];
3047 for (column, field) in fields.iter().enumerate() {
3048 if field.ty != LogicalType::Varchar {
3049 continue;
3050 }
3051 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3052 let end = page
3053 .offset
3054 .checked_add(u64::from(page.length))
3055 .ok_or_else(|| invalid("membership page offset overflow"))?;
3056 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3057 return Err(invalid("membership page range is outside the file"));
3058 }
3059 memberships[column] = Some(page);
3060 }
3061 let mut sieves = vec![None; width];
3062 for sieve in sieves.iter_mut().take(width) {
3063 match cur.u8()? {
3064 0 => continue,
3065 1 => {}
3066 _ => return Err(invalid("a sieve page has an unknown tag")),
3067 }
3068 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
3069 let end = page
3070 .offset
3071 .checked_add(u64::from(page.length))
3072 .ok_or_else(|| invalid("sieve page offset overflow"))?;
3073 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
3074 return Err(invalid("sieve page range is outside the file"));
3075 }
3076 *sieve = Some(page);
3077 }
3078 let mut ranges = Vec::with_capacity(width);
3079 for _ in 0..width {
3080 let low = cur.bound()?;
3081 let high = cur.bound()?;
3082 let nulls = cur.u32()? as usize;
3083 if nulls > stripe_rows {
3084 return Err(invalid("null count exceeds stripe rows"));
3085 }
3086 let exact = cur.u8()? != 0;
3087 let sum = match cur.u8()? {
3088 0 => None,
3089 1 => Some(i128::from_le_bytes(
3090 cur.take(16)?.try_into().map_err(|_| invalid("a stripe sum is truncated"))?,
3091 )),
3092 _ => return Err(invalid("a stripe sum has an unknown tag")),
3093 };
3094 ranges.push(Range { low, high, nulls, exact, sum });
3095 }
3096 stripes.push(Stripe {
3097 rows: stripe_rows,
3098 parts,
3099 index,
3100 pages,
3101 memberships,
3102 sieves,
3103 zone: Zone::from_ranges(ranges),
3104 });
3105 }
3106 if total != rows {
3107 return Err(invalid("table row count differs from stripes"));
3108 }
3109 let frequencies = if cur.at == bytes.len() {
3110 vec![None; width]
3111 } else {
3112 if cur.take(8)? != FREQUENCIES {
3113 return Err(invalid("directory extension magic differs"));
3114 }
3115 if cur.u16()? as usize != width {
3116 return Err(invalid("frequency column count differs"));
3117 }
3118 let mut frequencies = Vec::with_capacity(width);
3119 for field in &fields {
3120 let summary = match cur.u8()? {
3121 0 => None,
3122 1 => {
3123 let omitted_max = cur.u64()?;
3124 let count = cur.u32()? as usize;
3125 if count > FREQUENCY_ENTRIES {
3126 return Err(invalid("frequency entry count exceeds its bound"));
3127 }
3128 let mut entries = Vec::with_capacity(count);
3129 for _ in 0..count {
3131 let value = match cur.u8()? {
3132 0 => FrequencyValue::Null,
3133 1 => FrequencyValue::Integer(i128::from_le_bytes(
3134 cur.take(16)?.try_into().expect("sixteen bytes"),
3135 )),
3136 2 => FrequencyValue::Code(cur.u32()?),
3137 _ => return Err(invalid("frequency value tag differs")),
3138 };
3139 let valid = matches!(
3140 (&field.ty, value),
3141 (_, FrequencyValue::Null)
3142 | (LogicalType::Varchar, FrequencyValue::Code(_))
3143 | (
3144 LogicalType::TinyInt
3145 | LogicalType::SmallInt
3146 | LogicalType::Integer
3147 | LogicalType::BigInt
3148 | LogicalType::UTinyInt
3149 | LogicalType::USmallInt
3150 | LogicalType::UInteger
3151 | LogicalType::UBigInt
3152 | LogicalType::Date
3153 | LogicalType::Timestamp,
3154 FrequencyValue::Integer(_),
3155 )
3156 );
3157 if !valid {
3158 return Err(invalid("frequency value does not match its column"));
3159 }
3160 let count = cur.u64()?;
3161 if count == 0 || count > rows as u64 {
3162 return Err(invalid("frequency count is outside the table"));
3163 }
3164 entries.push(FrequencyEntry { value, count });
3165 }
3166 if entries.windows(2).any(|pair| pair[0].count < pair[1].count) {
3167 return Err(invalid("frequency entries are not descending"));
3168 }
3169 let ordinals = {
3170 let ordinal_count = cur.u32()? as usize;
3171 if ordinal_count > FREQUENCY_ORDINALS || ordinal_count > rows {
3172 return Err(invalid("frequency ordinal count exceeds its bound"));
3173 }
3174 let mut ordinals = Vec::with_capacity(ordinal_count);
3175 let mut previous = 0_u64;
3176 for at in 0..ordinal_count {
3177 let delta = cur.var_u64()?;
3178 if at != 0 && delta == 0 {
3179 return Err(invalid("frequency ordinals are not increasing"));
3180 }
3181 let ordinal = if at == 0 {
3182 delta
3183 } else {
3184 previous
3185 .checked_add(delta)
3186 .ok_or_else(|| invalid("frequency ordinal overflows"))?
3187 };
3188 if ordinal >= rows as u64 {
3189 return Err(invalid("frequency ordinal is outside the table"));
3190 }
3191 ordinals.push(ordinal);
3192 previous = ordinal;
3193 }
3194 ordinals
3195 };
3196 Some(FrequencySummary { entries, omitted_max, ordinals })
3197 }
3198 _ => return Err(invalid("frequency summary tag differs")),
3199 };
3200 frequencies.push(summary);
3201 }
3202 frequencies
3203 };
3204 if cur.at != bytes.len() {
3205 return Err(invalid("directory has trailing bytes"));
3206 }
3207 Ok(Table { name, fields, stripes, rows, dictionaries, frequencies })
3208}
3209
3210fn put_bound(out: &mut Vec<u8>, bound: Option<&Bound>) -> Result<()> {
3211 match bound {
3212 None => out.push(0),
3213 Some(Bound::Int(value)) => {
3214 out.push(1);
3215 out.extend_from_slice(&value.to_le_bytes());
3216 }
3217 Some(Bound::Real(value)) => {
3218 out.push(2);
3219 out.extend_from_slice(&value.to_le_bytes());
3220 }
3221 Some(Bound::Bytes(value)) => {
3222 out.push(3);
3223 put_u32(out, u32::try_from(value.len()).map_err(|_| invalid("bound length overflow"))?);
3224 out.extend_from_slice(value);
3225 }
3226 }
3227 Ok(())
3228}
3229
3230#[derive(Debug)]
3247struct Codes;
3248
3249impl chooser::Chooser for Codes {
3250 fn name(&self) -> &'static str {
3251 "codes"
3252 }
3253
3254 fn narrow_strings(
3255 &self,
3256 _values: &[&[u8]],
3257 offered: &[string::Kind],
3258 _depth: u8,
3259 ) -> Vec<string::Kind> {
3260 offered.to_vec()
3263 }
3264
3265 fn narrow_integers(
3266 &self,
3267 _values: &[i64],
3268 offered: &[integer::Kind],
3269 depth: u8,
3270 ) -> Vec<integer::Kind> {
3271 let keep: &[integer::Kind] = if depth == 0 {
3272 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Rle]
3273 } else {
3274 &[integer::Kind::Constant, integer::Kind::Packed]
3275 };
3276 let narrowed: Vec<integer::Kind> =
3277 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3278 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3281 }
3282}
3283
3284#[derive(Debug)]
3296struct Fixed;
3297
3298impl chooser::Chooser for Fixed {
3299 fn name(&self) -> &'static str {
3300 "fixed"
3301 }
3302
3303 fn narrow_strings(
3304 &self,
3305 _values: &[&[u8]],
3306 offered: &[string::Kind],
3307 _depth: u8,
3308 ) -> Vec<string::Kind> {
3309 offered.to_vec()
3310 }
3311
3312 fn narrow_integers(
3313 &self,
3314 _values: &[i64],
3315 offered: &[integer::Kind],
3316 depth: u8,
3317 ) -> Vec<integer::Kind> {
3318 let keep: &[integer::Kind] = if depth == 0 {
3319 &[
3320 integer::Kind::Constant,
3321 integer::Kind::Packed,
3322 integer::Kind::Delta,
3323 integer::Kind::Rle,
3324 integer::Kind::Sparse,
3325 integer::Kind::Strided,
3326 ]
3327 } else {
3328 &[integer::Kind::Constant, integer::Kind::Packed, integer::Kind::Delta]
3329 };
3330 let narrowed: Vec<integer::Kind> =
3331 offered.iter().copied().filter(|kind| keep.contains(kind)).collect();
3332 if narrowed.is_empty() { offered.to_vec() } else { narrowed }
3333 }
3334}
3335
3336fn widened(data: &Data) -> Option<Vec<i64>> {
3343 match data {
3344 Data::Int8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3345 Data::UInt8(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3346 Data::Int16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3347 Data::UInt16(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3348 Data::Int32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3349 Data::UInt32(values) => Some(values.iter().map(|value| i64::from(*value)).collect()),
3350 Data::Int64(values) => Some(values.to_vec()),
3351 _ => None,
3352 }
3353}
3354
3355fn narrowed(ty: &LogicalType, values: Vec<i64>) -> Result<Data> {
3360 fn fit<T: TryFrom<i64>>(values: &[i64]) -> Result<Vec<T>> {
3361 values
3362 .iter()
3363 .map(|value| T::try_from(*value).map_err(|_| invalid("page value is not of its type")))
3364 .collect()
3365 }
3366 Ok(match ty {
3367 LogicalType::TinyInt => Data::Int8(fit::<i8>(&values)?.into()),
3368 LogicalType::UTinyInt => Data::UInt8(fit::<u8>(&values)?.into()),
3369 LogicalType::SmallInt => Data::Int16(fit::<i16>(&values)?.into()),
3370 LogicalType::USmallInt => Data::UInt16(fit::<u16>(&values)?.into()),
3371 LogicalType::Integer | LogicalType::Date => Data::Int32(fit::<i32>(&values)?.into()),
3372 LogicalType::UInteger => Data::UInt32(fit::<u32>(&values)?.into()),
3373 LogicalType::BigInt | LogicalType::Timestamp => Data::Int64(values.into()),
3374 _ => return Err(invalid("cascade codec belongs to a page that is not integers")),
3375 })
3376}
3377
3378fn plain_width(ty: &LogicalType) -> Option<usize> {
3381 Some(match ty {
3382 LogicalType::TinyInt | LogicalType::UTinyInt => 1,
3383 LogicalType::SmallInt | LogicalType::USmallInt => 2,
3384 LogicalType::Integer | LogicalType::UInteger | LogicalType::Date => 4,
3385 LogicalType::BigInt | LogicalType::Timestamp => 8,
3386 _ => return None,
3387 })
3388}
3389
3390fn cascaded(
3396 flat: &Vector,
3397 ty: &LogicalType,
3398 packed: Option<&Packed<'_>>,
3399) -> Result<Option<Vec<u8>>> {
3400 let (Some(width), Some(data)) = (plain_width(ty), flat.data()) else { return Ok(None) };
3401 let Some(values) = widened(data) else { return Ok(None) };
3402 let plain = values.len().saturating_mul(width);
3403 let best = match packed {
3404 Some(packed) => plain.min(21 + size_of_val(packed.words())),
3406 None => plain,
3407 };
3408 let out = integer::encode_with(&values, &Fixed)?;
3409 Ok((out.len() < best).then_some(out))
3410}
3411
3412fn encoded_codes(codes: &[u32]) -> Result<Option<Vec<u8>>> {
3424 let wide: Vec<i64> = codes.iter().map(|code| i64::from(*code)).collect();
3425 let coded = integer::encode_with(&wide, &Codes)?;
3426 let plain = codes.len().saturating_mul(size_of::<u32>());
3427 Ok((coded.len() < plain).then_some(coded))
3428}
3429
3430fn encode(
3431 vector: &Vector,
3432 global: Option<&mut GlobalDictionary>,
3433) -> Result<(Vec<u8>, Option<Vec<u32>>)> {
3434 let ty = vector.logical_type();
3435 let flat = vector.flatten()?;
3437 let mut out = Vec::new();
3438 let mut global_codes = None;
3439 if let Some(global) = global {
3440 let mut codes = Vec::with_capacity(flat.len());
3441 for row in 0..flat.len() {
3442 let text = flat.text_at(row).unwrap_or("");
3443 let code = global.code(text)?;
3444 global.observe(code, flat.is_null_at(row))?;
3445 codes.push(code);
3446 }
3447 global_codes = Some(codes);
3448 }
3449 let membership = global_codes.as_deref().map(unique_codes);
3450 let dictionary = if global_codes.is_none() && ty == &LogicalType::Varchar {
3451 string_dictionary(&flat)?
3452 } else {
3453 None
3454 };
3455 let packed_vector = if dictionary.is_none() && global_codes.is_none() {
3456 Some(flat.bit_packed()?)
3457 } else {
3458 None
3459 };
3460 let packed = packed_vector.as_ref().and_then(Vector::packed_parts);
3461 let coded = match global_codes.as_deref() {
3462 Some(codes) => encoded_codes(codes)?,
3463 None => None,
3464 };
3465 let cascade = if dictionary.is_none() && global_codes.is_none() {
3469 cascaded(&flat, ty, packed.as_ref())?
3470 } else {
3471 None
3472 };
3473 out.push(if coded.is_some() {
3474 4
3475 } else if cascade.is_some() {
3476 5
3477 } else if global_codes.is_some() {
3478 3
3479 } else if dictionary.is_some() {
3480 1
3481 } else if packed.is_some() {
3482 2
3483 } else {
3484 0
3485 });
3486 let nulls = flat.validity();
3487 let flag = match nulls {
3488 Validity::AllValid => 0,
3489 Validity::AllInvalid => 1,
3490 Validity::Mask(_) => 2,
3491 };
3492 out.push(flag);
3493 if flag == 2 {
3494 for group in (0..vector.len()).step_by(8) {
3495 let mut bits = 0_u8;
3496 for bit in 0..8 {
3497 if group + bit < vector.len() && !flat.is_null_at(group + bit) {
3498 bits |= 1 << bit;
3499 }
3500 }
3501 out.push(bits);
3502 }
3503 }
3504 if let Some(coded) = coded {
3505 out.extend_from_slice(&coded);
3506 return Ok((out, membership));
3507 }
3508 if let Some(cascade) = cascade {
3509 out.extend_from_slice(&cascade);
3510 return Ok((out, membership));
3511 }
3512 if let Some(codes) = global_codes {
3513 for code in codes {
3514 put_u32(&mut out, code);
3515 }
3516 return Ok((out, membership));
3517 }
3518 if let Some(dictionary) = dictionary {
3519 out.extend_from_slice(&dictionary);
3520 return Ok((out, membership));
3521 }
3522 if let Some(packed) = packed {
3523 if packed.offset() != 0 {
3524 return Err(invalid("writer received a sliced packed vector"));
3525 }
3526 out.push(u8::try_from(packed.width()).map_err(|_| invalid("packed width overflow"))?);
3527 out.extend_from_slice(&packed.base().to_le_bytes());
3528 put_u32(
3529 &mut out,
3530 u32::try_from(packed.words().len()).map_err(|_| invalid("too many packed words"))?,
3531 );
3532 for word in packed.words() {
3533 put_u64(&mut out, *word);
3534 }
3535 return Ok((out, membership));
3536 }
3537 let data = flat.data().ok_or_else(|| invalid("scalar column did not flatten"))?;
3538 match (ty, data) {
3539 (LogicalType::TinyInt, Data::Int8(values)) => {
3540 for value in &**values {
3541 out.extend_from_slice(&value.to_le_bytes());
3542 }
3543 }
3544 (LogicalType::UTinyInt, Data::UInt8(values)) => {
3545 for value in &**values {
3546 out.extend_from_slice(&value.to_le_bytes());
3547 }
3548 }
3549 (LogicalType::SmallInt, Data::Int16(values)) => {
3550 for value in &**values {
3551 out.extend_from_slice(&value.to_le_bytes());
3552 }
3553 }
3554 (LogicalType::USmallInt, Data::UInt16(values)) => {
3555 for value in &**values {
3556 out.extend_from_slice(&value.to_le_bytes());
3557 }
3558 }
3559 (LogicalType::UInteger, Data::UInt32(values)) => {
3560 for value in &**values {
3561 out.extend_from_slice(&value.to_le_bytes());
3562 }
3563 }
3564 (LogicalType::UBigInt, Data::UInt64(values)) => {
3565 for value in &**values {
3566 out.extend_from_slice(&value.to_le_bytes());
3567 }
3568 }
3569 (LogicalType::Integer | LogicalType::Date, Data::Int32(values)) => {
3570 for value in &**values {
3571 out.extend_from_slice(&value.to_le_bytes());
3572 }
3573 }
3574 (LogicalType::BigInt | LogicalType::Timestamp, Data::Int64(values)) => {
3575 for value in &**values {
3576 out.extend_from_slice(&value.to_le_bytes());
3577 }
3578 }
3579 (LogicalType::Boolean, Data::Bool(values)) => {
3580 for value in &**values {
3581 out.push(u8::from(*value));
3582 }
3583 }
3584 (LogicalType::Varchar, Data::Varlen(values)) => {
3585 let mut bytes = Vec::new();
3586 put_u32(&mut out, 0);
3587 for row in 0..vector.len() {
3588 let value = values.bytes(row).ok_or_else(|| invalid("string view is invalid"))?;
3589 bytes.extend_from_slice(value);
3590 put_u32(
3591 &mut out,
3592 u32::try_from(bytes.len())
3593 .map_err(|_| invalid("string payload exceeds 4GiB"))?,
3594 );
3595 }
3596 out.extend_from_slice(&bytes);
3597 }
3598 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
3599 }
3600 Ok((out, membership))
3601}
3602
3603fn put_varint(out: &mut Vec<u8>, mut value: u32) {
3604 while value >= 0x80 {
3605 out.push((value as u8 & 0x7f) | 0x80);
3606 value >>= 7;
3607 }
3608 out.push(value as u8);
3609}
3610
3611fn unique_codes(codes: &[u32]) -> Vec<u32> {
3613 let mut unique = codes.to_vec();
3614 unique.sort_unstable();
3615 unique.dedup();
3616 unique
3617}
3618
3619fn merged_codes(lists: Vec<Vec<u32>>) -> Vec<u32> {
3625 let mut lists = lists;
3626 while lists.len() > 1 {
3627 let mut next = Vec::with_capacity(lists.len().div_ceil(2));
3628 for pair in lists.chunks(2) {
3629 match pair {
3630 [left, right] => next.push(merged_pair(left, right)),
3631 [only] => next.push(only.clone()),
3632 _ => {}
3633 }
3634 }
3635 lists = next;
3636 }
3637 lists.pop().unwrap_or_default()
3638}
3639
3640fn merged_pair(left: &[u32], right: &[u32]) -> Vec<u32> {
3641 let mut out = Vec::with_capacity(left.len().saturating_add(right.len()));
3642 let mut at = 0;
3643 let mut to = 0;
3644 while at < left.len() && to < right.len() {
3645 match left[at].cmp(&right[to]) {
3646 Ordering::Less => {
3647 out.push(left[at]);
3648 at += 1;
3649 }
3650 Ordering::Greater => {
3651 out.push(right[to]);
3652 to += 1;
3653 }
3654 Ordering::Equal => {
3655 out.push(left[at]);
3656 at += 1;
3657 to += 1;
3658 }
3659 }
3660 }
3661 out.extend_from_slice(&left[at..]);
3662 out.extend_from_slice(&right[to..]);
3663 out
3664}
3665
3666fn merged_range(ranges: impl Iterator<Item = Range>) -> Range {
3671 let mut merged = Range::default();
3672 let mut first = true;
3673 for range in ranges {
3674 merged.nulls = merged.nulls.saturating_add(range.nulls);
3675 merged.sum = match (merged.sum.take(), range.sum) {
3679 (Some(held), Some(next)) if !first => held.checked_add(next),
3680 (_, next) if first => next,
3681 _ => None,
3682 };
3683 merged.exact = if first { range.exact } else { merged.exact && range.exact };
3684 if first {
3685 merged.low = range.low;
3686 merged.high = range.high;
3687 first = false;
3688 continue;
3689 }
3690 merged.low = match (merged.low.take(), range.low) {
3691 (Some(held), Some(next)) => Some(held.smaller(next)),
3692 _ => None,
3693 };
3694 merged.high = match (merged.high.take(), range.high) {
3695 (Some(held), Some(next)) => Some(held.larger(next)),
3696 _ => None,
3697 };
3698 }
3699 merged
3700}
3701
3702fn encode_sieves<'a>(sieves: impl Iterator<Item = &'a Option<Sieve>>) -> Result<Vec<u8>> {
3708 let held: Vec<&Option<Sieve>> = sieves.collect();
3709 let mut out = Vec::new();
3710 put_u32(
3711 &mut out,
3712 u32::try_from(held.len()).map_err(|_| invalid("too many parts in a stripe"))?,
3713 );
3714 for sieve in &held {
3715 let length = sieve.as_ref().map_or(0, Sieve::len);
3716 put_u32(&mut out, u32::try_from(length).map_err(|_| invalid("sieve length overflow"))?);
3717 }
3718 for sieve in held.into_iter().flatten() {
3720 out.extend_from_slice(&sieve.to_bytes());
3721 }
3722 Ok(out)
3723}
3724
3725fn decode_sieves(bytes: &[u8]) -> Result<Vec<Option<Sieve>>> {
3731 let parts = u32::from_le_bytes(
3732 bytes
3733 .get(..4)
3734 .ok_or_else(|| invalid("sieve page is truncated"))?
3735 .try_into()
3736 .map_err(|_| invalid("sieve page is truncated"))?,
3737 ) as usize;
3738 let mut lengths = Vec::with_capacity(parts);
3739 for part in 0..parts {
3740 let at = 4 + part * 4;
3741 let field = bytes.get(at..at + 4).ok_or_else(|| invalid("sieve page is truncated"))?;
3742 lengths.push(u32::from_le_bytes(
3743 field.try_into().map_err(|_| invalid("sieve page is truncated"))?,
3744 ) as usize);
3745 }
3746 let mut at = 4 + parts * 4;
3747 let mut out = Vec::with_capacity(parts);
3748 for length in lengths {
3749 if length == 0 {
3750 out.push(None);
3751 continue;
3752 }
3753 let end = at.checked_add(length).ok_or_else(|| invalid("sieve page is truncated"))?;
3754 let field = bytes.get(at..end).ok_or_else(|| invalid("sieve page is truncated"))?;
3755 out.push(Sieve::from_bytes(field));
3756 at = end;
3757 }
3758 if at != bytes.len() {
3759 return Err(invalid("sieve page has trailing bytes"));
3760 }
3761 Ok(out)
3762}
3763
3764fn encode_membership(unique: &[u32]) -> Vec<u8> {
3770 let mut out = Vec::with_capacity(unique.len().saturating_mul(2).saturating_add(5));
3771 put_varint(&mut out, u32::try_from(unique.len()).unwrap_or(u32::MAX));
3772 let mut previous = 0;
3773 for (at, &code) in unique.iter().enumerate() {
3774 put_varint(&mut out, if at == 0 { code } else { code - previous });
3775 previous = code;
3776 }
3777 out
3778}
3779
3780fn take_varint(bytes: &[u8], at: &mut usize) -> Result<u32> {
3781 let mut value = 0_u32;
3782 for shift in (0..35).step_by(7) {
3783 let byte = *bytes.get(*at).ok_or_else(|| invalid("membership varint is truncated"))?;
3784 *at += 1;
3785 let part = u32::from(byte & 0x7f);
3786 if shift == 28 && part > 0x0f {
3787 return Err(invalid("membership varint overflow"));
3788 }
3789 value = value
3790 .checked_add(
3791 part.checked_shl(shift).ok_or_else(|| invalid("membership varint overflow"))?,
3792 )
3793 .ok_or_else(|| invalid("membership varint overflow"))?;
3794 if byte & 0x80 == 0 {
3795 return Ok(value);
3796 }
3797 }
3798 Err(invalid("membership varint is too long"))
3799}
3800
3801fn decode_membership(bytes: &[u8]) -> Result<Vec<u32>> {
3802 let mut at = 0;
3803 let count = take_varint(bytes, &mut at)? as usize;
3804 let mut codes = Vec::with_capacity(count);
3805 let mut previous = 0_u32;
3806 for index in 0..count {
3807 let delta = take_varint(bytes, &mut at)?;
3808 let code = if index == 0 {
3809 delta
3810 } else {
3811 previous.checked_add(delta).ok_or_else(|| invalid("membership code overflow"))?
3812 };
3813 if index > 0 && code <= previous {
3814 return Err(invalid("membership codes are not increasing"));
3815 }
3816 codes.push(code);
3817 previous = code;
3818 }
3819 if at != bytes.len() {
3820 return Err(invalid("membership page has trailing bytes"));
3821 }
3822 Ok(codes)
3823}
3824
3825fn string_dictionary(vector: &Vector) -> Result<Option<Vec<u8>>> {
3826 let mut by_text = HashMap::new();
3827 let mut values = Vec::new();
3828 let mut codes = Vec::with_capacity(vector.len());
3829 let mut plain_bytes = 0_usize;
3830 for row in 0..vector.len() {
3831 let text = vector.text_at(row).unwrap_or("");
3832 plain_bytes = plain_bytes.saturating_add(text.len());
3833 let code = match by_text.get(text) {
3834 Some(&code) => code,
3835 None => {
3836 let code = u32::try_from(values.len())
3837 .map_err(|_| invalid("too many dictionary values"))?;
3838 by_text.insert(text, code);
3839 values.push(text);
3840 code
3841 }
3842 };
3843 codes.push(code);
3844 }
3845 let dictionary_bytes = values.iter().map(|value| value.len()).sum::<usize>();
3846 let encoded = 8_usize
3847 .saturating_add((values.len() + 1).saturating_mul(4))
3848 .saturating_add(dictionary_bytes)
3849 .saturating_add(codes.len().saturating_mul(4));
3850 let plain = (vector.len() + 1).saturating_mul(4).saturating_add(plain_bytes);
3851 if encoded >= plain {
3852 return Ok(None);
3853 }
3854 let mut out = Vec::with_capacity(encoded);
3855 put_u32(
3856 &mut out,
3857 u32::try_from(values.len()).map_err(|_| invalid("too many dictionary values"))?,
3858 );
3859 put_u32(
3860 &mut out,
3861 u32::try_from(dictionary_bytes).map_err(|_| invalid("dictionary payload exceeds 4GiB"))?,
3862 );
3863 let mut offset = 0_u32;
3864 put_u32(&mut out, offset);
3865 for value in &values {
3866 offset = offset
3867 .checked_add(
3868 u32::try_from(value.len()).map_err(|_| invalid("dictionary value is too long"))?,
3869 )
3870 .ok_or_else(|| invalid("dictionary payload exceeds 4GiB"))?;
3871 put_u32(&mut out, offset);
3872 }
3873 for value in values {
3874 out.extend_from_slice(value.as_bytes());
3875 }
3876 for code in codes {
3877 put_u32(&mut out, code);
3878 }
3879 Ok(Some(out))
3880}
3881
3882struct EncodedDictionary {
3883 index: Vec<u8>,
3884 ranks: Vec<u8>,
3885 payload: Vec<Vec<u8>>,
3888}
3889
3890fn head(bytes: &[u8]) -> u64 {
3892 let mut word = [0; 8];
3893 let take = bytes.len().min(8);
3894 word[..take].copy_from_slice(&bytes[..take]);
3895 u64::from_be_bytes(word)
3896}
3897
3898fn rankings(dictionaries: &[Option<GlobalDictionary>]) -> Result<Vec<Vec<(u64, u32)>>> {
3906 let present =
3907 dictionaries.iter().enumerate().filter(|(_, held)| held.is_some()).map(|(at, _)| at);
3908 let present = present.collect::<Vec<_>>();
3909 let mut orders = vec![Vec::new(); dictionaries.len()];
3910 let workers = std::thread::available_parallelism()
3911 .map_or(1, usize::from)
3912 .min(MAX_FREQUENCY_WORKERS)
3913 .min(present.len());
3914 if workers <= 1 {
3915 for at in present {
3916 if let Some(dictionary) = &dictionaries[at] {
3917 orders[at] = dictionary.ranked();
3918 }
3919 }
3920 return Ok(orders);
3921 }
3922 let width = present.len().div_ceil(workers);
3923 let pieces = std::thread::scope(|scope| {
3924 present
3925 .chunks(width)
3926 .map(|columns| {
3927 scope.spawn(|| {
3928 columns
3929 .iter()
3930 .filter_map(|&at| dictionaries[at].as_ref().map(|held| (at, held.ranked())))
3931 .collect::<Vec<_>>()
3932 })
3933 })
3934 .collect::<Vec<_>>()
3935 .into_iter()
3936 .map(|handle| {
3937 handle.join().map_err(|_| Error::internal("a dictionary sort worker panicked"))
3938 })
3939 .collect::<Result<Vec<_>>>()
3940 })?;
3941 for piece in pieces {
3942 for (at, order) in piece {
3943 orders[at] = order;
3944 }
3945 }
3946 Ok(orders)
3947}
3948
3949fn encode_global_dictionary(
3950 dictionary: GlobalDictionary,
3951 order: &[(u64, u32)],
3952) -> Result<EncodedDictionary> {
3953 let values = dictionary.offsets.len() - 1;
3954 if order.len() != values {
3955 return Err(invalid("global dictionary order does not cover its values"));
3956 }
3957 let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
3958 let payload = encode_payload(&dictionary)?;
3959 if payload.len() != blocks {
3960 return Err(invalid("global dictionary payload is not the blocks it says it is"));
3961 }
3962 let ranks = encode_ranks(order);
3963 let rank_blocks = values.div_ceil(TEXT_RANK_BLOCK);
3964 let mut index = Vec::with_capacity(12 + (values + 1) * 4 + (blocks * 2 + rank_blocks) * 8);
3965 put_u32(
3966 &mut index,
3967 u32::try_from(values).map_err(|_| invalid("global dictionary has too many values"))?,
3968 );
3969 put_u32(&mut index, TEXT_PAYLOAD_VALUES as u32);
3970 put_u32(
3971 &mut index,
3972 u32::try_from(blocks).map_err(|_| invalid("global dictionary has too many blocks"))?,
3973 );
3974 for offset in dictionary.offsets {
3975 put_u32(&mut index, offset);
3976 }
3977 let mut at = 0_u64;
3981 for block in &payload {
3982 at = at
3983 .checked_add(block.len() as u64)
3984 .ok_or_else(|| invalid("global dictionary payload overflow"))?;
3985 put_u64(&mut index, at);
3986 }
3987 for block in &payload {
3988 put_u64(&mut index, checksum(block));
3989 }
3990 for block in ranks.chunks(TEXT_RANK_BLOCK * RANK_ENTRY) {
3991 put_u64(&mut index, checksum(block));
3992 }
3993 Ok(EncodedDictionary { index, ranks, payload })
3994}
3995
3996const PAYLOAD_SAMPLE_BLOCKS: usize = 8;
4003
4004fn payload_shapes() -> Vec<chooser::Settled> {
4030 let integers = vec![integer::Kind::Packed];
4031 [
4032 vec![string::Kind::Front, string::Kind::Lz],
4033 vec![string::Kind::Lz, string::Kind::Fsst],
4034 vec![string::Kind::Lz, string::Kind::Plain],
4035 vec![string::Kind::Fsst],
4036 vec![string::Kind::Plain],
4037 ]
4038 .into_iter()
4039 .map(|strings| chooser::Settled::new(strings, integers.clone()))
4040 .collect()
4041}
4042
4043fn encode_payload(dictionary: &GlobalDictionary) -> Result<Vec<Vec<u8>>> {
4049 let values = dictionary.offsets.len() - 1;
4050 let blocks = values.div_ceil(TEXT_PAYLOAD_VALUES);
4051 let run = |block: usize| {
4052 let first = block * TEXT_PAYLOAD_VALUES;
4053 let last = (first + TEXT_PAYLOAD_VALUES).min(values);
4054 (first..last)
4055 .map(|value| {
4056 let from = dictionary.offsets[value] as usize;
4057 let to = dictionary.offsets[value + 1] as usize;
4058 &dictionary.payload[from..to]
4059 })
4060 .collect::<Vec<_>>()
4061 };
4062 let shape = (blocks > PAYLOAD_SAMPLE_BLOCKS).then(|| settle_shape(&run, blocks)).transpose()?;
4065 let one = |block: usize| match &shape {
4066 Some(shape) => string::encode_with(&run(block), shape),
4067 None => string::encode(&run(block)),
4068 };
4069 let workers = std::thread::available_parallelism()
4070 .map_or(1, usize::from)
4071 .min(MAX_FREQUENCY_WORKERS)
4072 .min(blocks);
4073 if workers <= 1 {
4074 return (0..blocks).map(one).collect();
4075 }
4076 let next = AtomicUsize::new(0);
4077 let pieces = std::thread::scope(|scope| {
4078 (0..workers)
4079 .map(|_| {
4080 scope.spawn(|| {
4081 let mut mine = Vec::new();
4082 loop {
4083 let block = next.fetch_add(1, Atomic::Relaxed);
4084 if block >= blocks {
4085 break;
4086 }
4087 mine.push((block, one(block)?));
4088 }
4089 Ok(mine)
4090 })
4091 })
4092 .collect::<Vec<_>>()
4093 .into_iter()
4094 .map(|handle| {
4095 handle.join().map_err(|_| Error::internal("a dictionary encode worker panicked"))?
4096 })
4097 .collect::<Result<Vec<_>>>()
4098 })?;
4099 let mut payload = vec![Vec::new(); blocks];
4100 for piece in pieces {
4101 for (block, bytes) in piece {
4102 payload[block] = bytes;
4103 }
4104 }
4105 Ok(payload)
4106}
4107
4108fn settle_shape<'a>(
4116 run: &dyn Fn(usize) -> Vec<&'a [u8]>,
4117 blocks: usize,
4118) -> Result<chooser::Settled> {
4119 let last = blocks - 1;
4120 let sample = (0..PAYLOAD_SAMPLE_BLOCKS)
4121 .map(|region| run(region * last / (PAYLOAD_SAMPLE_BLOCKS - 1)))
4122 .collect::<Vec<_>>();
4123 let mut best: Option<(chooser::Settled, usize)> = None;
4124 for shape in payload_shapes() {
4125 let mut size = 0;
4126 for block in &sample {
4127 size += string::encode_with(block, &shape)?.len();
4128 }
4129 if best.as_ref().is_none_or(|(_, smallest)| size < *smallest) {
4130 best = Some((shape, size));
4131 }
4132 }
4133 best.map(|(shape, _)| shape)
4134 .ok_or_else(|| invalid("no shape applies to a global dictionary payload"))
4135}
4136
4137fn encode_ranks(order: &[(u64, u32)]) -> Vec<u8> {
4144 let mut out = Vec::with_capacity(order.len() * RANK_ENTRY);
4145 for block in order.chunks(TEXT_RANK_BLOCK) {
4146 for &(head, _) in block {
4147 put_u64(&mut out, head);
4148 }
4149 for &(_, code) in block {
4150 put_u32(&mut out, code);
4151 }
4152 }
4153 out
4154}
4155
4156fn open_global_dictionary(file: Arc<File>, page: Page, ty: &LogicalType) -> Result<Vector> {
4157 if ty != &LogicalType::Varchar {
4158 return Err(invalid("global dictionary belongs to a non-string column"));
4159 }
4160 let mut header = [0; 12];
4161 read_at(&file, page.offset, &mut header)?;
4162 let count = u32::from_le_bytes(header[0..4].try_into().expect("four bytes")) as usize;
4163 let per_block = u32::from_le_bytes(header[4..8].try_into().expect("four bytes")) as usize;
4164 let blocks = u32::from_le_bytes(header[8..12].try_into().expect("four bytes")) as usize;
4165 if per_block != TEXT_PAYLOAD_VALUES {
4166 return Err(invalid("global dictionary block width differs"));
4167 }
4168 if blocks != count.div_ceil(TEXT_PAYLOAD_VALUES) {
4169 return Err(invalid("global dictionary block count differs from its value count"));
4170 }
4171 let offset_len = (count + 1)
4172 .checked_mul(4)
4173 .ok_or_else(|| invalid("global dictionary offset count overflow"))?;
4174 let ranks = count;
4179 let rank_blocks = ranks.div_ceil(TEXT_RANK_BLOCK);
4180 let rank_len =
4181 ranks.checked_mul(RANK_ENTRY).ok_or_else(|| invalid("global dictionary rank overflow"))?;
4182 let hash_len = blocks
4185 .checked_mul(2)
4186 .and_then(|words| words.checked_add(rank_blocks))
4187 .and_then(|words| words.checked_mul(8))
4188 .ok_or_else(|| invalid("global dictionary block count overflow"))?;
4189 let index_len = 12usize
4190 .checked_add(offset_len)
4191 .and_then(|len| len.checked_add(hash_len))
4192 .ok_or_else(|| invalid("global dictionary header overflow"))?;
4193 let body_len = index_len
4194 .checked_add(rank_len)
4195 .ok_or_else(|| invalid("global dictionary header overflow"))?;
4196 if body_len > page.length as usize {
4197 return Err(invalid("global dictionary offset index exceeds its page"));
4198 }
4199 let mut index = vec![0; index_len];
4200 index[..12].copy_from_slice(&header);
4201 read_at(&file, page.offset + 12, &mut index[12..])?;
4202 if checksum(&index) != page.hash {
4203 return Err(invalid("global dictionary index checksum differs"));
4204 }
4205 let offsets = index[12..12 + offset_len]
4206 .chunks_exact(4)
4207 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4208 .collect::<Vec<_>>();
4209 let mut words = index[12 + offset_len..]
4210 .chunks_exact(8)
4211 .map(|part| u64::from_le_bytes(part.try_into().expect("eight bytes")))
4212 .collect::<Vec<_>>();
4213 let mut hashes = words.split_off(blocks);
4214 let rank_hashes = hashes.split_off(blocks);
4215 let ends = words;
4216 let stored_len = page.length as usize - body_len;
4219 if ends.last().copied().unwrap_or_default() as usize != stored_len
4220 || ends.windows(2).any(|pair| pair[0] > pair[1])
4221 {
4222 return Err(invalid("global dictionary blocks do not bound the payload"));
4223 }
4224 if offsets.first() != Some(&0) || offsets.windows(2).any(|pair| pair[0] > pair[1]) {
4225 return Err(invalid("global dictionary offsets do not bound the payload"));
4226 }
4227 Vector::external_text(
4228 LogicalType::Varchar,
4229 Arc::new(NativeText {
4230 file,
4231 offsets,
4232 ranks,
4233 rank_at: page.offset + index_len as u64,
4234 rank_hashes,
4235 rank_blocks: (0..rank_blocks).map(|_| OnceLock::new()).collect(),
4236 code_ranks: OnceLock::new(),
4237 payload: page.offset + body_len as u64,
4238 ends,
4239 hashes,
4240 blocks: (0..blocks).map(|_| OnceLock::new()).collect(),
4241 }),
4242 )
4243}
4244
4245fn decode(
4246 ty: &LogicalType,
4247 rows: usize,
4248 bytes: &[u8],
4249 global: Option<Arc<Vector>>,
4250) -> Result<Vector> {
4251 let mut cur = Cursor { bytes, at: 0 };
4252 let codec = cur.u8()?;
4253 let flag = cur.u8()?;
4254 let validity = match flag {
4255 0 => Validity::AllValid,
4256 1 => Validity::AllInvalid,
4257 2 => {
4258 let mask = cur.take(rows.div_ceil(8))?;
4259 Validity::from_iter(rows, |row| mask[row / 8] >> (row % 8) & 1 == 1)
4260 }
4261 _ => return Err(invalid("page validity tag differs")),
4262 };
4263 if codec == 1 {
4264 if ty != &LogicalType::Varchar {
4265 return Err(invalid("dictionary codec belongs to a non-string page"));
4266 }
4267 let count = cur.u32()? as usize;
4268 let payload_len = cur.u32()? as usize;
4269 let offset_bytes = cur.take(
4270 (count + 1)
4271 .checked_mul(4)
4272 .ok_or_else(|| invalid("dictionary offset count overflow"))?,
4273 )?;
4274 let offsets = offset_bytes
4275 .chunks_exact(4)
4276 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4277 .collect::<Vec<_>>();
4278 let payload = cur.take(payload_len)?.to_vec();
4279 if offsets.first() != Some(&0)
4280 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4281 || offsets.windows(2).any(|pair| pair[0] > pair[1])
4282 {
4283 return Err(invalid("dictionary offsets do not bound the payload"));
4284 }
4285 let mut strings = StringColumn::over(Buffer::from_vec(payload));
4286 for pair in offsets.windows(2) {
4287 strings.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4288 }
4289 let mut codes = Vec::with_capacity(rows);
4290 for _ in 0..rows {
4291 codes.push(cur.u32()?);
4292 }
4293 if codes.iter().any(|code| *code as usize >= count) {
4294 return Err(invalid("dictionary code is out of range"));
4295 }
4296 if cur.at != bytes.len() {
4297 return Err(invalid("dictionary page has trailing bytes"));
4298 }
4299 let dictionary = Vector::flat(LogicalType::Varchar, Data::Varlen(strings))?;
4300 return Ok(Vector::dictionary(codes, dictionary)?.with_validity(validity));
4301 }
4302 if codec == 3 || codec == 4 {
4303 let dictionary = global.ok_or_else(|| invalid("global code page has no dictionary"))?;
4304 let codes = if codec == 4 {
4305 let wide = integer::decode(&bytes[cur.at..])?;
4308 if wide.len() != rows {
4309 return Err(invalid("encoded code page holds the wrong number of rows"));
4310 }
4311 wide.into_iter()
4312 .map(|code| u32::try_from(code).map_err(|_| invalid("code is not a code")))
4313 .collect::<Result<Vec<u32>>>()?
4314 } else {
4315 let mut codes = Vec::with_capacity(rows);
4316 for _ in 0..rows {
4317 codes.push(cur.u32()?);
4318 }
4319 if cur.at != bytes.len() {
4320 return Err(invalid("global code page has trailing bytes"));
4321 }
4322 codes
4323 };
4324 let highest = codes.iter().copied().max();
4325 return Ok(Vector::stable_dictionary_validated(codes, dictionary, highest)?
4326 .with_validity(validity));
4327 }
4328 if codec == 5 {
4329 let values = integer::decode(&bytes[cur.at..])?;
4331 if values.len() != rows {
4332 return Err(invalid("cascade page holds the wrong number of rows"));
4333 }
4334 let data = narrowed(ty, values)?;
4335 return Ok(Vector::flat(ty.clone(), data)?.with_validity(validity));
4336 }
4337 if codec == 2 {
4338 let width = u32::from(cur.u8()?);
4339 let base = i128::from_le_bytes(cur.take(16)?.try_into().expect("sixteen bytes"));
4340 let count = cur.u32()? as usize;
4341 let mut words = Vec::with_capacity(count);
4342 for _ in 0..count {
4343 words.push(cur.u64()?);
4344 }
4345 if cur.at != bytes.len() {
4346 return Err(invalid("packed page has trailing bytes"));
4347 }
4348 return Ok(Vector::packed(ty.clone(), words, width, base, rows)?.with_validity(validity));
4349 }
4350 if codec != 0 {
4351 return Err(invalid("page codec is unknown"));
4352 }
4353 let data = match ty {
4354 LogicalType::TinyInt => {
4355 let values = cur.take(rows)?;
4356 Data::Int8(values.iter().map(|item| *item as i8).collect::<Vec<_>>().into())
4357 }
4358 LogicalType::UTinyInt => Data::UInt8(cur.take(rows)?.to_vec().into()),
4359 LogicalType::SmallInt => {
4360 let values =
4361 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4362 Data::Int16(
4363 values
4364 .chunks_exact(2)
4365 .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
4366 .collect::<Vec<_>>()
4367 .into(),
4368 )
4369 }
4370 LogicalType::USmallInt => {
4371 let values =
4372 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
4373 Data::UInt16(
4374 values
4375 .chunks_exact(2)
4376 .map(|item| u16::from_le_bytes(item.try_into().expect("two bytes")))
4377 .collect::<Vec<_>>()
4378 .into(),
4379 )
4380 }
4381 LogicalType::UInteger => {
4382 let values =
4383 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4384 Data::UInt32(
4385 values
4386 .chunks_exact(4)
4387 .map(|item| u32::from_le_bytes(item.try_into().expect("four bytes")))
4388 .collect::<Vec<_>>()
4389 .into(),
4390 )
4391 }
4392 LogicalType::UBigInt => {
4393 let values =
4394 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4395 Data::UInt64(
4396 values
4397 .chunks_exact(8)
4398 .map(|item| u64::from_le_bytes(item.try_into().expect("eight bytes")))
4399 .collect::<Vec<_>>()
4400 .into(),
4401 )
4402 }
4403 LogicalType::Integer | LogicalType::Date => {
4404 let values =
4405 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
4406 Data::Int32(
4407 values
4408 .chunks_exact(4)
4409 .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
4410 .collect::<Vec<_>>()
4411 .into(),
4412 )
4413 }
4414 LogicalType::BigInt | LogicalType::Timestamp => {
4415 let values =
4416 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
4417 Data::Int64(
4418 values
4419 .chunks_exact(8)
4420 .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
4421 .collect::<Vec<_>>()
4422 .into(),
4423 )
4424 }
4425 LogicalType::Boolean => {
4426 let values = cur.take(rows)?;
4427 if values.iter().any(|value| *value > 1) {
4428 return Err(invalid("boolean page has another value"));
4429 }
4430 Data::Bool(values.iter().map(|value| *value == 1).collect::<Vec<_>>().into())
4431 }
4432 LogicalType::Varchar => {
4433 let offset_bytes = cur
4434 .take((rows + 1).checked_mul(4).ok_or_else(|| invalid("offset count overflow"))?)?;
4435 let offsets = offset_bytes
4436 .chunks_exact(4)
4437 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
4438 .collect::<Vec<_>>();
4439 let payload = cur.take(bytes.len() - cur.at)?.to_vec();
4440 if offsets.first() != Some(&0)
4441 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
4442 || offsets.windows(2).any(|pair| pair[0] > pair[1])
4443 {
4444 return Err(invalid("string offsets do not bound the payload"));
4445 }
4446 let mut values = StringColumn::over(Buffer::from_vec(payload));
4447 for pair in offsets.windows(2) {
4448 values.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
4449 }
4450 Data::Varlen(values)
4451 }
4452 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
4453 };
4454 if cur.at != bytes.len() {
4455 return Err(invalid("page has trailing bytes"));
4456 }
4457 Ok(Vector::flat(ty.clone(), data)?.with_validity(validity))
4458}
4459
4460#[cfg(test)]
4461mod tests {
4462 use std::fs;
4463 use std::io::{Seek, SeekFrom, Write};
4464 use std::path::PathBuf;
4465 use std::time::{SystemTime, UNIX_EPOCH};
4466
4467 use rudb_common::Value;
4468 use rudb_common::bounds::Op;
4469
4470 use super::*;
4471
4472 #[test]
4473 fn checksum_matches_fixed_vectors() {
4474 assert_eq!(checksum(b""), 0xef46_db37_51d8_e999);
4475 assert_eq!(checksum(b"a"), 0xd24e_c4f1_a98c_6e5b);
4476 assert_eq!(checksum(b"abc"), 0x44bc_2cf5_ad77_0999);
4477 }
4478
4479 fn path(label: &str) -> PathBuf {
4480 let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
4481 std::env::temp_dir().join(format!("rudb-native-{label}-{}-{stamp}.rdb", std::process::id()))
4482 }
4483
4484 #[test]
4486 fn a_read_at_an_offset_ignores_where_another_thread_left_the_cursor() {
4487 const SPANS: usize = 64;
4488 const SPAN: usize = 512;
4489 let path = path("positional");
4490 let content: Vec<u8> =
4491 (0..SPANS).flat_map(|span| std::iter::repeat_n(span as u8, SPAN)).collect();
4492 fs::write(&path, &content).expect("the file is written");
4493 let file = Arc::new(File::open(&path).expect("the file opens"));
4494 std::thread::scope(|scope| {
4495 for _ in 0..8 {
4496 let file = Arc::clone(&file);
4497 scope.spawn(move || {
4498 for _ in 0..64 {
4499 for span in 0..SPANS {
4500 let mut bytes = [0_u8; SPAN];
4501 read_at(&file, (span * SPAN) as u64, &mut bytes)
4502 .expect("the span reads");
4503 assert!(
4504 bytes.iter().all(|byte| *byte == span as u8),
4505 "span {span} came back as {}",
4506 bytes[0],
4507 );
4508 }
4509 }
4510 });
4511 }
4512 });
4513 let mut past = [0_u8; SPAN];
4514 let end = (SPANS * SPAN) as u64;
4515 let error = read_at(&file, end, &mut past).expect_err("a read past the end is refused");
4516 assert!(error.message().contains("ends before its declared length"), "{error}");
4517 drop(file);
4518 let _ = fs::remove_file(&path);
4519 }
4520
4521 #[test]
4527 fn a_writer_puts_a_page_where_it_said_it_did_wherever_the_cursor_has_got_to() {
4528 let path = path("cursor");
4529 let mut writer = Writer::create(
4530 &path,
4531 "items",
4532 vec![
4533 Field::required("id", LogicalType::Integer),
4534 Field::new("text", LogicalType::Varchar),
4535 ],
4536 )
4537 .expect("new file");
4538 writer.append(&sample()).expect("first part");
4539 writer.file.seek(SeekFrom::Start(0)).expect("the cursor goes back to the header");
4540 writer.append(&sample()).expect("second part");
4541 writer.file.seek(SeekFrom::Start(1)).expect("and somewhere useless again");
4542 writer.finish().expect("commit");
4543 let reader = Reader::open(&path).expect("reopen from disk");
4544 assert_eq!(reader.table().rows(), 6);
4545 let ids = reader.read(0, &[0]).expect("the integer page reads back");
4546 assert_eq!(ids.value_at(0, 0), Value::Integer(4));
4547 assert_eq!(ids.value_at(2, 0), Value::Integer(-2));
4548 let text = reader.read(1, &[1]).expect("the text page reads back");
4549 assert_eq!(text.value_at(1, 0), Value::Null);
4550 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4551 let end = reader.table().stripes().iter().flat_map(|stripe| {
4554 stripe
4555 .pages
4556 .iter()
4557 .map(|page| page.offset + u64::from(page.length))
4558 .chain(std::iter::once(stripe.index.offset + u64::from(stripe.index.length)))
4559 });
4560 let last = end.fold(HEADER, u64::max);
4561 let directory = fs::metadata(&path).expect("the file is there").len();
4562 assert!(last <= directory, "a page runs to {last} in a file of {directory} bytes");
4563 fs::remove_file(path).expect("remove scratch file");
4564 }
4565
4566 fn sample() -> Chunk {
4567 Chunk::new(vec![
4568 Vector::from_values(
4569 LogicalType::Integer,
4570 &[Value::Integer(4), Value::Integer(9), Value::Integer(-2)],
4571 )
4572 .expect("integers"),
4573 Vector::from_values(
4574 LogicalType::Varchar,
4575 &[
4576 Value::Varchar("alpha".into()),
4577 Value::Null,
4578 Value::Varchar("long text after a slash".into()),
4579 ],
4580 )
4581 .expect("strings"),
4582 ])
4583 .expect("matching rows")
4584 }
4585
4586 fn sample_ids() -> Chunk {
4587 Chunk::new(vec![
4588 Vector::flat(LogicalType::Integer, Data::Int32(vec![7, 8, 9].into()))
4589 .expect("integers"),
4590 ])
4591 .expect("one column")
4592 }
4593
4594 #[test]
4595 fn committed_file_reopens_and_reads_only_requested_columns() {
4596 let path = path("reopen");
4597 let mut writer = Writer::create(
4598 &path,
4599 "items",
4600 vec![
4601 Field::required("id", LogicalType::Integer),
4602 Field::new("text", LogicalType::Varchar),
4603 ],
4604 )
4605 .expect("new file");
4606 writer.append(&sample()).expect("first part");
4607 writer.append(&sample()).expect("second part");
4608 writer.finish().expect("commit");
4609 let reader = Reader::open(&path).expect("reopen from disk");
4610 assert_eq!(reader.table().rows(), 6);
4611 assert_eq!(reader.table().stripes().len(), 1);
4614 assert_eq!(reader.parts(), 2);
4615 assert_eq!(reader.part_rows(0), 3);
4616 assert_eq!(reader.part_rows(1), 3);
4617 let text = reader.read(1, &[1]).expect("only text page");
4618 assert_eq!(text.width(), 1);
4619 assert_eq!(text.value_at(1, 0), Value::Null);
4620 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4621 let sparse = reader.read_sparse(1, &[1]).expect("one part without its whole page");
4622 assert_eq!(sparse.width(), 1);
4623 assert_eq!(sparse.value_at(1, 0), Value::Null);
4624 assert_eq!(sparse.value_at(2, 0), Value::Varchar("long text after a slash".into()));
4625 assert!(!reader.skips_codes(0, 1, &[0]).expect("alpha is in the stripe"));
4626 assert!(!reader.skips_codes(0, 1, &[2]).expect("long text is in the stripe"));
4627 assert!(reader.skips_codes(0, 1, &[3]).expect("unknown code is absent"));
4628 let count = reader.read(0, &[]).expect("no page is needed for count");
4629 assert_eq!(count.len(), 3);
4630 assert!(reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }]));
4631 assert!(!reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(0) }]));
4632 let integers = reader.top_frequencies(0, 1).expect("valid integer synopsis").expect("kept");
4633 assert_eq!(
4634 integers,
4635 vec![(Value::Integer(-2), 2), (Value::Integer(4), 2), (Value::Integer(9), 2),]
4636 );
4637 let strings = reader.top_frequencies(1, 1).expect("valid string synopsis").expect("kept");
4638 assert_eq!(strings.len(), 3);
4639 assert!(strings.contains(&(Value::Null, 2)));
4640 assert!(strings.contains(&(Value::Varchar("alpha".into()), 2)));
4641 assert!(strings.contains(&(Value::Varchar("long text after a slash".into()), 2)));
4642 fs::remove_file(path).expect("remove scratch file");
4643 }
4644
4645 #[test]
4653 fn runs_handed_over_out_of_order_still_read_back_in_source_order() {
4654 let path = path("interleaved-runs");
4655 let mut writer =
4656 Writer::create(&path, "interleaved", vec![Field::new("v", LogicalType::BigInt)])
4657 .expect("new file");
4658 for morsel in [2_u64, 0, 3, 1] {
4659 let parts = (0..4_u64)
4660 .map(|chunk| {
4661 let first = i64::try_from(morsel * 32 + chunk * 8).expect("small");
4662 let values =
4663 (0..8_i64).map(|row| Value::BigInt(first + row)).collect::<Vec<_>>();
4664 let column =
4665 Vector::from_values(LogicalType::BigInt, &values).expect("a column");
4666 ((morsel, chunk), Chunk::new(vec![column]).expect("one column"))
4667 })
4668 .collect::<Vec<_>>();
4669 writer.append_stripe(parts).expect("a stripe");
4670 }
4671 writer.finish().expect("commit");
4672
4673 let reader = Reader::open(&path).expect("valid directory");
4674 assert_eq!(reader.table().stripes().len(), 4, "a run is a stripe of its own");
4675 assert_eq!(reader.table().rows(), 128);
4676 for part in 0..16_usize {
4677 let read = reader.read(part, &[0]).expect("a part back");
4678 for row in 0..8_usize {
4679 let want = i64::try_from(part * 8 + row).expect("small");
4680 assert_eq!(read.value_at(row, 0), Value::BigInt(want), "part {part} row {row}");
4681 }
4682 }
4683 fs::remove_file(path).expect("remove scratch file");
4684 }
4685
4686 #[test]
4689 fn runs_that_overlap_each_other_are_refused_at_commit() {
4690 let path = path("overlapping-runs");
4691 let mut writer =
4692 Writer::create(&path, "overlapping", vec![Field::new("v", LogicalType::BigInt)])
4693 .expect("new file");
4694 let one = |order: (u64, u64)| {
4695 let column =
4696 Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)]).expect("a column");
4697 (order, Chunk::new(vec![column]).expect("one column"))
4698 };
4699 writer.append_stripe(vec![one((0, 0)), one((0, 2))]).expect("a stripe");
4702 writer.append_stripe(vec![one((0, 1))]).expect("a stripe");
4703 let error = writer.finish().expect_err("the runs overlap");
4704 assert!(error.message().contains("source order"), "{error}");
4705 fs::remove_file(path).expect("remove scratch file");
4706 }
4707
4708 #[test]
4711 fn a_run_longer_than_a_stripe_is_refused() {
4712 let path = path("overlong-run");
4713 let mut writer =
4714 Writer::create(&path, "overlong", vec![Field::new("v", LogicalType::BigInt)])
4715 .expect("new file");
4716 let parts = (0..=STRIPE_PARTS)
4717 .map(|at| {
4718 let column = Vector::from_values(LogicalType::BigInt, &[Value::BigInt(1)])
4719 .expect("a column");
4720 let chunk = Chunk::new(vec![column]).expect("one column");
4721 ((0, u64::try_from(at).expect("small")), chunk)
4722 })
4723 .collect::<Vec<_>>();
4724 let error = writer.append_stripe(parts).expect_err("one part too many");
4725 assert!(error.message().contains("more parts than it holds"), "{error}");
4726 fs::remove_file(path).expect("remove scratch file");
4727 }
4728
4729 #[test]
4735 fn parts_past_the_stripe_bound_start_a_new_stripe() {
4736 let path = path("stripe-bound");
4737 let mut writer = Writer::create(
4738 &path,
4739 "items",
4740 vec![
4741 Field::required("id", LogicalType::Integer),
4742 Field::new("text", LogicalType::Varchar),
4743 ],
4744 )
4745 .expect("new file");
4746 let parts = STRIPE_PARTS * 2 + 3;
4747 for part in 0..parts {
4748 let id = part as i32;
4749 let chunk = Chunk::new(vec![
4750 Vector::from_values(
4751 LogicalType::Integer,
4752 &[Value::Integer(id), Value::Integer(-id)],
4753 )
4754 .expect("integers"),
4755 Vector::from_values(
4756 LogicalType::Varchar,
4757 &[Value::Varchar(format!("value {part}")), Value::Null],
4758 )
4759 .expect("strings"),
4760 ])
4761 .expect("matching rows");
4762 writer.append(&chunk).expect("one part");
4763 }
4764 writer.finish().expect("commit");
4765
4766 let reader = Reader::open(&path).expect("reopen from disk");
4767 assert_eq!(reader.parts(), parts);
4768 assert_eq!(reader.table().rows(), parts * 2);
4769 assert_eq!(reader.table().stripes().len(), parts.div_ceil(STRIPE_PARTS));
4770 assert_eq!(reader.table().stripes()[0].parts(), STRIPE_PARTS);
4771 assert_eq!(reader.table().stripes()[0].rows(), STRIPE_PARTS * 2);
4772 assert_eq!(reader.table().stripes()[2].parts(), 3);
4773 for part in (0..parts).rev() {
4776 let dense = reader.read(part, &[0, 1]).expect("a whole page read");
4777 let sparse = reader.read_sparse(part, &[0, 1]).expect("one part read");
4778 for chunk in [&dense, &sparse] {
4779 assert_eq!(chunk.len(), 2, "part {part} has its own row count");
4780 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4781 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
4782 assert_eq!(chunk.value_at(0, 1), Value::Varchar(format!("value {part}")));
4783 assert_eq!(chunk.value_at(1, 1), Value::Null);
4784 }
4785 }
4786 let above = [Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }];
4789 assert!(reader.skips(0, &above), "the first stripe stops at 63");
4790 assert!(!reader.skips(STRIPE_PARTS * 2, &above), "the third stripe reaches 130");
4791 fs::remove_file(path).expect("remove scratch file");
4792 }
4793
4794 fn scattered(n: i64) -> i64 {
4796 n.wrapping_mul(-7_046_029_254_386_353_131)
4797 }
4798
4799 #[test]
4805 fn a_part_is_skipped_when_its_sieve_does_not_hold_the_constant() {
4806 let path = path("sieve-skip");
4807 let mut writer =
4808 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
4809 .expect("new file");
4810 let parts = STRIPE_PARTS + 3;
4811 let per_part = 128;
4815 for part in 0..parts {
4816 let held: Vec<Value> = (0..per_part)
4817 .map(|row| Value::BigInt(scattered((part * per_part + row) as i64)))
4818 .collect();
4819 let chunk =
4820 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
4821 .expect("one column");
4822 writer.append(&chunk).expect("one part");
4823 }
4824 writer.finish().expect("commit");
4825
4826 let reader = Reader::open(&path).expect("reopen from disk");
4827 let probe = |value: i64| Probe {
4828 column: 0,
4829 op: Op::Equal,
4830 value: Bound::Int(i128::from(scattered(value))),
4831 };
4832 for wanted in [0_i64, (per_part + 1) as i64, (parts * per_part - 1) as i64] {
4833 let tests = [probe(wanted)];
4834 let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &tests)).collect();
4835 let home = wanted as usize / per_part;
4836 assert!(kept.contains(&home), "the part holding {wanted} is read");
4837 assert!(kept.len() <= 2, "{wanted} keeps {kept:?}, which is more than one stray part");
4841 }
4842 let absent = [probe((parts * per_part) as i64 + 1)];
4843 let kept = (0..parts).filter(|&part| !reader.skips(part, &absent)).count();
4844 assert!(kept <= 1, "{kept} parts of {parts} kept a value no part holds");
4845 let tests = [probe(0)];
4848 assert!(
4849 reader.table().stripes().iter().all(|stripe| !stripe.zone.skips(&tests)),
4850 "the bounds rule out no stripe at all"
4851 );
4852 fs::remove_file(path).expect("remove scratch file");
4853 }
4854
4855 #[test]
4865 fn a_sieve_larger_than_the_part_it_indexes_is_not_written() {
4866 let path = path("sieve-pays");
4867 let fields = vec![
4868 Field::required("spread", LogicalType::BigInt),
4869 Field::required("repeated", LogicalType::BigInt),
4870 ];
4871 let mut writer = Writer::create(&path, "hits", fields).expect("new file");
4872 let parts = 3;
4873 let per_part = 1024;
4874 for part in 0..parts {
4875 let base = (part * per_part) as i64;
4876 let spread: Vec<Value> =
4877 (0..per_part).map(|row| Value::BigInt(scattered(base + row as i64))).collect();
4878 let repeated: Vec<Value> =
4879 (0..per_part).map(|row| Value::BigInt(scattered((row / 256) as i64))).collect();
4880 let chunk = Chunk::new(vec![
4881 Vector::from_values(LogicalType::BigInt, &spread).expect("numbers"),
4882 Vector::from_values(LogicalType::BigInt, &repeated).expect("numbers"),
4883 ])
4884 .expect("two columns");
4885 writer.append(&chunk).expect("one part");
4886 }
4887 writer.finish().expect("commit");
4888
4889 let reader = Reader::open(&path).expect("reopen from disk");
4890 let layout = reader.layout();
4891 let spread = &layout.columns[0];
4892 let repeated = &layout.columns[1];
4893 assert!(spread.sieves > 0, "a column whose parts are worth a filter keeps one");
4894 assert_eq!(
4895 repeated.sieves, 0,
4896 "a column whose filter costs more than its parts keeps none"
4897 );
4898 for column in &layout.columns {
4901 assert!(
4902 column.sieves < column.pages,
4903 "{} spends {} on sieves over {} of data",
4904 column.name,
4905 column.sieves,
4906 column.pages
4907 );
4908 }
4909 let absent = [Probe {
4911 column: 0,
4912 op: Op::Equal,
4913 value: Bound::Int(i128::from(scattered((parts * per_part) as i64 + 1))),
4914 }];
4915 assert!((0..parts).all(|part| reader.skips(part, &absent)), "no part holds it");
4916 fs::remove_file(path).expect("remove scratch file");
4917 }
4918
4919 #[test]
4925 fn a_damaged_sieve_page_is_read_through_rather_than_refused() {
4926 let path = path("sieve-damaged");
4927 let mut writer =
4928 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
4929 .expect("new file");
4930 let rows = 128;
4931 let held: Vec<Value> = (0..rows).map(|row| Value::BigInt(scattered(row))).collect();
4932 let chunk =
4933 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
4934 .expect("one column");
4935 writer.append(&chunk).expect("one part");
4936 writer.finish().expect("commit");
4937
4938 let page =
4939 Reader::open(&path).expect("reopen").table.stripes[0].sieves[0].expect("a sieve page");
4940 let mut file = OpenOptions::new().write(true).open(&path).expect("open the sieve page");
4941 file.seek(SeekFrom::Start(page.offset + u64::from(page.length) - 1)).expect("seek");
4942 file.write_all(&[0xff]).expect("damage one byte");
4943 drop(file);
4944
4945 let reader = Reader::open(&path).expect("reopen the damaged file");
4946 let absent =
4947 [Probe { column: 0, op: Op::Equal, value: Bound::Int(i128::from(scattered(99))) }];
4948 assert!(!reader.skips(0, &absent), "a sieve that cannot be read skips nothing");
4949 assert_eq!(
4950 reader.read(0, &[0]).expect("the rows are untouched").len(),
4951 usize::try_from(rows).expect("a small count")
4952 );
4953 fs::remove_file(path).expect("remove scratch file");
4954 }
4955
4956 #[test]
4967 fn workers_that_want_the_same_stripe_read_it_once() {
4968 let path = path("single-flight");
4969 let mut writer =
4970 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
4971 .expect("new file");
4972 for part in 0..STRIPE_PARTS {
4973 let id = part as i32;
4974 let chunk = Chunk::new(vec![
4975 Vector::from_values(
4976 LogicalType::Integer,
4977 &[Value::Integer(id), Value::Integer(-id)],
4978 )
4979 .expect("integers"),
4980 ])
4981 .expect("matching rows");
4982 writer.append(&chunk).expect("one part");
4983 }
4984 writer.finish().expect("commit");
4985
4986 let reader = Reader::open(&path).expect("reopen from disk");
4987 assert_eq!(reader.table().stripes().len(), 1, "one stripe is the point of the test");
4988 let barrier = std::sync::Barrier::new(8);
4989 std::thread::scope(|scope| {
4990 for worker in 0..8 {
4991 let reader = &reader;
4992 let barrier = &barrier;
4993 scope.spawn(move || {
4994 barrier.wait();
4995 for part in (worker..STRIPE_PARTS).step_by(8) {
4996 let chunk = reader.read(part, &[0]).expect("a whole page read");
4997 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
4998 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
4999 }
5000 });
5001 }
5002 });
5003 assert_eq!(reader.pages.load(Atomic::Relaxed), 1, "one stripe, one page read, whoever won");
5004 fs::remove_file(path).expect("remove scratch file");
5005 }
5006
5007 #[test]
5020 fn opening_costs_the_same_over_a_thousand_times_the_rows() {
5021 let opened = |label: &str, rows_per_part: i32| {
5022 let path = path(label);
5023 let mut writer =
5024 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5025 .expect("new file");
5026 for part in 0..STRIPE_PARTS * 3 {
5027 let values = (0..rows_per_part)
5031 .map(|row| {
5032 Value::Integer((part as i32 * rows_per_part + row).wrapping_mul(2_654_435))
5033 })
5034 .collect::<Vec<_>>();
5035 let chunk = Chunk::new(vec![
5036 Vector::from_values(LogicalType::Integer, &values).expect("integers"),
5037 ])
5038 .expect("matching rows");
5039 writer.append(&chunk).expect("one part");
5040 }
5041 writer.finish().expect("commit");
5042 let reader = Reader::open(&path).expect("reopen from disk");
5043 let size = fs::metadata(&path).expect("the file is there").len();
5044 let out = (reader.reads(), reader.table().stripes().len(), size);
5045 fs::remove_file(path).expect("remove scratch file");
5046 out
5047 };
5048
5049 let (thin, thin_stripes, thin_size) = opened("open-thin", 1);
5050 let (fat, fat_stripes, fat_size) = opened("open-fat", 1000);
5051 assert_eq!(
5052 thin_stripes, fat_stripes,
5053 "the same stripe count is what makes this a fair ask"
5054 );
5055 assert!(
5056 fat_size > thin_size * 50,
5057 "the fat file has to actually be larger, and it is {fat_size} against {thin_size}"
5058 );
5059
5060 assert_eq!(thin.opening.reads, fat.opening.reads, "the same reads either way");
5061 assert_eq!(thin.pages, 0, "opening read a page");
5062 assert_eq!(fat.pages, 0, "opening read a page");
5063 assert_eq!(thin.indexes, 0, "opening read an index");
5064 assert_eq!(fat.indexes, 0, "opening read an index");
5065 assert!(
5068 fat.opening.bytes < thin.opening.bytes * 2,
5069 "opening the thin file read {} bytes and the fat one read {}",
5070 thin.opening.bytes,
5071 fat.opening.bytes
5072 );
5073 }
5074
5075 #[test]
5083 fn two_opens_of_one_file_cost_the_same_and_the_second_is_not_cheaper() {
5084 let path = path("open-twice");
5085 let mut writer =
5086 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5087 .expect("new file");
5088 for part in 0..STRIPE_PARTS * 3 {
5089 let chunk = Chunk::new(vec![
5090 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
5091 .expect("integers"),
5092 ])
5093 .expect("matching rows");
5094 writer.append(&chunk).expect("one part");
5095 }
5096 writer.finish().expect("commit");
5097
5098 let first = Reader::open(&path).expect("open");
5099 for part in 0..first.parts() {
5102 first.read(part, &[0]).expect("a part");
5103 }
5104 assert!(first.reads().pages > 0, "the scan has to have read something");
5105 let second = Reader::open(&path).expect("open again");
5106
5107 assert_eq!(first.reads().opening, second.reads().opening);
5108 assert_eq!(
5109 second.reads().pages,
5110 0,
5111 "the second open read a page off the back of the first"
5112 );
5113 assert_eq!(second.reads().indexes, 0, "the second open read an index it inherited");
5114 fs::remove_file(path).expect("remove scratch file");
5115 }
5116
5117 #[test]
5125 fn an_index_is_read_once_per_stripe_however_often_the_page_is_evicted() {
5126 let path = path("index-cache");
5127 let mut writer =
5128 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5129 .expect("new file");
5130 let parts = STRIPE_PARTS * (CACHED_STRIPES_PER_COLUMN + 2);
5131 for part in 0..parts {
5132 let id = part as i32;
5133 let chunk = Chunk::new(vec![
5134 Vector::from_values(LogicalType::Integer, &[Value::Integer(id)]).expect("integers"),
5135 ])
5136 .expect("matching rows");
5137 writer.append(&chunk).expect("one part");
5138 }
5139 writer.finish().expect("commit");
5140
5141 let reader = Reader::open(&path).expect("reopen from disk");
5142 let stripes = reader.table().stripes().len();
5143 assert!(stripes > CACHED_STRIPES_PER_COLUMN, "the page cache has to be too small for this");
5144 for _ in 0..2 {
5146 for part in 0..parts {
5147 let chunk = reader.read(part, &[0]).expect("a part");
5148 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5149 }
5150 }
5151 assert_eq!(reader.indexes.load(Atomic::Relaxed), stripes, "one index read per stripe");
5152 assert!(
5153 reader.pages.load(Atomic::Relaxed) > stripes,
5154 "the pages are the ones that get read again, which is what makes the index count mean \
5155 something"
5156 );
5157 fs::remove_file(path).expect("remove scratch file");
5158 }
5159
5160 #[test]
5169 fn a_worker_per_stripe_reads_its_page_once_when_the_cache_was_told_to_expect_it() {
5170 let workers = CACHED_STRIPES_PER_COLUMN + 4;
5171 let path = path("stripe-per-worker");
5172 let mut writer =
5173 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5174 .expect("new file");
5175 for part in 0..STRIPE_PARTS * workers {
5176 let chunk = Chunk::new(vec![
5177 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
5178 .expect("integers"),
5179 ])
5180 .expect("matching rows");
5181 writer.append(&chunk).expect("one part");
5182 }
5183 writer.finish().expect("commit");
5184
5185 let read = |told: bool| {
5186 let reader = Reader::open(&path).expect("reopen from disk");
5187 assert_eq!(reader.table().stripes().len(), workers, "a stripe per worker");
5188 if told {
5189 reader.keep_stripes(workers);
5190 }
5191 let barrier = std::sync::Barrier::new(workers);
5192 std::thread::scope(|scope| {
5193 for (worker, run) in reader.stripe_parts().into_iter().enumerate() {
5194 let reader = &reader;
5195 let barrier = &barrier;
5196 scope.spawn(move || {
5197 for part in run {
5198 barrier.wait();
5199 let chunk = reader.read(part, &[0]).expect("a part of my own stripe");
5200 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
5201 }
5202 assert!(worker < workers);
5203 });
5204 }
5205 });
5206 reader.pages.load(Atomic::Relaxed)
5207 };
5208
5209 assert_eq!(read(true), workers, "one page read per stripe and no more");
5210 assert!(read(false) > workers, "a cache that small is read again on every part");
5211 fs::remove_file(path).expect("remove scratch file");
5212 }
5213
5214 #[test]
5219 fn a_damaged_index_page_is_an_error() {
5220 let path = path("damaged-index");
5221 let mut writer =
5222 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
5223 .expect("new file");
5224 writer.append(&sample_ids()).expect("first part");
5225 writer.append(&sample_ids()).expect("second part");
5226 writer.finish().expect("commit");
5227
5228 let reader = Reader::open(&path).expect("valid directory");
5229 let index = reader.table.stripes[0].index;
5230 let mut byte = [0; 1];
5231 read_at(&reader.file, index.offset, &mut byte).expect("the first part length");
5232 let mut file = OpenOptions::new().write(true).open(&path).expect("open index page");
5233 file.seek(SeekFrom::Start(index.offset)).expect("index start");
5234 file.write_all(&[!byte[0]]).expect("damage the first part length");
5235 let error = reader.read(1, &[0]).expect_err("a damaged index must not be used");
5236 assert!(error.message().contains("index page section checksum differs"), "{error}");
5237 fs::remove_file(path).expect("remove scratch file");
5238 }
5239
5240 #[test]
5247 fn every_integer_width_round_trips_through_a_page() {
5248 let path = path("integer-widths");
5249 let columns = [
5250 (LogicalType::TinyInt, vec![Value::TinyInt(i8::MIN), Value::TinyInt(i8::MAX)]),
5251 (LogicalType::UTinyInt, vec![Value::UTinyInt(0), Value::UTinyInt(u8::MAX)]),
5252 (LogicalType::SmallInt, vec![Value::SmallInt(i16::MIN), Value::SmallInt(i16::MAX)]),
5253 (LogicalType::USmallInt, vec![Value::USmallInt(0), Value::USmallInt(u16::MAX)]),
5254 (LogicalType::Integer, vec![Value::Integer(i32::MIN), Value::Integer(i32::MAX)]),
5255 (LogicalType::UInteger, vec![Value::UInteger(0), Value::UInteger(u32::MAX)]),
5256 (LogicalType::BigInt, vec![Value::BigInt(i64::MIN), Value::BigInt(i64::MAX)]),
5257 (LogicalType::UBigInt, vec![Value::UBigInt(0), Value::UBigInt(u64::MAX)]),
5258 ];
5259 let fields = columns
5260 .iter()
5261 .enumerate()
5262 .map(|(at, (ty, _))| Field::required(format!("c{at}"), ty.clone()))
5263 .collect::<Vec<_>>();
5264 let vectors = columns
5265 .iter()
5266 .map(|(ty, values)| Vector::from_values(ty.clone(), values).expect("a vector"))
5267 .collect::<Vec<_>>();
5268 let mut writer = Writer::create(&path, "widths", fields).expect("new file");
5269 writer.append(&Chunk::new(vectors).expect("matching rows")).expect("one stripe");
5270 writer.finish().expect("commit");
5271
5272 let reader = Reader::open(&path).expect("reopen from disk");
5273 let wanted = (0..columns.len()).collect::<Vec<_>>();
5274 let read = reader.read(0, &wanted).expect("every column");
5275 assert_eq!(read.len(), 2);
5276 for (at, (ty, values)) in columns.iter().enumerate() {
5278 assert_eq!(read.value_at(0, at), values[0], "the low end of {ty}");
5279 assert_eq!(read.value_at(1, at), values[1], "the high end of {ty}");
5280 }
5281 fs::remove_file(path).expect("remove scratch file");
5282 }
5283
5284 #[test]
5285 fn numeric_frequency_candidates_keep_bounded_row_ordinals() {
5286 let path = path("frequency-ordinals");
5287 let mut writer =
5288 Writer::create(&path, "items", vec![Field::required("id", LogicalType::BigInt)])
5289 .expect("new file");
5290 let mut values = Vec::new();
5291 for leader in 0..10_i64 {
5292 values.extend(std::iter::repeat_n(leader, 100));
5293 }
5294 values.extend(1_000_i64..41_000);
5295 for part in values.chunks(1_024) {
5296 let vector = Vector::flat(LogicalType::BigInt, Data::Int64(part.to_vec().into()))
5297 .expect("big integers");
5298 writer.append(&Chunk::new(vec![vector]).expect("one column")).expect("one stripe");
5299 }
5300 writer.finish().expect("commit");
5301
5302 let reader = Reader::open(&path).expect("reopen from disk");
5303 let occurrences =
5304 reader.frequency_occurrences(0).expect("valid metadata").expect("bounded ordinals");
5305 assert!(occurrences.omitted_max < 100);
5306 assert!(occurrences.ordinals.len() <= FREQUENCY_ORDINALS);
5307 assert!(occurrences.ordinals.windows(2).all(|pair| pair[0] < pair[1]));
5308 assert_eq!(&occurrences.ordinals[..1_000], &(0_u64..1_000).collect::<Vec<_>>());
5309 fs::remove_file(path).expect("remove scratch file");
5310 }
5311
5312 #[test]
5318 fn a_file_from_another_format_says_which_format_it_is() {
5319 let older = path("older-format");
5320 let mut writer =
5321 Writer::create(&older, "items", vec![Field::new("id", LogicalType::Integer)])
5322 .expect("new file");
5323 let chunk = Chunk::new(vec![
5324 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
5325 .expect("integers"),
5326 ])
5327 .expect("chunk");
5328 writer.append(&chunk).expect("page written");
5329 writer.finish().expect("commit");
5330
5331 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
5332 file.seek(SeekFrom::Start(8)).expect("the version follows the magic");
5333 file.write_all(&(FORMAT - 1).to_le_bytes()).expect("write an older version");
5334 drop(file);
5335 let complaint = Reader::open(&older).expect_err("an older format is refused").to_string();
5336 assert!(complaint.contains(&format!("format {}", FORMAT - 1)), "{complaint}");
5337 assert!(complaint.contains(&format!("format {FORMAT}")), "{complaint}");
5338
5339 let mut file = OpenOptions::new().write(true).open(&older).expect("open for the header");
5340 file.seek(SeekFrom::Start(0)).expect("the magic is first");
5341 file.write_all(b"NOTRUDB!").expect("write another engine's magic");
5342 drop(file);
5343 let complaint = Reader::open(&older).expect_err("a foreign file is refused").to_string();
5344 assert!(complaint.contains("magic"), "{complaint}");
5345 assert!(!complaint.contains("format"), "a version has nothing to do with it: {complaint}");
5346 fs::remove_file(older).expect("remove scratch file");
5347 }
5348
5349 #[test]
5350 fn an_unfinished_or_damaged_file_does_not_answer_with_partial_rows() {
5351 let unfinished = path("unfinished");
5352 let mut writer =
5353 Writer::create(&unfinished, "items", vec![Field::new("id", LogicalType::Integer)])
5354 .expect("new file");
5355 let chunk = Chunk::new(vec![
5356 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
5357 .expect("integers"),
5358 ])
5359 .expect("chunk");
5360 writer.append(&chunk).expect("page written");
5361 drop(writer);
5362 assert!(Reader::open(&unfinished).is_err(), "no directory was committed");
5363 fs::remove_file(unfinished).expect("remove scratch file");
5364
5365 let damaged = path("damaged");
5366 let mut writer =
5367 Writer::create(&damaged, "items", vec![Field::new("id", LogicalType::Integer)])
5368 .expect("new file");
5369 writer.append(&chunk).expect("page written");
5370 writer.finish().expect("commit");
5371 let reader = Reader::open(&damaged).expect("valid directory");
5372 let mut file =
5373 OpenOptions::new().write(true).open(&damaged).expect("open for a damaged page");
5374 file.seek(SeekFrom::Start(HEADER + 1)).expect("inside first page");
5375 file.write_all(&[255]).expect("damage one byte");
5376 assert!(reader.read(0, &[0]).is_err(), "page checksum rejects corruption");
5377 fs::remove_file(damaged).expect("remove scratch file");
5378 }
5379
5380 #[test]
5381 fn damaged_lazy_dictionary_payload_is_an_error() {
5382 let path = path("damaged-dictionary");
5383 let mut writer = Writer::create(
5384 &path,
5385 "items",
5386 vec![
5387 Field::required("id", LogicalType::Integer),
5388 Field::new("text", LogicalType::Varchar),
5389 ],
5390 )
5391 .expect("new file");
5392 writer.append(&sample()).expect("stripe written");
5393 writer.finish().expect("commit");
5394
5395 let reader = Reader::open(&path).expect("valid directory");
5396 let dictionary = reader.table.dictionaries[1].expect("string dictionary page");
5397 let mut header = [0; 12];
5400 read_at(&reader.file, dictionary.offset, &mut header).expect("dictionary header");
5401 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
5402 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
5403 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
5404 let index_len =
5405 12 + (count + 1) * 4 + (blocks * 2 + rank_blocks) * 8 + count * RANK_ENTRY as u64;
5406 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5407 file.seek(SeekFrom::Start(dictionary.offset + index_len))
5408 .expect("inside dictionary payload");
5409 file.write_all(&[255]).expect("damage dictionary payload");
5410
5411 let chunk = reader.read(0, &[1]).expect("code page and dictionary index remain valid");
5412 let error =
5413 chunk.validate_external().expect_err("payload corruption must reach the caller");
5414 assert!(error.message().contains("payload checksum differs"), "{error}");
5415 fs::remove_file(path).expect("remove scratch file");
5416 }
5417
5418 #[test]
5425 fn a_dictionary_over_many_blocks_checks_every_block_of_it() {
5426 let path = path("dictionary-blocks");
5427 let value =
5428 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5429 let parts = 30;
5430 let per_part = 1000;
5431 let mut writer =
5432 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5433 .expect("new file");
5434 for part in 0..parts {
5435 let values = (0..per_part)
5436 .map(|row| Value::Varchar(value(part * per_part + row)))
5437 .collect::<Vec<_>>();
5438 let chunk = Chunk::new(vec![
5439 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5440 ])
5441 .expect("matching rows");
5442 writer.append(&chunk).expect("a part");
5443 }
5444 writer.finish().expect("commit");
5445
5446 let reader = Reader::open(&path).expect("reopen from disk");
5447 let dictionary = reader.table.dictionaries[0].expect("string dictionary page");
5448 assert!(
5449 parts * per_part > TEXT_PAYLOAD_VALUES * 4,
5450 "the dictionary has to be several blocks for this to be testing anything"
5451 );
5452 for part in [0, parts - 1] {
5453 let chunk = reader.read(part, &[0]).expect("a part");
5454 chunk.validate_external().expect("every payload block checks out");
5455 assert_eq!(chunk.value_at(0, 0), Value::Varchar(value(part * per_part)));
5456 }
5457
5458 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5459 file.seek(SeekFrom::Start(dictionary.offset + u64::from(dictionary.length) - 4))
5460 .expect("the last bytes of the page are payload");
5461 file.write_all(&[255]).expect("damage the last payload block");
5462 let reader = Reader::open(&path).expect("the directory and the index are untouched");
5463 let chunk = reader.read(parts - 1, &[0]).expect("the code page remains valid");
5464 let error = chunk.validate_external().expect_err("the damage must reach the caller");
5465 assert!(error.message().contains("payload checksum differs"), "{error}");
5466 fs::remove_file(path).expect("remove scratch file");
5467 }
5468
5469 #[test]
5481 fn a_global_dictionary_is_opened_once_however_many_workers_ask_at_once() {
5482 let path = path("dictionary-once");
5483 let parts = 8;
5484 let per_part = 500;
5485 let value =
5486 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
5487 let mut writer =
5488 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
5489 .expect("new file");
5490 for part in 0..parts {
5491 let values = (0..per_part)
5492 .map(|row| Value::Varchar(value(part * per_part + row)))
5493 .collect::<Vec<_>>();
5494 let chunk = Chunk::new(vec![
5495 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
5496 ])
5497 .expect("matching rows");
5498 writer.append(&chunk).expect("a part");
5499 }
5500 writer.finish().expect("commit");
5501
5502 let reader = Reader::open(&path).expect("reopen from disk");
5503 assert!(reader.table.dictionaries[0].is_some(), "the column has to have one to share");
5504 assert_eq!(reader.reads().dictionaries, 0, "opening the file does not open a dictionary");
5505
5506 let workers = 16;
5507 let gate = std::sync::Barrier::new(workers);
5508 std::thread::scope(|scope| {
5509 for worker in 0..workers {
5510 let reader = reader.clone();
5511 let gate = &gate;
5512 scope.spawn(move || {
5513 gate.wait();
5514 let chunk = reader.read(worker % parts, &[0]).expect("a part");
5515 assert_eq!(
5516 chunk.value_at(0, 0),
5517 Value::Varchar(value((worker % parts) * per_part))
5518 );
5519 });
5520 }
5521 });
5522
5523 assert_eq!(reader.reads().dictionaries, 1, "sixteen workers, one dictionary, one open");
5524 fs::remove_file(path).expect("remove scratch file");
5525 }
5526
5527 #[test]
5532 fn a_damaged_sorted_order_is_an_error() {
5533 let path = path("damaged-order");
5534 let mut writer = Writer::create(
5535 &path,
5536 "items",
5537 vec![
5538 Field::required("id", LogicalType::Integer),
5539 Field::new("text", LogicalType::Varchar),
5540 ],
5541 )
5542 .expect("new file");
5543 writer.append(&sample()).expect("stripe written");
5544 writer.finish().expect("commit");
5545
5546 let reader = Reader::open(&path).expect("valid directory");
5547 let page = reader.table.dictionaries[1].expect("string dictionary page");
5548 let mut header = [0; 12];
5549 read_at(&reader.file, page.offset, &mut header).expect("dictionary header");
5550 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
5551 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
5552 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
5553 let index_len = 12 + (count + 1) * 4 + (blocks * 2 + rank_blocks) * 8;
5554 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
5555 file.seek(SeekFrom::Start(page.offset + index_len)).expect("the first head");
5556 file.write_all(&[255]).expect("damage the order");
5557
5558 let dictionary = reader.dictionary(1).expect("read").expect("a string column has one");
5559 let error = dictionary.compare_rank(0, b"anything").expect_err("a damaged order is caught");
5560 assert!(error.message().contains("rank checksum differs"), "{error}");
5561 fs::remove_file(path).expect("remove scratch file");
5562 }
5563
5564 #[test]
5568 fn a_global_dictionary_carries_the_sorted_order_of_its_values() {
5569 let spellings = ["overlong1z", "b", "", "overlong1a", "overlong", "ab", "a", "overlong1"];
5572 let path = path("dictionary-order");
5573 let mut writer =
5574 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
5575 .expect("new file");
5576 writer
5577 .append(
5578 &Chunk::new(vec![
5579 Vector::from_values(
5580 LogicalType::Varchar,
5581 &spellings.map(|text| Value::Varchar(text.into())),
5582 )
5583 .expect("strings"),
5584 ])
5585 .expect("one column"),
5586 )
5587 .expect("stripe written");
5588 writer.finish().expect("commit");
5589
5590 let reader = Reader::open(&path).expect("valid directory");
5591 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
5592 let count = dictionary.ranks().expect("a v10 file stores one");
5593 assert_eq!(count, spellings.len(), "every distinct value has a rank");
5594 let order = (0..count)
5595 .map(|rank| dictionary.code_at_rank(rank).expect("a code"))
5596 .collect::<Vec<_>>();
5597 let mut seen = order.clone();
5598 seen.sort_unstable();
5599 assert_eq!(seen, (0..spellings.len() as u32).collect::<Vec<_>>(), "a permutation of codes");
5600
5601 let ranked = order
5602 .iter()
5603 .map(|&code| {
5604 dictionary.try_bytes_at(code as usize).expect("read").expect("a value").to_vec()
5605 })
5606 .collect::<Vec<_>>();
5607 let mut expected = spellings.map(|text| text.as_bytes().to_vec()).to_vec();
5608 expected.sort();
5609 assert_eq!(ranked, expected, "rank order is value order");
5610
5611 for (rank, value) in expected.iter().enumerate() {
5614 assert_eq!(
5615 dictionary.compare_rank(rank, value).expect("compare"),
5616 Ordering::Equal,
5617 "rank {rank} is its own value"
5618 );
5619 if rank > 0 {
5620 assert_eq!(
5621 dictionary.compare_rank(rank - 1, value).expect("compare"),
5622 Ordering::Less,
5623 "rank {rank} follows the one before it"
5624 );
5625 }
5626 }
5627 fs::remove_file(path).expect("remove scratch file");
5628 }
5629
5630 #[test]
5631 fn damaged_membership_cannot_skip_a_string_page() {
5632 let path = path("damaged-membership");
5633 let mut writer = Writer::create(
5634 &path,
5635 "items",
5636 vec![
5637 Field::required("id", LogicalType::Integer),
5638 Field::new("text", LogicalType::Varchar),
5639 ],
5640 )
5641 .expect("new file");
5642 writer.append(&sample()).expect("stripe written");
5643 writer.finish().expect("commit");
5644
5645 let reader = Reader::open(&path).expect("valid directory");
5646 let membership = reader.table.stripes[0].memberships[1].expect("string membership");
5647 let mut file = OpenOptions::new().write(true).open(&path).expect("open membership page");
5648 file.seek(SeekFrom::Start(membership.offset)).expect("membership start");
5649 file.write_all(&[255]).expect("damage membership");
5650 let error = reader.skips_codes(0, 1, &[3]).expect_err("corruption must not skip rows");
5651 assert!(error.message().contains("membership page checksum differs"), "{error}");
5652 fs::remove_file(path).expect("remove scratch file");
5653 }
5654
5655 #[test]
5656 fn membership_delta_stream_is_sorted_exact_and_bounded() {
5657 let unique = unique_codes(&[900, 4, 4, 72, 9, u32::MAX]);
5658 assert_eq!(unique, [4, 9, 72, 900, u32::MAX]);
5659 let encoded = encode_membership(&unique);
5660 assert_eq!(
5661 decode_membership(&encoded).expect("valid membership"),
5662 [4, 9, 72, 900, u32::MAX]
5663 );
5664 let merged = merged_codes(vec![vec![4, 900], vec![9, 900, u32::MAX], vec![72]]);
5667 assert_eq!(merged, [4, 9, 72, 900, u32::MAX]);
5668 assert_eq!(
5669 decode_membership(&encode_membership(&merged)).expect("valid membership"),
5670 unique
5671 );
5672 assert!(decode_membership(&[1, 0x80]).is_err(), "a truncated varint is invalid");
5673 assert!(
5674 decode_membership(&[1, 0xff, 0xff, 0xff, 0xff, 0x10]).is_err(),
5675 "a value past u32 is invalid"
5676 );
5677 }
5678
5679 #[test]
5680 fn a_global_dictionary_may_be_larger_than_one_column_page() {
5681 let dictionary = Page {
5682 offset: HEADER,
5683 length: u32::try_from(MAX_PAGE + 1).expect("the page bound fits on disk"),
5684 hash: 0,
5685 };
5686 let table = Table {
5687 name: "items".to_owned(),
5688 fields: vec![Field::new("text", LogicalType::Varchar)],
5689 stripes: Vec::new(),
5690 rows: 0,
5691 dictionaries: vec![Some(dictionary)],
5692 frequencies: vec![None],
5693 };
5694 let directory = encode_directory(&table).expect("directory");
5695 let file_size = dictionary.offset + u64::from(dictionary.length) + 1;
5696
5697 let decoded = decode_directory(&directory, file_size).expect("large lazy dictionary");
5698 assert_eq!(decoded.dictionaries[0].expect("dictionary").length, dictionary.length);
5699 }
5700
5701 #[test]
5702 fn a_column_with_one_value_everywhere_costs_almost_nothing_a_row() {
5703 let path = path("constant-codes");
5704 let mut writer =
5705 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
5706 .expect("new file");
5707 let empty = vec![Value::Varchar(String::new()); 1024];
5708 for _ in 0..4 {
5709 let column = Vector::from_values(LogicalType::Varchar, &empty).expect("strings");
5710 writer.append(&Chunk::new(vec![column]).expect("one column")).expect("a part");
5711 }
5712 writer.finish().expect("commit");
5713
5714 let reader = Reader::open(&path).expect("valid directory");
5715 let pages = reader.layout().columns.first().expect("one column").pages;
5716 assert!(pages < 256, "{pages} bytes of pages for 4,096 rows of one value");
5720 let read = reader.read(3, &[0]).expect("the last part back");
5721 assert_eq!(read.value_at(0, 0), Value::Varchar(String::new()));
5722 assert_eq!(read.value_at(1023, 0), Value::Varchar(String::new()));
5723 fs::remove_file(path).expect("remove scratch file");
5724 }
5725
5726 #[test]
5727 fn a_cascade_value_too_wide_for_its_column_is_refused_rather_than_cut() {
5728 let over = vec![i64::from(i32::MAX) + 1];
5731 let error = narrowed(&LogicalType::Integer, over).expect_err("a page that disagrees");
5732 assert!(format!("{error}").contains("not of its type"), "{error}");
5733 assert!(narrowed(&LogicalType::BigInt, vec![i64::MIN]).is_ok(), "bigint holds all of i64");
5734 assert!(narrowed(&LogicalType::Varchar, vec![0]).is_err(), "strings are not integers");
5735 }
5736
5737 #[test]
5738 fn a_code_stream_the_cascade_cannot_shrink_is_left_alone() {
5739 let mut state: u32 = 0x9e37_79b9;
5743 let spread: Vec<u32> = (0..1024)
5744 .map(|_| {
5745 state ^= state << 13;
5746 state ^= state >> 17;
5747 state ^= state << 5;
5748 state
5749 })
5750 .collect();
5751 assert_eq!(encoded_codes(&spread).expect("no failure"), None);
5752 let near: Vec<u32> = (0..1024).collect();
5753 let coded = encoded_codes(&near).expect("no failure").expect("counting up is packable");
5754 assert!(coded.len() < near.len() * 4, "{} bytes for a run of 1,024", coded.len());
5755 }
5756
5757 #[test]
5763 fn two_writes_of_the_same_rows_give_the_same_bytes() {
5764 fn written(path: &PathBuf) {
5765 let fields = (0..40)
5766 .map(|column| {
5767 let ty =
5768 if column % 4 == 0 { LogicalType::Varchar } else { LogicalType::BigInt };
5769 Field::new(format!("c{column}"), ty)
5770 })
5771 .collect::<Vec<_>>();
5772 let mut writer = Writer::create(path, "wide", fields).expect("new file");
5773 for part in 0..70_u64 {
5774 let columns = (0..40)
5775 .map(|column| {
5776 let values = (0..64_u64)
5777 .map(|row| {
5778 let seed = part.wrapping_mul(31).wrapping_add(row);
5779 if column % 4 == 0 {
5780 Value::Varchar(format!("v{}", seed % 17))
5781 } else {
5782 Value::BigInt(i64::try_from(seed % 97).expect("small"))
5783 }
5784 })
5785 .collect::<Vec<_>>();
5786 let ty = if column % 4 == 0 {
5787 LogicalType::Varchar
5788 } else {
5789 LogicalType::BigInt
5790 };
5791 Vector::from_values(ty, &values).expect("a column")
5792 })
5793 .collect::<Vec<_>>();
5794 writer.append(&Chunk::new(columns).expect("forty columns")).expect("a part");
5795 }
5796 writer.finish().expect("commit");
5797 }
5798
5799 let first = path("repeatable-one");
5800 let second = path("repeatable-two");
5801 written(&first);
5802 written(&second);
5803 let left = fs::read(&first).expect("the first file");
5804 let right = fs::read(&second).expect("the second file");
5805 assert_eq!(left.len(), right.len(), "two writes of the same rows differ in length");
5806 assert!(left == right, "two writes of the same rows differ in their bytes");
5807
5808 let reader = Reader::open(&first).expect("valid directory");
5811 assert_eq!(reader.table().rows(), 70 * 64);
5812 let read = reader.read(0, &[0, 1]).expect("the first part back");
5813 assert_eq!(read.value_at(0, 0), Value::Varchar("v0".to_owned()));
5814 assert_eq!(read.value_at(0, 1), Value::BigInt(0));
5815 fs::remove_file(first).expect("remove scratch file");
5816 fs::remove_file(second).expect("remove scratch file");
5817 }
5818}