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, Write};
34use std::mem::size_of;
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_storage::sieve::Sieve;
42use rudb_storage::{Probe, Range, Zone};
43use rudb_vector::string::StringColumn;
44use rudb_vector::validity::Validity;
45use rudb_vector::{Buffer, Chunk, Data, TextSource, Vector};
46
47const MAGIC: &[u8; 8] = b"RUDBNV10";
48const DIRECTORY: &[u8; 8] = b"RUDBDI10";
49const FORMAT: u32 = 12;
50const HEADER: u64 = 80;
51const SLOT_BYTES: usize = 28;
52const MAX_PAGE: usize = 256 * 1024 * 1024;
53const MAX_DIRECTORY: usize = 128 * 1024 * 1024;
54const FREQUENCIES: &[u8; 8] = b"RUDBFQ2\0";
55const FREQUENCY_CANDIDATES: usize = 32_768;
56const FREQUENCY_ENTRIES: usize = 512;
57const FREQUENCY_BUILD_RANK: usize = 10;
58const FREQUENCY_ORDINALS: usize = 65_536;
59const MAX_FREQUENCY_WORKERS: usize = 16;
60
61const SIEVE_BUDGET: usize = 8 * 1024;
67
68fn io(error: std::io::Error) -> Error {
69 Error::io(error.to_string())
70}
71
72fn invalid(message: &str) -> Error {
73 Error::invalid_input(format!("invalid rudb native file: {message}"))
74}
75
76fn checksum(bytes: &[u8]) -> u64 {
77 const P1: u64 = 11_400_714_785_074_694_791;
78 const P2: u64 = 14_029_467_366_897_019_727;
79 const P3: u64 = 1_609_587_929_392_839_161;
80 const P4: u64 = 9_650_029_242_287_828_579;
81 const P5: u64 = 2_870_177_450_012_600_261;
82 let round = |state: u64, word: u64| {
83 state.wrapping_add(word.wrapping_mul(P2)).rotate_left(31).wrapping_mul(P1)
84 };
85 let merge = |state: u64, lane: u64| (state ^ round(0, lane)).wrapping_mul(P1).wrapping_add(P4);
86 let word =
87 |at: usize| u64::from_le_bytes(bytes[at..at + 8].try_into().expect("eight checksum bytes"));
88
89 let mut at = 0;
90 let mut hash = if bytes.len() >= 32 {
91 let mut one = P1.wrapping_add(P2);
92 let mut two = P2;
93 let mut three = 0;
94 let mut four = 0_u64.wrapping_sub(P1);
95 while at + 32 <= bytes.len() {
96 one = round(one, word(at));
97 two = round(two, word(at + 8));
98 three = round(three, word(at + 16));
99 four = round(four, word(at + 24));
100 at += 32;
101 }
102 let combined = one
103 .rotate_left(1)
104 .wrapping_add(two.rotate_left(7))
105 .wrapping_add(three.rotate_left(12))
106 .wrapping_add(four.rotate_left(18));
107 merge(merge(merge(merge(combined, one), two), three), four)
108 } else {
109 P5
110 };
111 hash = hash.wrapping_add(bytes.len() as u64);
112 while at + 8 <= bytes.len() {
113 hash ^= round(0, word(at));
114 hash = hash.rotate_left(27).wrapping_mul(P1).wrapping_add(P4);
115 at += 8;
116 }
117 if at + 4 <= bytes.len() {
118 let tail = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four checksum bytes"));
119 hash ^= u64::from(tail).wrapping_mul(P1);
120 hash = hash.rotate_left(23).wrapping_mul(P2).wrapping_add(P3);
121 at += 4;
122 }
123 while at < bytes.len() {
124 hash ^= u64::from(bytes[at]).wrapping_mul(P5);
125 hash = hash.rotate_left(11).wrapping_mul(P1);
126 at += 1;
127 }
128 hash ^= hash >> 33;
129 hash = hash.wrapping_mul(P2);
130 hash ^= hash >> 29;
131 hash = hash.wrapping_mul(P3);
132 hash ^ (hash >> 32)
133}
134
135#[derive(Debug, Clone, Copy)]
136struct Slot {
137 offset: u64,
138 length: u32,
139 generation: u64,
140 hash: u64,
141}
142
143impl Slot {
144 fn bytes(self) -> [u8; SLOT_BYTES] {
145 let mut result = [0; SLOT_BYTES];
146 result[..8].copy_from_slice(&self.offset.to_le_bytes());
147 result[8..12].copy_from_slice(&self.length.to_le_bytes());
148 result[12..20].copy_from_slice(&self.generation.to_le_bytes());
149 result[20..28].copy_from_slice(&self.hash.to_le_bytes());
150 result
151 }
152
153 fn read(bytes: &[u8]) -> Self {
154 Self {
155 offset: u64::from_le_bytes(bytes[..8].try_into().expect("eight bytes")),
156 length: u32::from_le_bytes(bytes[8..12].try_into().expect("four bytes")),
157 generation: u64::from_le_bytes(bytes[12..20].try_into().expect("eight bytes")),
158 hash: u64::from_le_bytes(bytes[20..28].try_into().expect("eight bytes")),
159 }
160 }
161}
162
163#[derive(Debug, Clone, Copy)]
164struct Page {
165 offset: u64,
166 length: u32,
167 hash: u64,
168}
169
170#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
171enum FrequencyValue {
172 Null,
173 Integer(i128),
174 Code(u32),
175}
176
177#[derive(Debug, Clone)]
178struct FrequencyEntry {
179 value: FrequencyValue,
180 count: u64,
181}
182
183#[derive(Debug, Clone)]
188struct FrequencySummary {
189 entries: Vec<FrequencyEntry>,
190 omitted_max: u64,
191 ordinals: Vec<u64>,
192}
193
194#[derive(Debug, Clone, PartialEq, Eq)]
196pub struct FrequencyOccurrences {
197 pub omitted_max: u64,
199 pub ordinals: Vec<u64>,
201}
202
203#[derive(Debug, Clone, Copy, Default)]
210struct Span {
211 offset: u64,
212 length: u32,
213}
214
215#[derive(Debug, Clone)]
217pub struct Stripe {
218 rows: usize,
219 parts: Vec<u32>,
222 index: Span,
226 pages: Vec<Span>,
227 memberships: Vec<Option<Page>>,
228 sieves: Vec<Option<Page>>,
231 zone: Zone,
232}
233
234impl Stripe {
235 #[must_use]
237 pub fn rows(&self) -> usize {
238 self.rows
239 }
240
241 #[must_use]
243 pub fn parts(&self) -> usize {
244 self.parts.len()
245 }
246}
247
248#[derive(Debug, Clone)]
250pub struct Table {
251 name: String,
252 fields: Vec<Field>,
253 stripes: Vec<Stripe>,
254 rows: usize,
255 dictionaries: Vec<Option<Page>>,
256 frequencies: Vec<Option<FrequencySummary>>,
257}
258
259impl Table {
260 #[must_use]
262 pub fn name(&self) -> &str {
263 &self.name
264 }
265
266 #[must_use]
268 pub fn fields(&self) -> &[Field] {
269 &self.fields
270 }
271
272 #[must_use]
274 pub fn rows(&self) -> usize {
275 self.rows
276 }
277
278 #[must_use]
280 pub fn stripes(&self) -> &[Stripe] {
281 &self.stripes
282 }
283}
284
285#[derive(Debug)]
287struct GlobalDictionary {
288 primary: HashMap<u64, u32>,
289 collisions: HashMap<u64, Vec<u32>>,
290 offsets: Vec<u32>,
291 payload: Vec<u8>,
292 counts: Vec<u64>,
293 nulls: u64,
294}
295
296impl GlobalDictionary {
297 fn new() -> Self {
298 Self {
299 primary: HashMap::new(),
300 collisions: HashMap::new(),
301 offsets: vec![0],
302 payload: Vec::new(),
303 counts: Vec::new(),
304 nulls: 0,
305 }
306 }
307
308 fn bytes(&self, code: u32) -> Option<&[u8]> {
309 let start = *self.offsets.get(code as usize)? as usize;
310 let end = *self.offsets.get(code as usize + 1)? as usize;
311 self.payload.get(start..end)
312 }
313
314 fn code(&mut self, text: &str) -> Result<u32> {
315 let hash = checksum(text.as_bytes());
316 if let Some(&code) = self.primary.get(&hash) {
317 if self.bytes(code) == Some(text.as_bytes()) {
318 return Ok(code);
319 }
320 if let Some(codes) = self.collisions.get(&hash) {
321 if let Some(code) =
322 codes.iter().copied().find(|&code| self.bytes(code) == Some(text.as_bytes()))
323 {
324 return Ok(code);
325 }
326 }
327 let code = self.insert(text)?;
328 self.collisions.entry(hash).or_default().push(code);
329 return Ok(code);
330 }
331 let code = self.insert(text)?;
332 self.primary.insert(hash, code);
333 Ok(code)
334 }
335
336 fn insert(&mut self, text: &str) -> Result<u32> {
337 let code = u32::try_from(self.offsets.len() - 1)
338 .map_err(|_| invalid("global dictionary has too many values"))?;
339 self.payload.extend_from_slice(text.as_bytes());
340 self.offsets.push(
341 u32::try_from(self.payload.len())
342 .map_err(|_| invalid("global dictionary payload exceeds 4 GiB"))?,
343 );
344 self.counts.push(0);
345 Ok(code)
346 }
347
348 fn ranked(&self) -> Vec<(u64, u32)> {
368 let count = self.offsets.len() - 1;
369 let mut ranked = (0..count)
370 .map(|code| {
371 let code = code as u32;
372 (head(self.bytes(code).unwrap_or_default()), code)
373 })
374 .collect::<Vec<_>>();
375 ranked.sort_unstable_by(|left, right| {
376 left.0.cmp(&right.0).then_with(|| self.bytes(left.1).cmp(&self.bytes(right.1)))
377 });
378 ranked
379 }
380
381 fn observe(&mut self, code: u32, null: bool) -> Result<()> {
382 if null {
383 self.nulls = self.nulls.saturating_add(1);
384 return Ok(());
385 }
386 let count = self
387 .counts
388 .get_mut(code as usize)
389 .ok_or_else(|| invalid("global dictionary count code is out of range"))?;
390 *count = count.saturating_add(1);
391 Ok(())
392 }
393}
394
395#[derive(Debug)]
397pub struct Writer {
398 file: File,
399 table: Table,
400 generation: u64,
401 order: Vec<((u64, u64), (u64, u64))>,
404 next_order: u64,
405 dictionaries: Vec<Option<GlobalDictionary>>,
406 pending: Vec<PendingPart>,
407}
408
409#[derive(Debug)]
410struct PendingPart {
411 order: (u64, u64),
412 rows: usize,
413 pages: Vec<Vec<u8>>,
414 codes: Vec<Option<Vec<u32>>>,
415 zone: Zone,
416 sieves: Vec<Option<Sieve>>,
417}
418
419const STRIPE_PARTS: usize = 64;
426
427const INDEX_ENTRY: usize = size_of::<u32>() + size_of::<u64>();
429
430fn index_section(parts: usize) -> Result<usize> {
432 parts
433 .checked_mul(INDEX_ENTRY)
434 .and_then(|bytes| bytes.checked_add(size_of::<u64>()))
435 .ok_or_else(|| invalid("index page length overflow"))
436}
437
438impl Writer {
439 pub fn create(
445 path: impl AsRef<Path>,
446 name: impl Into<String>,
447 fields: Vec<Field>,
448 ) -> Result<Self> {
449 for field in &fields {
450 type_tag(&field.ty)?;
451 }
452 let mut file =
453 OpenOptions::new().write(true).read(true).create_new(true).open(path).map_err(io)?;
454 let mut header = [0; HEADER as usize];
455 header[..8].copy_from_slice(MAGIC);
456 header[8..12].copy_from_slice(&FORMAT.to_le_bytes());
457 file.write_all(&header).map_err(io)?;
458 Ok(Self {
459 file,
460 dictionaries: fields
461 .iter()
462 .map(|field| (field.ty == LogicalType::Varchar).then(GlobalDictionary::new))
463 .collect(),
464 table: Table {
465 name: name.into(),
466 dictionaries: vec![None; fields.len()],
467 fields,
468 stripes: Vec::new(),
469 rows: 0,
470 frequencies: Vec::new(),
471 },
472 generation: 1,
473 order: Vec::new(),
474 next_order: 0,
475 pending: Vec::with_capacity(STRIPE_PARTS),
476 })
477 }
478
479 pub fn append(&mut self, chunk: &Chunk) -> Result<()> {
485 let order = (self.next_order, 0);
486 self.next_order = self.next_order.saturating_add(1);
487 self.append_at(order, chunk)
488 }
489
490 pub fn append_at(&mut self, order: (u64, u64), chunk: &Chunk) -> Result<()> {
501 if chunk.is_empty() {
502 return Ok(());
503 }
504 if chunk.width() != self.table.fields.len() {
505 return Err(invalid("chunk width differs from table schema"));
506 }
507 let mut pages = Vec::with_capacity(chunk.width());
508 let mut codes = Vec::with_capacity(chunk.width());
509 for (index, field) in self.table.fields.iter().enumerate() {
510 let column = chunk.column(index)?;
511 if column.logical_type() != &field.ty {
512 return Err(invalid("chunk type differs from table schema"));
513 }
514 let (bytes, unique) = encode(column, self.dictionaries[index].as_mut())?;
515 if bytes.len() > MAX_PAGE {
516 return Err(invalid("column page exceeds the configured bound"));
517 }
518 pages.push(bytes);
519 codes.push(unique);
520 }
521 self.table.rows = self
522 .table
523 .rows
524 .checked_add(chunk.len())
525 .ok_or_else(|| invalid("row count overflow"))?;
526 if self.pending.last().is_some_and(|last| last.order > order) {
527 self.flush_pending()?;
528 }
529 let zone = Zone::of(chunk);
532 let mut sieves = Vec::with_capacity(chunk.width());
533 for (index, column) in chunk.columns().iter().enumerate() {
534 let range =
535 zone.column(index).ok_or_else(|| invalid("a zone is narrower than its chunk"))?;
536 if self.dictionaries[index].is_some() {
541 sieves.push(None);
542 continue;
543 }
544 sieves.push(Sieve::of(column, range, SIEVE_BUDGET));
545 }
546 self.pending.push(PendingPart { order, rows: chunk.len(), pages, codes, zone, sieves });
547 if self.pending.len() == STRIPE_PARTS {
548 self.flush_pending()?;
549 }
550 Ok(())
551 }
552
553 fn flush_pending(&mut self) -> Result<()> {
555 if self.pending.is_empty() {
556 return Ok(());
557 }
558 let width = self.table.fields.len();
559 let parts = self.pending.len();
560 let mut pages = Vec::with_capacity(width);
561 let mut memberships = vec![None; width];
562 let mut ranges = Vec::with_capacity(width);
563 let mut index = Vec::with_capacity(width.saturating_mul(index_section(parts)?));
564 for column in 0..width {
565 let offset = self.file.stream_position().map_err(io)?;
566 let section = index.len();
567 let mut length = 0_usize;
568 for pending in &self.pending {
569 let bytes = &pending.pages[column];
570 self.file.write_all(bytes).map_err(io)?;
571 put_u32(
572 &mut index,
573 u32::try_from(bytes.len()).map_err(|_| invalid("part length overflow"))?,
574 );
575 put_u64(&mut index, checksum(bytes));
576 length = length
577 .checked_add(bytes.len())
578 .ok_or_else(|| invalid("column page length overflow"))?;
579 }
580 let hash = checksum(&index[section..]);
581 put_u64(&mut index, hash);
582 if length > MAX_PAGE {
583 return Err(invalid("column page exceeds the configured bound"));
584 }
585 pages.push(Span {
586 offset,
587 length: u32::try_from(length).map_err(|_| invalid("page length overflow"))?,
588 });
589 ranges.push(merged_range(
590 self.pending
591 .iter()
592 .map(|pending| pending.zone.column(column).cloned().unwrap_or_default()),
593 ));
594 }
595 for (column, membership) in memberships.iter_mut().enumerate() {
596 if self.pending.iter().all(|pending| pending.codes[column].is_none()) {
597 continue;
598 }
599 let lists = self
600 .pending
601 .iter()
602 .map(|pending| pending.codes[column].clone().unwrap_or_default())
603 .collect::<Vec<_>>();
604 let bytes = encode_membership(&merged_codes(lists));
605 let offset = self.file.stream_position().map_err(io)?;
606 self.file.write_all(&bytes).map_err(io)?;
607 *membership = Some(Page {
608 offset,
609 length: u32::try_from(bytes.len())
610 .map_err(|_| invalid("membership page length overflow"))?,
611 hash: checksum(&bytes),
612 });
613 }
614 let mut sieves = vec![None; width];
615 for (column, page) in sieves.iter_mut().enumerate() {
616 if self.pending.iter().all(|pending| pending.sieves[column].is_none()) {
617 continue;
618 }
619 let bytes = encode_sieves(self.pending.iter().map(|pending| &pending.sieves[column]))?;
620 let offset = self.file.stream_position().map_err(io)?;
621 self.file.write_all(&bytes).map_err(io)?;
622 *page = Some(Page {
623 offset,
624 length: u32::try_from(bytes.len())
625 .map_err(|_| invalid("sieve page length overflow"))?,
626 hash: checksum(&bytes),
627 });
628 }
629 let offset = self.file.stream_position().map_err(io)?;
630 self.file.write_all(&index).map_err(io)?;
631 let index = Span {
632 offset,
633 length: u32::try_from(index.len())
634 .map_err(|_| invalid("index page length overflow"))?,
635 };
636 let mut rows = 0_usize;
637 let mut lengths = Vec::with_capacity(parts);
638 let mut span = None;
639 for pending in self.pending.drain(..) {
640 rows = rows.checked_add(pending.rows).ok_or_else(|| invalid("row count overflow"))?;
641 lengths
642 .push(u32::try_from(pending.rows).map_err(|_| invalid("part row count overflow"))?);
643 span = Some(
644 span.map_or((pending.order, pending.order), |(first, _)| (first, pending.order)),
645 );
646 }
647 self.order.push(span.ok_or_else(|| invalid("a stripe was flushed with no parts"))?);
648 self.table.stripes.push(Stripe {
649 rows,
650 parts: lengths,
651 index,
652 pages,
653 memberships,
654 sieves,
655 zone: Zone::from_ranges(ranges),
656 });
657 Ok(())
658 }
659
660 fn numeric_frequency(&self, column: usize) -> Result<Option<FrequencySummary>> {
664 let ty = &self.table.fields[column].ty;
665 if !matches!(
666 ty,
667 LogicalType::TinyInt
668 | LogicalType::SmallInt
669 | LogicalType::Integer
670 | LogicalType::BigInt
671 | LogicalType::UTinyInt
672 | LogicalType::USmallInt
673 | LogicalType::UInteger
674 | LogicalType::UBigInt
675 | LogicalType::Date
676 | LogicalType::Timestamp
677 ) {
678 return Ok(None);
679 }
680 let mut candidates: HashMap<FrequencyValue, u32> = HashMap::new();
681 let mut decrements = 0_u64;
682 self.visit_numeric(column, |_, value| {
683 if let Some(count) = candidates.get_mut(&value) {
684 *count = count.saturating_add(1);
685 } else if candidates.len() < FREQUENCY_CANDIDATES {
686 candidates.insert(value, 1);
687 } else {
688 candidates.retain(|_, count| {
689 *count -= 1;
690 *count != 0
691 });
692 decrements = decrements.saturating_add(1);
693 }
694 })?;
695 let (exact, ordinals) = if decrements == 0 {
696 (
697 candidates
698 .into_iter()
699 .map(|(value, count)| (value, u64::from(count)))
700 .collect::<HashMap<_, _>>(),
701 Vec::new(),
702 )
703 } else {
704 let mut lower = candidates.values().copied().collect::<Vec<_>>();
705 lower.sort_unstable_by(|left, right| right.cmp(left));
706 if lower.len() < FREQUENCY_BUILD_RANK
707 || u64::from(lower[FREQUENCY_BUILD_RANK - 1]) <= decrements
708 {
709 return Ok(None);
710 }
711 let mut exact =
712 candidates.into_keys().map(|value| (value, 0_u64)).collect::<HashMap<_, _>>();
713 let mut ordinals = Vec::new();
714 let mut exceeded = false;
715 self.visit_numeric(column, |ordinal, value| {
716 if let Some(count) = exact.get_mut(&value) {
717 *count = count.saturating_add(1);
718 if !exceeded {
719 if ordinals.len() < FREQUENCY_ORDINALS {
720 ordinals.push(ordinal);
721 } else {
722 ordinals.clear();
723 exceeded = true;
724 }
725 }
726 }
727 })?;
728 (exact, ordinals)
729 };
730 let mut entries = exact
731 .into_iter()
732 .map(|(value, count)| FrequencyEntry { value, count })
733 .collect::<Vec<_>>();
734 entries.sort_unstable_by(|left, right| {
735 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
736 });
737 let omitted_max =
738 entries.get(FREQUENCY_ENTRIES).map_or(decrements, |entry| decrements.max(entry.count));
739 entries.truncate(FREQUENCY_ENTRIES);
740 Ok(Some(FrequencySummary { entries, omitted_max, ordinals }))
741 }
742
743 fn visit_numeric(
744 &self,
745 column: usize,
746 mut visit: impl FnMut(u64, FrequencyValue),
747 ) -> Result<()> {
748 let ty = &self.table.fields[column].ty;
749 let mut start = 0_u64;
750 for stripe in &self.table.stripes {
751 let spans = read_index(&self.file, stripe, column)?;
752 let page = stripe.pages[column];
753 let mut bytes = vec![0; page.length as usize];
754 read_at(&self.file, page.offset, &mut bytes)?;
755 for (span, &rows) in spans.iter().zip(&stripe.parts) {
756 let part = part_bytes(&bytes, *span)?;
757 if checksum(part) != span.hash {
758 return Err(invalid("column page checksum differs while building frequencies"));
759 }
760 let rows = rows as usize;
761 let vector = decode(ty, rows, part, None)?;
762 for row in 0..rows {
764 let value = if vector.is_null_at(row) {
765 FrequencyValue::Null
766 } else {
767 let widened = match vector.signed_at(row) {
771 Some(value) => Some(value),
772 None => match vector.value_at(row) {
773 Value::UTinyInt(value) => Some(i128::from(value)),
774 Value::USmallInt(value) => Some(i128::from(value)),
775 Value::UInteger(value) => Some(i128::from(value)),
776 Value::UBigInt(value) => Some(i128::from(value)),
777 _ => None,
778 },
779 };
780 FrequencyValue::Integer(widened.ok_or_else(|| {
781 invalid("numeric frequency page did not contain an integer value")
782 })?)
783 };
784 visit(start.saturating_add(row as u64), value);
785 }
786 start = start.saturating_add(rows as u64);
787 }
788 }
789 Ok(())
790 }
791
792 fn numeric_frequencies(&self) -> Result<Vec<Option<FrequencySummary>>> {
794 let columns = self
795 .table
796 .fields
797 .iter()
798 .enumerate()
799 .filter_map(|(column, field)| {
800 matches!(
801 field.ty,
802 LogicalType::TinyInt
803 | LogicalType::SmallInt
804 | LogicalType::Integer
805 | LogicalType::BigInt
806 | LogicalType::UTinyInt
807 | LogicalType::USmallInt
808 | LogicalType::UInteger
809 | LogicalType::UBigInt
810 | LogicalType::Date
811 | LogicalType::Timestamp
812 )
813 .then_some(column)
814 })
815 .collect::<Vec<_>>();
816 let workers = std::thread::available_parallelism()
817 .map_or(1, usize::from)
818 .min(MAX_FREQUENCY_WORKERS)
819 .min(columns.len());
820 if workers <= 1 {
821 let mut frequencies = vec![None; self.table.fields.len()];
822 for column in columns {
823 frequencies[column] = self.numeric_frequency(column)?;
824 }
825 return Ok(frequencies);
826 }
827 let width = columns.len().div_ceil(workers);
828 let pieces = std::thread::scope(|scope| {
829 columns
830 .chunks(width)
831 .map(|columns| {
832 scope.spawn(|| {
833 columns
834 .iter()
835 .map(|&column| Ok((column, self.numeric_frequency(column)?)))
836 .collect::<Result<Vec<_>>>()
837 })
838 })
839 .collect::<Vec<_>>()
840 .into_iter()
841 .map(|handle| {
842 handle
843 .join()
844 .map_err(|_| Error::internal("a native frequency worker panicked"))?
845 })
846 .collect::<Result<Vec<_>>>()
847 })?;
848 let mut frequencies = vec![None; self.table.fields.len()];
849 for piece in pieces {
850 for (column, summary) in piece {
851 frequencies[column] = summary;
852 }
853 }
854 Ok(frequencies)
855 }
856
857 pub fn finish(mut self) -> Result<Table> {
863 self.flush_pending()?;
864 let mut stripes = std::mem::take(&mut self.order)
865 .into_iter()
866 .zip(std::mem::take(&mut self.table.stripes))
867 .collect::<Vec<_>>();
868 stripes.sort_by_key(|(order, _)| order.0);
869 let mut previous: Option<(u64, u64)> = None;
870 for ((first, last), _) in &stripes {
871 if previous.is_some_and(|previous| previous >= *first) {
872 return Err(invalid("chunks did not arrive in source order"));
873 }
874 previous = Some(*last);
875 }
876 self.table.stripes = stripes.into_iter().map(|(_, stripe)| stripe).collect();
877 self.table.frequencies = self.numeric_frequencies()?;
878 let dictionaries = std::mem::take(&mut self.dictionaries);
879 let orders = rankings(&dictionaries)?;
880 for (index, (dictionary, order)) in dictionaries.into_iter().zip(orders).enumerate() {
881 let Some(dictionary) = dictionary else { continue };
882 self.table.frequencies[index] = Some(code_frequency(&dictionary));
883 let encoded = encode_global_dictionary(dictionary, &order)?;
884 let offset = self.file.stream_position().map_err(io)?;
885 self.file.write_all(&encoded.index).map_err(io)?;
886 self.file.write_all(&encoded.ranks).map_err(io)?;
887 self.file.write_all(&encoded.payload).map_err(io)?;
888 let length = encoded
889 .index
890 .len()
891 .checked_add(encoded.ranks.len())
892 .and_then(|len| len.checked_add(encoded.payload.len()))
893 .ok_or_else(|| invalid("dictionary page length overflow"))?;
894 self.table.dictionaries[index] = Some(Page {
895 offset,
896 length: u32::try_from(length)
897 .map_err(|_| invalid("dictionary page length overflow"))?,
898 hash: checksum(&encoded.index),
899 });
900 }
901 let directory = encode_directory(&self.table)?;
902 if directory.len() > MAX_DIRECTORY {
903 return Err(invalid("directory exceeds the configured bound"));
904 }
905 let offset = self.file.stream_position().map_err(io)?;
906 self.file.write_all(&directory).map_err(io)?;
907 self.file.sync_all().map_err(io)?;
908 let slot = Slot {
909 offset,
910 length: u32::try_from(directory.len())
911 .map_err(|_| invalid("directory length overflow"))?,
912 generation: self.generation,
913 hash: checksum(&directory),
914 };
915 self.file.seek(SeekFrom::Start(16)).map_err(io)?;
916 self.file.write_all(&slot.bytes()).map_err(io)?;
917 self.file.sync_all().map_err(io)?;
918 Ok(self.table)
919 }
920}
921
922#[derive(Debug, Clone)]
924pub struct Reader {
925 file: Arc<File>,
926 table: Arc<Table>,
927 dictionaries: Arc<Vec<OnceLock<Arc<Vector>>>>,
928 sieves: Arc<Vec<Vec<SieveSlot>>>,
932 places: Arc<Vec<Place>>,
934 cache: Arc<Vec<Mutex<Cached>>>,
935 pages: Arc<AtomicUsize>,
938 indexes: Arc<AtomicUsize>,
941 kept: Arc<AtomicUsize>,
944}
945
946#[derive(Debug, Clone, Copy)]
948struct Place {
949 stripe: u32,
950 part: u32,
951 rows: u32,
952}
953
954#[derive(Debug, Clone, Copy)]
956struct PartSpan {
957 start: usize,
958 length: usize,
959 hash: u64,
960}
961
962#[derive(Debug, Clone)]
968struct CachedColumn {
969 stripe: usize,
970 index: Arc<Vec<PartSpan>>,
971 page: Option<Arc<Vec<u8>>>,
972}
973
974#[derive(Debug, Default)]
994struct Cached {
995 pages: Vec<Option<Arc<Vec<u8>>>>,
996 order: VecDeque<usize>,
997 loading: Vec<usize>,
998 index: Vec<Option<Arc<Vec<PartSpan>>>>,
999}
1000
1001const CACHED_STRIPES_PER_COLUMN: usize = 4;
1013
1014type SieveSlot = OnceLock<Arc<Vec<Option<Sieve>>>>;
1016
1017type CrossingCache = OnceLock<Box<[OnceLock<Result<Vec<u8>>>]>>;
1018
1019#[derive(Debug)]
1020struct NativeText {
1021 file: Arc<File>,
1022 offsets: Vec<u32>,
1023 ranks: usize,
1025 rank_at: u64,
1029 rank_hashes: Vec<u64>,
1030 rank_blocks: Vec<OnceLock<Result<Vec<u8>>>>,
1031 payload: u64,
1032 payload_len: usize,
1033 hashes: Vec<u64>,
1034 payload_extents: Vec<OnceLock<Result<Vec<u8>>>>,
1036 crossing: Vec<CrossingCache>,
1037}
1038
1039const TEXT_PAYLOAD_BLOCK: usize = 64 * 1024;
1040
1041const TEXT_PAYLOAD_EXTENT: usize = 8;
1062const TEXT_CROSSING_BLOCK: usize = 1024;
1063
1064const TEXT_RANK_BLOCK: usize = 512;
1074
1075const RANK_ENTRY: usize = size_of::<u64>() + size_of::<u32>();
1077
1078impl NativeText {
1079 fn payload_block(&self, block: usize) -> Result<Option<&[u8]>> {
1080 if block >= self.hashes.len() {
1081 return Ok(None);
1082 }
1083 let extent = block / TEXT_PAYLOAD_EXTENT;
1084 let Some(slot) = self.payload_extents.get(extent) else { return Ok(None) };
1085 let bytes = slot
1086 .get_or_init(|| {
1087 let start = extent
1088 .checked_mul(TEXT_PAYLOAD_EXTENT * TEXT_PAYLOAD_BLOCK)
1089 .ok_or_else(|| invalid("global dictionary block offset overflow"))?;
1090 let len = (TEXT_PAYLOAD_EXTENT * TEXT_PAYLOAD_BLOCK).min(
1091 self.payload_len
1092 .checked_sub(start)
1093 .ok_or_else(|| invalid("global dictionary block starts past payload"))?,
1094 );
1095 let mut bytes = vec![0; len];
1096 read_at(&self.file, self.payload + start as u64, &mut bytes)?;
1097 for (within, piece) in bytes.chunks(TEXT_PAYLOAD_BLOCK).enumerate() {
1100 if checksum(piece)
1101 != *self
1102 .hashes
1103 .get(extent * TEXT_PAYLOAD_EXTENT + within)
1104 .ok_or_else(|| invalid("global dictionary block has no checksum"))?
1105 {
1106 return Err(invalid("global dictionary payload checksum differs"));
1107 }
1108 }
1109 Ok(bytes)
1110 })
1111 .as_ref()
1112 .map_err(Clone::clone)?;
1113 let within = (block % TEXT_PAYLOAD_EXTENT) * TEXT_PAYLOAD_BLOCK;
1114 let end = (within + TEXT_PAYLOAD_BLOCK).min(bytes.len());
1115 Ok(bytes.get(within..end))
1116 }
1117
1118 fn rank_parts(&self, rank: usize) -> Result<(&[u8], usize)> {
1125 let slot = self
1126 .rank_blocks
1127 .get(rank / TEXT_RANK_BLOCK)
1128 .ok_or_else(|| invalid("global dictionary rank is past the order"))?;
1129 let block = slot
1130 .get_or_init(|| {
1131 let first = rank / TEXT_RANK_BLOCK * TEXT_RANK_BLOCK;
1132 let len = TEXT_RANK_BLOCK.min(self.ranks - first) * RANK_ENTRY;
1133 let mut bytes = vec![0; len];
1134 read_at(&self.file, self.rank_at + (first * RANK_ENTRY) as u64, &mut bytes)?;
1135 if checksum(&bytes)
1136 != *self
1137 .rank_hashes
1138 .get(rank / TEXT_RANK_BLOCK)
1139 .ok_or_else(|| invalid("global dictionary rank block has no checksum"))?
1140 {
1141 return Err(invalid("global dictionary rank checksum differs"));
1142 }
1143 Ok(bytes)
1144 })
1145 .as_ref()
1146 .map_err(Clone::clone)?;
1147 Ok((block.as_slice(), rank % TEXT_RANK_BLOCK))
1148 }
1149
1150 fn head_at(&self, rank: usize) -> Result<u64> {
1152 let (block, within) = self.rank_parts(rank)?;
1153 let at = within * size_of::<u64>();
1154 let bytes = block
1155 .get(at..at + size_of::<u64>())
1156 .ok_or_else(|| invalid("global dictionary rank block is short of heads"))?;
1157 Ok(u64::from_le_bytes(bytes.try_into().expect("eight bytes")))
1158 }
1159}
1160
1161impl TextSource for NativeText {
1162 fn len(&self) -> usize {
1163 self.offsets.len().saturating_sub(1)
1164 }
1165
1166 fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
1167 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1168 else {
1169 return Ok(None);
1170 };
1171 if start == end {
1172 return Ok(Some(&[]));
1173 }
1174 let first = start as usize / TEXT_PAYLOAD_BLOCK;
1175 let last = (end as usize - 1) / TEXT_PAYLOAD_BLOCK;
1176 if first == last {
1177 let Some(block) = self.payload_block(first)? else { return Ok(None) };
1178 let within = start as usize % TEXT_PAYLOAD_BLOCK;
1179 return Ok(block.get(within..within + (end - start) as usize));
1180 }
1181 let Some(crossing) = self.crossing.get(index / TEXT_CROSSING_BLOCK) else {
1182 return Ok(None);
1183 };
1184 let block = crossing.get_or_init(|| {
1185 (0..TEXT_CROSSING_BLOCK).map(|_| OnceLock::new()).collect::<Vec<_>>().into_boxed_slice()
1186 });
1187 block[index % TEXT_CROSSING_BLOCK]
1188 .get_or_init(|| {
1189 let mut bytes = Vec::with_capacity((end - start) as usize);
1190 for part in first..=last {
1191 let source = self
1192 .payload_block(part)?
1193 .ok_or_else(|| invalid("global dictionary block is missing"))?;
1194 let from = if part == first { start as usize % TEXT_PAYLOAD_BLOCK } else { 0 };
1195 let to = if part == last {
1196 (end as usize - 1) % TEXT_PAYLOAD_BLOCK + 1
1197 } else {
1198 source.len()
1199 };
1200 bytes.extend_from_slice(source.get(from..to).ok_or_else(|| {
1201 invalid("global dictionary value exceeds its payload block")
1202 })?);
1203 }
1204 Ok(bytes)
1205 })
1206 .as_ref()
1207 .map(|bytes| Some(bytes.as_slice()))
1208 .map_err(Clone::clone)
1209 }
1210
1211 fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
1212 let (Some(&start), Some(&end)) = (self.offsets.get(index), self.offsets.get(index + 1))
1213 else {
1214 return Ok(None);
1215 };
1216 Ok(Some((end - start) as usize))
1217 }
1218
1219 fn ranks(&self) -> Option<usize> {
1220 (self.ranks > 0).then_some(self.ranks)
1221 }
1222
1223 fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
1224 let settled = self.head_at(rank)?.cmp(&head(wanted));
1228 if settled != Ordering::Equal {
1229 return Ok(settled);
1230 }
1231 let code = self.code_at_rank(rank)?;
1232 let bytes = self
1233 .bytes_at(code as usize)?
1234 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
1235 Ok(bytes.cmp(wanted))
1236 }
1237
1238 fn code_at_rank(&self, rank: usize) -> Result<u32> {
1239 let (block, within) = self.rank_parts(rank)?;
1240 let heads = block.len() / RANK_ENTRY * size_of::<u64>();
1241 let at = heads + within * size_of::<u32>();
1242 let bytes = block
1243 .get(at..at + size_of::<u32>())
1244 .ok_or_else(|| invalid("global dictionary rank block is short of codes"))?;
1245 let code = u32::from_le_bytes(bytes.try_into().expect("four bytes"));
1246 if code as usize >= self.len() {
1247 return Err(invalid("global dictionary order names a code it does not have"));
1248 }
1249 Ok(code)
1250 }
1251
1252 fn footprint(&self) -> usize {
1253 self.offsets.capacity() * size_of::<u32>()
1254 + self.rank_hashes.capacity() * size_of::<u64>()
1255 + self.rank_blocks.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1256 + self
1257 .rank_blocks
1258 .iter()
1259 .filter_map(OnceLock::get)
1260 .filter_map(|result| result.as_ref().ok())
1261 .map(Vec::capacity)
1262 .sum::<usize>()
1263 + self.payload_extents.capacity() * size_of::<OnceLock<Result<Vec<u8>>>>()
1264 + self.hashes.capacity() * size_of::<u64>()
1265 + self
1266 .payload_extents
1267 .iter()
1268 .filter_map(OnceLock::get)
1269 .filter_map(|result| result.as_ref().ok())
1270 .map(Vec::capacity)
1271 .sum::<usize>()
1272 + self.crossing.capacity() * size_of::<CrossingCache>()
1273 + self
1274 .crossing
1275 .iter()
1276 .filter_map(OnceLock::get)
1277 .map(|block| {
1278 block.len() * size_of::<OnceLock<Result<Vec<u8>>>>()
1279 + block
1280 .iter()
1281 .filter_map(OnceLock::get)
1282 .filter_map(|result| result.as_ref().ok())
1283 .map(Vec::capacity)
1284 .sum::<usize>()
1285 })
1286 .sum::<usize>()
1287 }
1288}
1289
1290fn places(table: &Table) -> Result<Vec<Place>> {
1292 let mut places = Vec::with_capacity(table.stripes.len().saturating_mul(STRIPE_PARTS));
1293 for (at, stripe) in table.stripes.iter().enumerate() {
1294 let index = u32::try_from(at).map_err(|_| invalid("too many stripes"))?;
1295 for (part, &rows) in stripe.parts.iter().enumerate() {
1296 places.push(Place {
1297 stripe: index,
1298 part: u32::try_from(part).map_err(|_| invalid("too many parts in a stripe"))?,
1299 rows,
1300 });
1301 }
1302 }
1303 Ok(places)
1304}
1305
1306fn read_index(file: &File, stripe: &Stripe, column: usize) -> Result<Vec<PartSpan>> {
1311 let parts = stripe.parts.len();
1312 let section = index_section(parts)?;
1313 let at = column.checked_mul(section).ok_or_else(|| invalid("index page offset overflow"))?;
1314 let end = at.checked_add(section).ok_or_else(|| invalid("index page offset overflow"))?;
1315 if end > stripe.index.length as usize {
1316 return Err(invalid("index page is shorter than its columns"));
1317 }
1318 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1319 let mut bytes = vec![0; section];
1320 let offset = stripe
1321 .index
1322 .offset
1323 .checked_add(at as u64)
1324 .ok_or_else(|| invalid("index page offset overflow"))?;
1325 read_at(file, offset, &mut bytes)?;
1326 let entries = section - size_of::<u64>();
1327 let stored = u64::from_le_bytes(bytes[entries..].try_into().expect("eight bytes"));
1328 if checksum(&bytes[..entries]) != stored {
1329 return Err(invalid("index page section checksum differs"));
1330 }
1331 let mut spans = Vec::with_capacity(parts);
1332 let mut start = 0_usize;
1333 for part in 0..parts {
1334 let at = part * INDEX_ENTRY;
1335 let length = u32::from_le_bytes(bytes[at..at + 4].try_into().expect("four bytes")) as usize;
1336 let hash = u64::from_le_bytes(bytes[at + 4..at + 12].try_into().expect("eight bytes"));
1337 spans.push(PartSpan { start, length, hash });
1338 start = start.checked_add(length).ok_or_else(|| invalid("column page length overflow"))?;
1339 }
1340 if start != page.length as usize {
1341 return Err(invalid("column page length differs from its index"));
1342 }
1343 Ok(spans)
1344}
1345
1346fn part_bytes(page: &[u8], span: PartSpan) -> Result<&[u8]> {
1348 let end = span.start.checked_add(span.length).ok_or_else(|| invalid("part range overflow"))?;
1349 page.get(span.start..end).ok_or_else(|| invalid("part exceeds its column page"))
1350}
1351
1352fn remember(cached: &mut Cached, held: &CachedColumn, kept: usize) {
1357 if let Some(slot) = cached.index.get_mut(held.stripe) {
1358 if slot.is_none() {
1359 *slot = Some(Arc::clone(&held.index));
1360 }
1361 }
1362 let Some(page) = held.page.clone() else { return };
1363 let Some(slot) = cached.pages.get_mut(held.stripe) else { return };
1364 if slot.is_none() {
1365 cached.order.push_back(held.stripe);
1366 }
1367 *slot = Some(page);
1368 while cached.order.len() > kept.max(1) {
1369 let Some(oldest) = cached.order.pop_front() else { break };
1370 if let Some(slot) = cached.pages.get_mut(oldest) {
1371 *slot = None;
1372 }
1373 }
1374}
1375
1376impl Reader {
1377 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
1383 let mut file = File::open(path).map_err(io)?;
1384 let size = file.metadata().map_err(io)?.len();
1385 if size < HEADER {
1386 return Err(invalid("file is shorter than its header"));
1387 }
1388 let mut header = [0; HEADER as usize];
1389 file.read_exact(&mut header).map_err(io)?;
1390 let version = u32::from_le_bytes([header[8], header[9], header[10], header[11]]);
1391 if &header[..8] != MAGIC || version != FORMAT {
1392 return Err(invalid("magic or major version is unsupported"));
1393 }
1394 let mut selected = None;
1395 for start in [16, 16 + SLOT_BYTES] {
1396 let slot = Slot::read(&header[start..start + SLOT_BYTES]);
1397 if slot.generation == 0 || slot.length == 0 || slot.length as usize > MAX_DIRECTORY {
1398 continue;
1399 }
1400 let Some(end) = slot.offset.checked_add(u64::from(slot.length)) else { continue };
1401 if slot.offset < HEADER || end > size {
1402 continue;
1403 }
1404 let mut bytes = vec![0; slot.length as usize];
1405 file.seek(SeekFrom::Start(slot.offset)).map_err(io)?;
1406 file.read_exact(&mut bytes).map_err(io)?;
1407 if checksum(&bytes) == slot.hash
1408 && selected
1409 .as_ref()
1410 .is_none_or(|(old, _): &(Slot, Vec<u8>)| old.generation < slot.generation)
1411 {
1412 selected = Some((slot, bytes));
1413 }
1414 }
1415 let (_, bytes) = selected.ok_or_else(|| invalid("no committed directory slot is valid"))?;
1416 let table = decode_directory(&bytes, size)?;
1417 let places = places(&table)?;
1418 let dictionaries = (0..table.fields.len()).map(|_| OnceLock::new()).collect();
1419 let stripes = table.stripes.len();
1420 let cache = (0..table.fields.len())
1421 .map(|_| {
1422 Mutex::new(Cached {
1423 pages: (0..stripes).map(|_| None).collect(),
1424 index: (0..stripes).map(|_| None).collect(),
1425 ..Cached::default()
1426 })
1427 })
1428 .collect::<Vec<_>>();
1429 let sieves = (0..table.fields.len())
1430 .map(|_| table.stripes.iter().map(|_| OnceLock::new()).collect())
1431 .collect();
1432 Ok(Self {
1433 file: Arc::new(file),
1434 table: Arc::new(table),
1435 dictionaries: Arc::new(dictionaries),
1436 sieves: Arc::new(sieves),
1437 places: Arc::new(places),
1438 cache: Arc::new(cache),
1439 pages: Arc::new(AtomicUsize::new(0)),
1440 indexes: Arc::new(AtomicUsize::new(0)),
1441 kept: Arc::new(AtomicUsize::new(CACHED_STRIPES_PER_COLUMN)),
1442 })
1443 }
1444
1445 #[must_use]
1447 pub fn parts(&self) -> usize {
1448 self.places.len()
1449 }
1450
1451 #[must_use]
1458 pub fn stripe_parts(&self) -> Vec<std::ops::Range<usize>> {
1459 let mut runs = Vec::with_capacity(self.table.stripes.len());
1460 let mut start = 0;
1461 for stripe in &self.table.stripes {
1462 let end = start + stripe.parts.len();
1463 runs.push(start..end);
1464 start = end;
1465 }
1466 runs
1467 }
1468
1469 pub fn keep_stripes(&self, stripes: usize) {
1476 self.kept.fetch_max(stripes, Atomic::Relaxed);
1477 }
1478
1479 #[must_use]
1481 pub fn part_rows(&self, at: usize) -> usize {
1482 self.places.get(at).map_or(0, |place| place.rows as usize)
1483 }
1484
1485 #[must_use]
1487 pub fn table(&self) -> &Table {
1488 &self.table
1489 }
1490
1491 pub fn top_frequencies(&self, column: usize, top: usize) -> Result<Option<Vec<(Value, u64)>>> {
1500 let field = self
1501 .table
1502 .fields
1503 .get(column)
1504 .ok_or_else(|| invalid("frequency column index out of range"))?;
1505 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1506 return Ok(None);
1507 };
1508 if top == 0 || summary.entries.len() < top {
1509 return Ok(None);
1510 }
1511 let boundary = summary.entries[top - 1].count;
1512 if boundary <= summary.omitted_max {
1513 return Ok(None);
1514 }
1515 let dictionary =
1516 if field.ty == LogicalType::Varchar { self.dictionary(column)? } else { None };
1517 let mut out = Vec::with_capacity(summary.entries.len());
1518 for entry in &summary.entries {
1519 let value = match entry.value {
1520 FrequencyValue::Null => Value::Null,
1521 FrequencyValue::Integer(value) => match field.ty {
1522 LogicalType::TinyInt => Value::TinyInt(
1523 i8::try_from(value)
1524 .map_err(|_| invalid("frequency TINYINT is out of range"))?,
1525 ),
1526 LogicalType::UTinyInt => Value::UTinyInt(
1527 u8::try_from(value)
1528 .map_err(|_| invalid("frequency UTINYINT is out of range"))?,
1529 ),
1530 LogicalType::USmallInt => Value::USmallInt(
1531 u16::try_from(value)
1532 .map_err(|_| invalid("frequency USMALLINT is out of range"))?,
1533 ),
1534 LogicalType::UInteger => Value::UInteger(
1535 u32::try_from(value)
1536 .map_err(|_| invalid("frequency UINTEGER is out of range"))?,
1537 ),
1538 LogicalType::UBigInt => Value::UBigInt(
1539 u64::try_from(value)
1540 .map_err(|_| invalid("frequency UBIGINT is out of range"))?,
1541 ),
1542 LogicalType::SmallInt => Value::SmallInt(
1543 i16::try_from(value)
1544 .map_err(|_| invalid("frequency SMALLINT is out of range"))?,
1545 ),
1546 LogicalType::Integer => Value::Integer(
1547 i32::try_from(value)
1548 .map_err(|_| invalid("frequency INTEGER is out of range"))?,
1549 ),
1550 LogicalType::BigInt => Value::BigInt(
1551 i64::try_from(value)
1552 .map_err(|_| invalid("frequency BIGINT is out of range"))?,
1553 ),
1554 LogicalType::Date => Value::Date(
1555 i32::try_from(value)
1556 .map_err(|_| invalid("frequency DATE is out of range"))?,
1557 ),
1558 LogicalType::Timestamp => Value::Timestamp(
1559 i64::try_from(value)
1560 .map_err(|_| invalid("frequency TIMESTAMP is out of range"))?,
1561 ),
1562 _ => return Err(invalid("integer frequency belongs to another type")),
1563 },
1564 FrequencyValue::Code(code) => dictionary
1565 .as_ref()
1566 .ok_or_else(|| invalid("frequency code has no dictionary"))?
1567 .try_value_at(code as usize)?,
1568 };
1569 out.push((value, entry.count));
1570 }
1571 Ok(Some(out))
1572 }
1573
1574 pub fn frequency_occurrences(&self, column: usize) -> Result<Option<FrequencyOccurrences>> {
1584 self.table
1585 .fields
1586 .get(column)
1587 .ok_or_else(|| invalid("frequency column index out of range"))?;
1588 let Some(summary) = self.table.frequencies.get(column).and_then(Option::as_ref) else {
1589 return Ok(None);
1590 };
1591 if summary.ordinals.is_empty() {
1592 return Ok(None);
1593 }
1594 Ok(Some(FrequencyOccurrences {
1595 omitted_max: summary.omitted_max,
1596 ordinals: summary.ordinals.clone(),
1597 }))
1598 }
1599
1600 pub fn distinct_values(&self, column: usize) -> Result<Option<u64>> {
1620 if self.null_count(column)? > 0 {
1621 return Ok(None);
1622 }
1623 Ok(self.dictionary(column)?.map(|dictionary| dictionary.len() as u64))
1624 }
1625
1626 pub fn null_count(&self, column: usize) -> Result<u64> {
1637 if column >= self.table.fields.len() {
1638 return Err(invalid("null count column index out of range"));
1639 }
1640 let mut nulls = 0_u64;
1641 for stripe in &self.table.stripes {
1642 let range = stripe
1643 .zone
1644 .column(column)
1645 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
1646 nulls = nulls
1647 .checked_add(range.nulls as u64)
1648 .ok_or_else(|| invalid("null count overflow"))?;
1649 }
1650 Ok(nulls)
1651 }
1652
1653 pub fn text_extremes(&self, column: usize) -> Result<Option<(Value, Value)>> {
1668 if self.null_count(column)? > 0 {
1669 return Ok(None);
1670 }
1671 let Some(dictionary) = self.dictionary(column)? else { return Ok(None) };
1672 let Some(ranks) = dictionary.ranks() else { return Ok(None) };
1673 if ranks == 0 {
1674 return Ok(None);
1675 }
1676 let low = text_at_rank(&dictionary, 0)?;
1677 let high = text_at_rank(&dictionary, ranks - 1)?;
1678 Ok(Some((low, high)))
1679 }
1680
1681 pub fn exact_extremes(&self, column: usize) -> Result<Option<(Bound, Bound)>> {
1704 if column >= self.table.fields.len() {
1705 return Err(invalid("extremes column index out of range"));
1706 }
1707 let mut low: Option<Bound> = None;
1708 let mut high: Option<Bound> = None;
1709 for stripe in &self.table.stripes {
1710 let range = stripe
1711 .zone
1712 .column(column)
1713 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
1714 if !range.exact {
1715 return Ok(None);
1716 }
1717 let (Some(small), Some(large)) = (range.low.as_ref(), range.high.as_ref()) else {
1722 if stripe.rows > range.nulls {
1723 return Ok(None);
1724 }
1725 continue;
1726 };
1727 low = Some(low.map_or_else(|| small.clone(), |held| held.smaller(small.clone())));
1728 high = Some(high.map_or_else(|| large.clone(), |held| held.larger(large.clone())));
1729 }
1730 Ok(low.zip(high))
1731 }
1732
1733 pub fn exact_sum(&self, column: usize) -> Result<Option<(i128, u64)>> {
1746 if column >= self.table.fields.len() {
1747 return Err(invalid("sum column index out of range"));
1748 }
1749 let mut total = 0_i128;
1750 let mut rows = 0_u64;
1751 for stripe in &self.table.stripes {
1752 let range = stripe
1753 .zone
1754 .column(column)
1755 .ok_or_else(|| invalid("stripe zone is narrower than the schema"))?;
1756 let Some(part) = range.sum else { return Ok(None) };
1757 let Some(sum) = total.checked_add(part) else { return Ok(None) };
1758 total = sum;
1759 rows = rows.saturating_add(stripe.rows as u64 - range.nulls as u64);
1760 }
1761 Ok(Some((total, rows)))
1762 }
1763
1764 fn dictionary(&self, column: usize) -> Result<Option<Arc<Vector>>> {
1765 let Some(page) = self.table.dictionaries[column] else { return Ok(None) };
1766 if let Some(dictionary) = self.dictionaries[column].get() {
1767 return Ok(Some(Arc::clone(dictionary)));
1768 }
1769 let dictionary = Arc::new(open_global_dictionary(
1770 Arc::clone(&self.file),
1771 page,
1772 &self.table.fields[column].ty,
1773 )?);
1774 let _ = self.dictionaries[column].set(Arc::clone(&dictionary));
1775 Ok(Some(self.dictionaries[column].get().map_or(dictionary, Arc::clone)))
1776 }
1777
1778 pub fn read(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
1787 self.read_impl(part, columns, true)
1788 }
1789
1790 pub fn read_sparse(&self, part: usize, columns: &[usize]) -> Result<Chunk> {
1800 self.read_impl(part, columns, false)
1801 }
1802
1803 pub fn skips_codes(&self, part: usize, column: usize, candidates: &[u32]) -> Result<bool> {
1810 if candidates.is_empty() {
1811 return Ok(true);
1812 }
1813 if candidates.windows(2).any(|pair| pair[0] >= pair[1]) {
1814 return Err(Error::internal("native code candidates are not sorted and unique"));
1815 }
1816 let stripe = self.stripe_of(part)?;
1817 let Some(page) = stripe.memberships.get(column).copied().flatten() else {
1818 return Ok(false);
1819 };
1820 let mut bytes = vec![0; page.length as usize];
1821 read_at(&self.file, page.offset, &mut bytes)?;
1822 if checksum(&bytes) != page.hash {
1823 return Err(invalid("membership page checksum differs"));
1824 }
1825 let codes = decode_membership(&bytes)?;
1826 let mut left = 0;
1827 let mut right = 0;
1828 while left < codes.len() && right < candidates.len() {
1829 match codes[left].cmp(&candidates[right]) {
1830 Ordering::Less => left += 1,
1831 Ordering::Greater => right += 1,
1832 Ordering::Equal => return Ok(false),
1833 }
1834 }
1835 Ok(true)
1836 }
1837
1838 fn stripe_of(&self, part: usize) -> Result<&Stripe> {
1839 let place = self.places.get(part).ok_or_else(|| invalid("part index out of range"))?;
1840 self.table
1841 .stripes
1842 .get(place.stripe as usize)
1843 .ok_or_else(|| invalid("stripe index out of range"))
1844 }
1845
1846 fn held(&self, at: usize, stripe: &Stripe, column: usize, whole: bool) -> Result<CachedColumn> {
1863 let cache = self.cache.get(column).ok_or_else(|| invalid("column index out of range"))?;
1864 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
1865 let known = cached.index.get(at).and_then(Clone::clone);
1866 let page = cached.pages.get(at).and_then(Clone::clone);
1867 if let Some(index) = known.clone() {
1868 if !whole || page.is_some() {
1869 return Ok(CachedColumn { stripe: at, index, page });
1870 }
1871 }
1872 if cached.loading.contains(&at) {
1873 drop(cached);
1874 if let Some(index) = known {
1878 return Ok(CachedColumn { stripe: at, index, page: None });
1879 }
1880 let held = self.page_of(stripe, column, at, false, None)?;
1881 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
1882 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
1883 return Ok(held);
1884 }
1885 cached.loading.push(at);
1886 drop(cached);
1887
1888 let read = self.page_of(stripe, column, at, whole, known);
1889
1890 let mut cached = cache.lock().map_err(|_| invalid("column page cache is poisoned"))?;
1894 if let Some(position) = cached.loading.iter().position(|loading| *loading == at) {
1895 cached.loading.remove(position);
1896 }
1897 let held = read?;
1898 remember(&mut cached, &held, self.kept.load(Atomic::Relaxed));
1899 Ok(held)
1900 }
1901
1902 fn page_of(
1908 &self,
1909 stripe: &Stripe,
1910 column: usize,
1911 at: usize,
1912 whole: bool,
1913 known: Option<Arc<Vec<PartSpan>>>,
1914 ) -> Result<CachedColumn> {
1915 let index = match known {
1916 Some(index) => index,
1917 None => {
1918 self.indexes.fetch_add(1, Atomic::Relaxed);
1919 Arc::new(read_index(&self.file, stripe, column)?)
1920 }
1921 };
1922 let page = if whole {
1923 self.pages.fetch_add(1, Atomic::Relaxed);
1924 let span = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1925 let mut bytes = vec![0; span.length as usize];
1926 read_at(&self.file, span.offset, &mut bytes)?;
1927 Some(Arc::new(bytes))
1928 } else {
1929 None
1930 };
1931 Ok(CachedColumn { stripe: at, index, page })
1932 }
1933
1934 fn read_impl(&self, at: usize, columns: &[usize], whole: bool) -> Result<Chunk> {
1935 let place = *self.places.get(at).ok_or_else(|| invalid("part index out of range"))?;
1936 let index = place.stripe as usize;
1937 let stripe =
1938 self.table.stripes.get(index).ok_or_else(|| invalid("stripe index out of range"))?;
1939 let rows = place.rows as usize;
1940 let mut picked = Vec::with_capacity(columns.len());
1941 for &column in columns {
1942 let field = self
1943 .table
1944 .fields
1945 .get(column)
1946 .ok_or_else(|| invalid("column index out of range"))?;
1947 let page = stripe.pages.get(column).ok_or_else(|| invalid("stripe page is missing"))?;
1948 let held = self.held(index, stripe, column, whole)?;
1949 let span = *held
1950 .index
1951 .get(place.part as usize)
1952 .ok_or_else(|| invalid("part index out of range"))?;
1953 let owned;
1954 let bytes = match &held.page {
1955 Some(held) => part_bytes(held, span)?,
1956 None => {
1957 let offset = page
1958 .offset
1959 .checked_add(span.start as u64)
1960 .ok_or_else(|| invalid("part range overflow"))?;
1961 let mut bytes = vec![0; span.length];
1962 read_at(&self.file, offset, &mut bytes)?;
1963 owned = bytes;
1964 &owned
1965 }
1966 };
1967 if checksum(bytes) != span.hash {
1968 return Err(invalid("column page checksum differs"));
1969 }
1970 let dictionary = self.dictionary(column)?;
1971 picked.push(decode(&field.ty, rows, bytes, dictionary)?);
1972 }
1973 Chunk::with_rows(picked, rows)
1974 }
1975
1976 #[must_use]
1986 pub fn skips(&self, part: usize, probes: &[Probe]) -> bool {
1987 let Some(place) = self.places.get(part).copied() else { return false };
1988 let Some(stripe) = self.table.stripes.get(place.stripe as usize) else { return false };
1989 if stripe.zone.skips(probes) {
1990 return true;
1991 }
1992 probes.iter().any(|probe| self.sifted(place, probe))
1993 }
1994
1995 fn sifted(&self, place: Place, probe: &Probe) -> bool {
2001 if probe.op != Op::Equal {
2002 return false;
2003 }
2004 match self.stripe_sieves(place.stripe as usize, probe.column) {
2005 Some(sieves) => sieves
2006 .get(place.part as usize)
2007 .and_then(Option::as_ref)
2008 .is_some_and(|sieve| sieve.excludes(&probe.value)),
2009 None => false,
2010 }
2011 }
2012
2013 fn stripe_sieves(&self, stripe: usize, column: usize) -> Option<&[Option<Sieve>]> {
2020 let slot = self.sieves.get(column)?.get(stripe)?;
2021 if let Some(held) = slot.get() {
2022 return Some(held);
2023 }
2024 let page = self.table.stripes.get(stripe)?.sieves.get(column).copied().flatten()?;
2025 let mut bytes = vec![0; page.length as usize];
2026 read_at(&self.file, page.offset, &mut bytes).ok()?;
2027 if checksum(&bytes) != page.hash {
2028 return None;
2029 }
2030 let sieves = Arc::new(decode_sieves(&bytes).ok()?);
2031 let _ = slot.set(sieves);
2032 slot.get().map(|held| held.as_slice())
2033 }
2034}
2035
2036fn text_at_rank(dictionary: &Vector, rank: usize) -> Result<Value> {
2038 let code = dictionary.code_at_rank(rank)? as usize;
2039 let text = dictionary
2040 .try_text_at(code)?
2041 .ok_or_else(|| invalid("global dictionary order names a code it does not have"))?;
2042 Ok(Value::Varchar(text.into()))
2043}
2044
2045#[cfg(unix)]
2046fn read_at(file: &File, mut offset: u64, mut bytes: &mut [u8]) -> Result<()> {
2047 use std::os::unix::fs::FileExt;
2048 while !bytes.is_empty() {
2049 let read = file.read_at(bytes, offset).map_err(io)?;
2050 if read == 0 {
2051 return Err(invalid("column page ends before its declared length"));
2052 }
2053 offset += read as u64;
2054 bytes = &mut bytes[read..];
2055 }
2056 Ok(())
2057}
2058
2059#[cfg(not(unix))]
2060fn read_at(file: &File, offset: u64, bytes: &mut [u8]) -> Result<()> {
2061 let mut file = file.try_clone().map_err(io)?;
2062 file.seek(SeekFrom::Start(offset)).map_err(io)?;
2063 file.read_exact(bytes).map_err(io)
2064}
2065
2066fn type_tag(ty: &LogicalType) -> Result<u8> {
2067 match ty {
2068 LogicalType::SmallInt => Ok(1),
2069 LogicalType::Integer => Ok(2),
2070 LogicalType::BigInt => Ok(3),
2071 LogicalType::Varchar => Ok(4),
2072 LogicalType::Date => Ok(5),
2073 LogicalType::Timestamp => Ok(6),
2074 LogicalType::Boolean => Ok(7),
2075 LogicalType::TinyInt => Ok(8),
2076 LogicalType::UTinyInt => Ok(9),
2077 LogicalType::USmallInt => Ok(10),
2078 LogicalType::UInteger => Ok(11),
2079 LogicalType::UBigInt => Ok(12),
2080 _ => Err(Error::not_implemented(format!("native storage for {ty}"))),
2081 }
2082}
2083
2084fn tag_type(tag: u8) -> Result<LogicalType> {
2085 match tag {
2086 1 => Ok(LogicalType::SmallInt),
2087 2 => Ok(LogicalType::Integer),
2088 3 => Ok(LogicalType::BigInt),
2089 4 => Ok(LogicalType::Varchar),
2090 5 => Ok(LogicalType::Date),
2091 6 => Ok(LogicalType::Timestamp),
2092 7 => Ok(LogicalType::Boolean),
2093 8 => Ok(LogicalType::TinyInt),
2094 9 => Ok(LogicalType::UTinyInt),
2095 10 => Ok(LogicalType::USmallInt),
2096 11 => Ok(LogicalType::UInteger),
2097 12 => Ok(LogicalType::UBigInt),
2098 _ => Err(invalid("column type tag is unknown")),
2099 }
2100}
2101
2102fn put_u16(out: &mut Vec<u8>, value: u16) {
2103 out.extend_from_slice(&value.to_le_bytes());
2104}
2105fn put_u32(out: &mut Vec<u8>, value: u32) {
2106 out.extend_from_slice(&value.to_le_bytes());
2107}
2108fn put_u64(out: &mut Vec<u8>, value: u64) {
2109 out.extend_from_slice(&value.to_le_bytes());
2110}
2111fn put_var_u64(out: &mut Vec<u8>, mut value: u64) {
2112 while value >= 0x80 {
2113 out.push((value as u8 & 0x7f) | 0x80);
2114 value >>= 7;
2115 }
2116 out.push(value as u8);
2117}
2118
2119fn frequency_order(left: FrequencyValue, right: FrequencyValue) -> Ordering {
2120 match (left, right) {
2121 (FrequencyValue::Null, FrequencyValue::Null) => Ordering::Equal,
2122 (FrequencyValue::Null, _) => Ordering::Less,
2123 (_, FrequencyValue::Null) => Ordering::Greater,
2124 (FrequencyValue::Integer(left), FrequencyValue::Integer(right)) => left.cmp(&right),
2125 (FrequencyValue::Code(left), FrequencyValue::Code(right)) => left.cmp(&right),
2126 (FrequencyValue::Integer(_), FrequencyValue::Code(_)) => Ordering::Less,
2127 (FrequencyValue::Code(_), FrequencyValue::Integer(_)) => Ordering::Greater,
2128 }
2129}
2130
2131fn code_frequency(dictionary: &GlobalDictionary) -> FrequencySummary {
2132 let mut entries = dictionary
2133 .counts
2134 .iter()
2135 .enumerate()
2136 .filter(|(_, count)| **count != 0)
2137 .map(|(code, &count)| FrequencyEntry { value: FrequencyValue::Code(code as u32), count })
2138 .collect::<Vec<_>>();
2139 if dictionary.nulls != 0 {
2140 entries.push(FrequencyEntry { value: FrequencyValue::Null, count: dictionary.nulls });
2141 }
2142 entries.sort_unstable_by(|left, right| {
2143 right.count.cmp(&left.count).then_with(|| frequency_order(left.value, right.value))
2144 });
2145 let omitted_max = entries.get(FREQUENCY_ENTRIES).map_or(0, |entry| entry.count);
2146 entries.truncate(FREQUENCY_ENTRIES);
2147 FrequencySummary { entries, omitted_max, ordinals: Vec::new() }
2148}
2149
2150fn encode_directory(table: &Table) -> Result<Vec<u8>> {
2151 let mut out = DIRECTORY.to_vec();
2152 let name = table.name.as_bytes();
2153 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("table name too long"))?);
2154 out.extend_from_slice(name);
2155 put_u16(&mut out, u16::try_from(table.fields.len()).map_err(|_| invalid("too many columns"))?);
2156 for field in &table.fields {
2157 let name = field.name.as_bytes();
2158 put_u16(&mut out, u16::try_from(name.len()).map_err(|_| invalid("column name too long"))?);
2159 out.extend_from_slice(name);
2160 out.push(type_tag(&field.ty)?);
2161 out.push(u8::from(field.not_null));
2162 }
2163 for dictionary in &table.dictionaries {
2164 match dictionary {
2165 None => out.push(0),
2166 Some(page) => {
2167 out.push(1);
2168 put_u64(&mut out, page.offset);
2169 put_u32(&mut out, page.length);
2170 put_u64(&mut out, page.hash);
2171 }
2172 }
2173 }
2174 put_u64(&mut out, u64::try_from(table.rows).map_err(|_| invalid("row count overflow"))?);
2175 put_u32(&mut out, u32::try_from(table.stripes.len()).map_err(|_| invalid("too many stripes"))?);
2176 for stripe in &table.stripes {
2177 put_u32(
2178 &mut out,
2179 u32::try_from(stripe.parts.len()).map_err(|_| invalid("too many parts in a stripe"))?,
2180 );
2181 for &rows in &stripe.parts {
2182 put_u32(&mut out, rows);
2183 }
2184 put_u64(&mut out, stripe.index.offset);
2185 put_u32(&mut out, stripe.index.length);
2186 for page in &stripe.pages {
2187 put_u64(&mut out, page.offset);
2188 put_u32(&mut out, page.length);
2189 }
2190 for (field, membership) in table.fields.iter().zip(&stripe.memberships) {
2191 if field.ty != LogicalType::Varchar {
2192 continue;
2193 }
2194 let page =
2195 membership.ok_or_else(|| invalid("string page has no code membership index"))?;
2196 put_u64(&mut out, page.offset);
2197 put_u32(&mut out, page.length);
2198 put_u64(&mut out, page.hash);
2199 }
2200 for sieve in &stripe.sieves {
2201 match sieve {
2202 None => out.push(0),
2203 Some(page) => {
2204 out.push(1);
2205 put_u64(&mut out, page.offset);
2206 put_u32(&mut out, page.length);
2207 put_u64(&mut out, page.hash);
2208 }
2209 }
2210 }
2211 for range in stripe.zone.columns() {
2212 put_bound(&mut out, range.low.as_ref())?;
2213 put_bound(&mut out, range.high.as_ref())?;
2214 put_u32(
2215 &mut out,
2216 u32::try_from(range.nulls).map_err(|_| invalid("null count overflow"))?,
2217 );
2218 out.push(u8::from(range.exact));
2219 match range.sum {
2220 None => out.push(0),
2221 Some(total) => {
2222 out.push(1);
2223 out.extend_from_slice(&total.to_le_bytes());
2224 }
2225 }
2226 }
2227 }
2228 out.extend_from_slice(FREQUENCIES);
2229 put_u16(
2230 &mut out,
2231 u16::try_from(table.frequencies.len())
2232 .map_err(|_| invalid("too many frequency columns"))?,
2233 );
2234 for summary in &table.frequencies {
2235 let Some(summary) = summary else {
2236 out.push(0);
2237 continue;
2238 };
2239 out.push(1);
2240 put_u64(&mut out, summary.omitted_max);
2241 put_u32(
2242 &mut out,
2243 u32::try_from(summary.entries.len())
2244 .map_err(|_| invalid("too many frequency entries"))?,
2245 );
2246 for entry in &summary.entries {
2247 match entry.value {
2248 FrequencyValue::Null => out.push(0),
2249 FrequencyValue::Integer(value) => {
2250 out.push(1);
2251 out.extend_from_slice(&value.to_le_bytes());
2252 }
2253 FrequencyValue::Code(value) => {
2254 out.push(2);
2255 put_u32(&mut out, value);
2256 }
2257 }
2258 put_u64(&mut out, entry.count);
2259 }
2260 put_u32(
2261 &mut out,
2262 u32::try_from(summary.ordinals.len())
2263 .map_err(|_| invalid("too many frequency ordinals"))?,
2264 );
2265 let mut previous = 0_u64;
2266 for (at, &ordinal) in summary.ordinals.iter().enumerate() {
2267 let delta = if at == 0 {
2268 ordinal
2269 } else {
2270 ordinal
2271 .checked_sub(previous)
2272 .ok_or_else(|| invalid("frequency ordinals are not ordered"))?
2273 };
2274 if at != 0 && delta == 0 {
2275 return Err(invalid("frequency ordinals are not unique"));
2276 }
2277 put_var_u64(&mut out, delta);
2278 previous = ordinal;
2279 }
2280 }
2281 Ok(out)
2282}
2283
2284struct Cursor<'a> {
2285 bytes: &'a [u8],
2286 at: usize,
2287}
2288impl<'a> Cursor<'a> {
2289 fn take(&mut self, len: usize) -> Result<&'a [u8]> {
2290 let end = self.at.checked_add(len).ok_or_else(|| invalid("directory offset overflow"))?;
2291 let bytes =
2292 self.bytes.get(self.at..end).ok_or_else(|| invalid("directory is truncated"))?;
2293 self.at = end;
2294 Ok(bytes)
2295 }
2296 fn u8(&mut self) -> Result<u8> {
2297 Ok(self.take(1)?[0])
2298 }
2299 fn u16(&mut self) -> Result<u16> {
2300 Ok(u16::from_le_bytes(self.take(2)?.try_into().expect("two bytes")))
2301 }
2302 fn u32(&mut self) -> Result<u32> {
2303 Ok(u32::from_le_bytes(self.take(4)?.try_into().expect("four bytes")))
2304 }
2305 fn u64(&mut self) -> Result<u64> {
2306 Ok(u64::from_le_bytes(self.take(8)?.try_into().expect("eight bytes")))
2307 }
2308 fn var_u64(&mut self) -> Result<u64> {
2309 let mut value = 0_u64;
2310 for shift in (0..=63).step_by(7) {
2311 let byte = self.u8()?;
2312 let part = u64::from(byte & 0x7f);
2313 if shift == 63 && part > 1 {
2314 return Err(invalid("frequency ordinal varint overflows"));
2315 }
2316 value |= part << shift;
2317 if byte & 0x80 == 0 {
2318 return Ok(value);
2319 }
2320 }
2321 Err(invalid("frequency ordinal varint is too long"))
2322 }
2323 fn bound(&mut self) -> Result<Option<Bound>> {
2324 Ok(match self.u8()? {
2325 0 => None,
2326 1 => Some(Bound::Int(i128::from_le_bytes(
2327 self.take(16)?.try_into().expect("sixteen bytes"),
2328 ))),
2329 2 => Some(Bound::Real(f64::from_le_bytes(
2330 self.take(8)?.try_into().expect("eight bytes"),
2331 ))),
2332 3 => {
2333 let length = self.u32()? as usize;
2334 Some(Bound::Bytes(self.take(length)?.to_vec()))
2335 }
2336 _ => return Err(invalid("bound tag differs")),
2337 })
2338 }
2339 fn text(&mut self) -> Result<String> {
2340 let len = self.u16()? as usize;
2341 String::from_utf8(self.take(len)?.to_vec()).map_err(|_| invalid("name is not UTF-8"))
2342 }
2343}
2344
2345fn decode_directory(bytes: &[u8], size: u64) -> Result<Table> {
2346 let mut cur = Cursor { bytes, at: 0 };
2347 if cur.take(8)? != DIRECTORY {
2348 return Err(invalid("directory magic differs"));
2349 }
2350 let name = cur.text()?;
2351 let width = cur.u16()? as usize;
2352 let mut fields = Vec::with_capacity(width);
2353 for _ in 0..width {
2354 let name = cur.text()?;
2355 let ty = tag_type(cur.u8()?)?;
2356 let not_null = match cur.u8()? {
2357 0 => false,
2358 1 => true,
2359 _ => return Err(invalid("nullability flag differs")),
2360 };
2361 fields.push(Field { name, ty, not_null });
2362 }
2363 let mut dictionaries = Vec::with_capacity(width);
2364 for _ in 0..width {
2365 dictionaries.push(match cur.u8()? {
2366 0 => None,
2367 1 => {
2368 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2369 let end = page
2370 .offset
2371 .checked_add(u64::from(page.length))
2372 .ok_or_else(|| invalid("dictionary page offset overflow"))?;
2373 if page.offset < HEADER || end > size {
2378 return Err(invalid("dictionary page range is outside the file"));
2379 }
2380 Some(page)
2381 }
2382 _ => return Err(invalid("dictionary page tag differs")),
2383 });
2384 }
2385 let rows = usize::try_from(cur.u64()?).map_err(|_| invalid("row count does not fit"))?;
2386 let count = cur.u32()? as usize;
2387 let mut stripes = Vec::with_capacity(count);
2388 let mut total = 0_usize;
2389 for _ in 0..count {
2390 let count = cur.u32()? as usize;
2391 if count == 0 || count > STRIPE_PARTS {
2392 return Err(invalid("stripe part count is outside its bound"));
2393 }
2394 let mut parts = Vec::with_capacity(count);
2395 let mut stripe_rows = 0_usize;
2396 for _ in 0..count {
2397 let rows = cur.u32()?;
2398 if rows == 0 {
2399 return Err(invalid("empty part"));
2400 }
2401 parts.push(rows);
2402 stripe_rows = stripe_rows
2403 .checked_add(rows as usize)
2404 .ok_or_else(|| invalid("stripe row count overflow"))?;
2405 }
2406 total =
2407 total.checked_add(stripe_rows).ok_or_else(|| invalid("stripe row count overflow"))?;
2408 let index = Span { offset: cur.u64()?, length: cur.u32()? };
2409 let section = index_section(count)?;
2410 let wanted = section
2411 .checked_mul(width)
2412 .and_then(|bytes| u32::try_from(bytes).ok())
2413 .ok_or_else(|| invalid("index page length overflow"))?;
2414 let end = index
2415 .offset
2416 .checked_add(u64::from(index.length))
2417 .ok_or_else(|| invalid("index page offset overflow"))?;
2418 if index.offset < HEADER || end > size || index.length != wanted {
2419 return Err(invalid("index page range is outside the file"));
2420 }
2421 let mut pages = Vec::with_capacity(width);
2422 for _ in 0..width {
2423 let offset = cur.u64()?;
2424 let length = cur.u32()?;
2425 let end = offset
2426 .checked_add(u64::from(length))
2427 .ok_or_else(|| invalid("page offset overflow"))?;
2428 if offset < HEADER || end > size || length as usize > MAX_PAGE {
2429 return Err(invalid("page range is outside the file"));
2430 }
2431 pages.push(Span { offset, length });
2432 }
2433 let mut memberships = vec![None; width];
2434 for (column, field) in fields.iter().enumerate() {
2435 if field.ty != LogicalType::Varchar {
2436 continue;
2437 }
2438 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2439 let end = page
2440 .offset
2441 .checked_add(u64::from(page.length))
2442 .ok_or_else(|| invalid("membership page offset overflow"))?;
2443 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
2444 return Err(invalid("membership page range is outside the file"));
2445 }
2446 memberships[column] = Some(page);
2447 }
2448 let mut sieves = vec![None; width];
2449 for sieve in sieves.iter_mut().take(width) {
2450 match cur.u8()? {
2451 0 => continue,
2452 1 => {}
2453 _ => return Err(invalid("a sieve page has an unknown tag")),
2454 }
2455 let page = Page { offset: cur.u64()?, length: cur.u32()?, hash: cur.u64()? };
2456 let end = page
2457 .offset
2458 .checked_add(u64::from(page.length))
2459 .ok_or_else(|| invalid("sieve page offset overflow"))?;
2460 if page.offset < HEADER || end > size || page.length as usize > MAX_PAGE {
2461 return Err(invalid("sieve page range is outside the file"));
2462 }
2463 *sieve = Some(page);
2464 }
2465 let mut ranges = Vec::with_capacity(width);
2466 for _ in 0..width {
2467 let low = cur.bound()?;
2468 let high = cur.bound()?;
2469 let nulls = cur.u32()? as usize;
2470 if nulls > stripe_rows {
2471 return Err(invalid("null count exceeds stripe rows"));
2472 }
2473 let exact = cur.u8()? != 0;
2474 let sum = match cur.u8()? {
2475 0 => None,
2476 1 => Some(i128::from_le_bytes(
2477 cur.take(16)?.try_into().map_err(|_| invalid("a stripe sum is truncated"))?,
2478 )),
2479 _ => return Err(invalid("a stripe sum has an unknown tag")),
2480 };
2481 ranges.push(Range { low, high, nulls, exact, sum });
2482 }
2483 stripes.push(Stripe {
2484 rows: stripe_rows,
2485 parts,
2486 index,
2487 pages,
2488 memberships,
2489 sieves,
2490 zone: Zone::from_ranges(ranges),
2491 });
2492 }
2493 if total != rows {
2494 return Err(invalid("table row count differs from stripes"));
2495 }
2496 let frequencies = if cur.at == bytes.len() {
2497 vec![None; width]
2498 } else {
2499 if cur.take(8)? != FREQUENCIES {
2500 return Err(invalid("directory extension magic differs"));
2501 }
2502 if cur.u16()? as usize != width {
2503 return Err(invalid("frequency column count differs"));
2504 }
2505 let mut frequencies = Vec::with_capacity(width);
2506 for field in &fields {
2507 let summary = match cur.u8()? {
2508 0 => None,
2509 1 => {
2510 let omitted_max = cur.u64()?;
2511 let count = cur.u32()? as usize;
2512 if count > FREQUENCY_ENTRIES {
2513 return Err(invalid("frequency entry count exceeds its bound"));
2514 }
2515 let mut entries = Vec::with_capacity(count);
2516 for _ in 0..count {
2518 let value = match cur.u8()? {
2519 0 => FrequencyValue::Null,
2520 1 => FrequencyValue::Integer(i128::from_le_bytes(
2521 cur.take(16)?.try_into().expect("sixteen bytes"),
2522 )),
2523 2 => FrequencyValue::Code(cur.u32()?),
2524 _ => return Err(invalid("frequency value tag differs")),
2525 };
2526 let valid = matches!(
2527 (&field.ty, value),
2528 (_, FrequencyValue::Null)
2529 | (LogicalType::Varchar, FrequencyValue::Code(_))
2530 | (
2531 LogicalType::TinyInt
2532 | LogicalType::SmallInt
2533 | LogicalType::Integer
2534 | LogicalType::BigInt
2535 | LogicalType::UTinyInt
2536 | LogicalType::USmallInt
2537 | LogicalType::UInteger
2538 | LogicalType::UBigInt
2539 | LogicalType::Date
2540 | LogicalType::Timestamp,
2541 FrequencyValue::Integer(_),
2542 )
2543 );
2544 if !valid {
2545 return Err(invalid("frequency value does not match its column"));
2546 }
2547 let count = cur.u64()?;
2548 if count == 0 || count > rows as u64 {
2549 return Err(invalid("frequency count is outside the table"));
2550 }
2551 entries.push(FrequencyEntry { value, count });
2552 }
2553 if entries.windows(2).any(|pair| pair[0].count < pair[1].count) {
2554 return Err(invalid("frequency entries are not descending"));
2555 }
2556 let ordinals = {
2557 let ordinal_count = cur.u32()? as usize;
2558 if ordinal_count > FREQUENCY_ORDINALS || ordinal_count > rows {
2559 return Err(invalid("frequency ordinal count exceeds its bound"));
2560 }
2561 let mut ordinals = Vec::with_capacity(ordinal_count);
2562 let mut previous = 0_u64;
2563 for at in 0..ordinal_count {
2564 let delta = cur.var_u64()?;
2565 if at != 0 && delta == 0 {
2566 return Err(invalid("frequency ordinals are not increasing"));
2567 }
2568 let ordinal = if at == 0 {
2569 delta
2570 } else {
2571 previous
2572 .checked_add(delta)
2573 .ok_or_else(|| invalid("frequency ordinal overflows"))?
2574 };
2575 if ordinal >= rows as u64 {
2576 return Err(invalid("frequency ordinal is outside the table"));
2577 }
2578 ordinals.push(ordinal);
2579 previous = ordinal;
2580 }
2581 ordinals
2582 };
2583 Some(FrequencySummary { entries, omitted_max, ordinals })
2584 }
2585 _ => return Err(invalid("frequency summary tag differs")),
2586 };
2587 frequencies.push(summary);
2588 }
2589 frequencies
2590 };
2591 if cur.at != bytes.len() {
2592 return Err(invalid("directory has trailing bytes"));
2593 }
2594 Ok(Table { name, fields, stripes, rows, dictionaries, frequencies })
2595}
2596
2597fn put_bound(out: &mut Vec<u8>, bound: Option<&Bound>) -> Result<()> {
2598 match bound {
2599 None => out.push(0),
2600 Some(Bound::Int(value)) => {
2601 out.push(1);
2602 out.extend_from_slice(&value.to_le_bytes());
2603 }
2604 Some(Bound::Real(value)) => {
2605 out.push(2);
2606 out.extend_from_slice(&value.to_le_bytes());
2607 }
2608 Some(Bound::Bytes(value)) => {
2609 out.push(3);
2610 put_u32(out, u32::try_from(value.len()).map_err(|_| invalid("bound length overflow"))?);
2611 out.extend_from_slice(value);
2612 }
2613 }
2614 Ok(())
2615}
2616
2617fn encode(
2618 vector: &Vector,
2619 global: Option<&mut GlobalDictionary>,
2620) -> Result<(Vec<u8>, Option<Vec<u32>>)> {
2621 let ty = vector.logical_type();
2622 let flat = vector.flatten()?;
2624 let mut out = Vec::new();
2625 let mut global_codes = None;
2626 if let Some(global) = global {
2627 let mut codes = Vec::with_capacity(flat.len());
2628 for row in 0..flat.len() {
2629 let text = flat.text_at(row).unwrap_or("");
2630 let code = global.code(text)?;
2631 global.observe(code, flat.is_null_at(row))?;
2632 codes.push(code);
2633 }
2634 global_codes = Some(codes);
2635 }
2636 let membership = global_codes.as_deref().map(unique_codes);
2637 let dictionary = if global_codes.is_none() && ty == &LogicalType::Varchar {
2638 string_dictionary(&flat)?
2639 } else {
2640 None
2641 };
2642 let packed_vector = if dictionary.is_none() && global_codes.is_none() {
2643 Some(flat.bit_packed()?)
2644 } else {
2645 None
2646 };
2647 let packed = packed_vector.as_ref().and_then(Vector::packed_parts);
2648 out.push(if global_codes.is_some() {
2649 3
2650 } else if dictionary.is_some() {
2651 1
2652 } else if packed.is_some() {
2653 2
2654 } else {
2655 0
2656 });
2657 let nulls = flat.validity();
2658 let flag = match nulls {
2659 Validity::AllValid => 0,
2660 Validity::AllInvalid => 1,
2661 Validity::Mask(_) => 2,
2662 };
2663 out.push(flag);
2664 if flag == 2 {
2665 for group in (0..vector.len()).step_by(8) {
2666 let mut bits = 0_u8;
2667 for bit in 0..8 {
2668 if group + bit < vector.len() && !flat.is_null_at(group + bit) {
2669 bits |= 1 << bit;
2670 }
2671 }
2672 out.push(bits);
2673 }
2674 }
2675 if let Some(codes) = global_codes {
2676 for code in codes {
2677 put_u32(&mut out, code);
2678 }
2679 return Ok((out, membership));
2680 }
2681 if let Some(dictionary) = dictionary {
2682 out.extend_from_slice(&dictionary);
2683 return Ok((out, membership));
2684 }
2685 if let Some(packed) = packed {
2686 if packed.offset() != 0 {
2687 return Err(invalid("writer received a sliced packed vector"));
2688 }
2689 out.push(u8::try_from(packed.width()).map_err(|_| invalid("packed width overflow"))?);
2690 out.extend_from_slice(&packed.base().to_le_bytes());
2691 put_u32(
2692 &mut out,
2693 u32::try_from(packed.words().len()).map_err(|_| invalid("too many packed words"))?,
2694 );
2695 for word in packed.words() {
2696 put_u64(&mut out, *word);
2697 }
2698 return Ok((out, membership));
2699 }
2700 let data = flat.data().ok_or_else(|| invalid("scalar column did not flatten"))?;
2701 match (ty, data) {
2702 (LogicalType::TinyInt, Data::Int8(values)) => {
2703 for value in &**values {
2704 out.extend_from_slice(&value.to_le_bytes());
2705 }
2706 }
2707 (LogicalType::UTinyInt, Data::UInt8(values)) => {
2708 for value in &**values {
2709 out.extend_from_slice(&value.to_le_bytes());
2710 }
2711 }
2712 (LogicalType::SmallInt, Data::Int16(values)) => {
2713 for value in &**values {
2714 out.extend_from_slice(&value.to_le_bytes());
2715 }
2716 }
2717 (LogicalType::USmallInt, Data::UInt16(values)) => {
2718 for value in &**values {
2719 out.extend_from_slice(&value.to_le_bytes());
2720 }
2721 }
2722 (LogicalType::UInteger, Data::UInt32(values)) => {
2723 for value in &**values {
2724 out.extend_from_slice(&value.to_le_bytes());
2725 }
2726 }
2727 (LogicalType::UBigInt, Data::UInt64(values)) => {
2728 for value in &**values {
2729 out.extend_from_slice(&value.to_le_bytes());
2730 }
2731 }
2732 (LogicalType::Integer | LogicalType::Date, Data::Int32(values)) => {
2733 for value in &**values {
2734 out.extend_from_slice(&value.to_le_bytes());
2735 }
2736 }
2737 (LogicalType::BigInt | LogicalType::Timestamp, Data::Int64(values)) => {
2738 for value in &**values {
2739 out.extend_from_slice(&value.to_le_bytes());
2740 }
2741 }
2742 (LogicalType::Boolean, Data::Bool(values)) => {
2743 for value in &**values {
2744 out.push(u8::from(*value));
2745 }
2746 }
2747 (LogicalType::Varchar, Data::Varlen(values)) => {
2748 let mut bytes = Vec::new();
2749 put_u32(&mut out, 0);
2750 for row in 0..vector.len() {
2751 let value = values.bytes(row).ok_or_else(|| invalid("string view is invalid"))?;
2752 bytes.extend_from_slice(value);
2753 put_u32(
2754 &mut out,
2755 u32::try_from(bytes.len())
2756 .map_err(|_| invalid("string payload exceeds 4GiB"))?,
2757 );
2758 }
2759 out.extend_from_slice(&bytes);
2760 }
2761 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
2762 }
2763 Ok((out, membership))
2764}
2765
2766fn put_varint(out: &mut Vec<u8>, mut value: u32) {
2767 while value >= 0x80 {
2768 out.push((value as u8 & 0x7f) | 0x80);
2769 value >>= 7;
2770 }
2771 out.push(value as u8);
2772}
2773
2774fn unique_codes(codes: &[u32]) -> Vec<u32> {
2776 let mut unique = codes.to_vec();
2777 unique.sort_unstable();
2778 unique.dedup();
2779 unique
2780}
2781
2782fn merged_codes(lists: Vec<Vec<u32>>) -> Vec<u32> {
2788 let mut lists = lists;
2789 while lists.len() > 1 {
2790 let mut next = Vec::with_capacity(lists.len().div_ceil(2));
2791 for pair in lists.chunks(2) {
2792 match pair {
2793 [left, right] => next.push(merged_pair(left, right)),
2794 [only] => next.push(only.clone()),
2795 _ => {}
2796 }
2797 }
2798 lists = next;
2799 }
2800 lists.pop().unwrap_or_default()
2801}
2802
2803fn merged_pair(left: &[u32], right: &[u32]) -> Vec<u32> {
2804 let mut out = Vec::with_capacity(left.len().saturating_add(right.len()));
2805 let mut at = 0;
2806 let mut to = 0;
2807 while at < left.len() && to < right.len() {
2808 match left[at].cmp(&right[to]) {
2809 Ordering::Less => {
2810 out.push(left[at]);
2811 at += 1;
2812 }
2813 Ordering::Greater => {
2814 out.push(right[to]);
2815 to += 1;
2816 }
2817 Ordering::Equal => {
2818 out.push(left[at]);
2819 at += 1;
2820 to += 1;
2821 }
2822 }
2823 }
2824 out.extend_from_slice(&left[at..]);
2825 out.extend_from_slice(&right[to..]);
2826 out
2827}
2828
2829fn merged_range(ranges: impl Iterator<Item = Range>) -> Range {
2834 let mut merged = Range::default();
2835 let mut first = true;
2836 for range in ranges {
2837 merged.nulls = merged.nulls.saturating_add(range.nulls);
2838 merged.sum = match (merged.sum.take(), range.sum) {
2842 (Some(held), Some(next)) if !first => held.checked_add(next),
2843 (_, next) if first => next,
2844 _ => None,
2845 };
2846 merged.exact = if first { range.exact } else { merged.exact && range.exact };
2847 if first {
2848 merged.low = range.low;
2849 merged.high = range.high;
2850 first = false;
2851 continue;
2852 }
2853 merged.low = match (merged.low.take(), range.low) {
2854 (Some(held), Some(next)) => Some(held.smaller(next)),
2855 _ => None,
2856 };
2857 merged.high = match (merged.high.take(), range.high) {
2858 (Some(held), Some(next)) => Some(held.larger(next)),
2859 _ => None,
2860 };
2861 }
2862 merged
2863}
2864
2865fn encode_sieves<'a>(sieves: impl Iterator<Item = &'a Option<Sieve>>) -> Result<Vec<u8>> {
2871 let held: Vec<&Option<Sieve>> = sieves.collect();
2872 let mut out = Vec::new();
2873 put_u32(
2874 &mut out,
2875 u32::try_from(held.len()).map_err(|_| invalid("too many parts in a stripe"))?,
2876 );
2877 for sieve in &held {
2878 let length = sieve.as_ref().map_or(0, Sieve::len);
2879 put_u32(&mut out, u32::try_from(length).map_err(|_| invalid("sieve length overflow"))?);
2880 }
2881 for sieve in held.into_iter().flatten() {
2883 out.extend_from_slice(&sieve.to_bytes());
2884 }
2885 Ok(out)
2886}
2887
2888fn decode_sieves(bytes: &[u8]) -> Result<Vec<Option<Sieve>>> {
2894 let parts = u32::from_le_bytes(
2895 bytes
2896 .get(..4)
2897 .ok_or_else(|| invalid("sieve page is truncated"))?
2898 .try_into()
2899 .map_err(|_| invalid("sieve page is truncated"))?,
2900 ) as usize;
2901 let mut lengths = Vec::with_capacity(parts);
2902 for part in 0..parts {
2903 let at = 4 + part * 4;
2904 let field = bytes.get(at..at + 4).ok_or_else(|| invalid("sieve page is truncated"))?;
2905 lengths.push(u32::from_le_bytes(
2906 field.try_into().map_err(|_| invalid("sieve page is truncated"))?,
2907 ) as usize);
2908 }
2909 let mut at = 4 + parts * 4;
2910 let mut out = Vec::with_capacity(parts);
2911 for length in lengths {
2912 if length == 0 {
2913 out.push(None);
2914 continue;
2915 }
2916 let end = at.checked_add(length).ok_or_else(|| invalid("sieve page is truncated"))?;
2917 let field = bytes.get(at..end).ok_or_else(|| invalid("sieve page is truncated"))?;
2918 out.push(Sieve::from_bytes(field));
2919 at = end;
2920 }
2921 if at != bytes.len() {
2922 return Err(invalid("sieve page has trailing bytes"));
2923 }
2924 Ok(out)
2925}
2926
2927fn encode_membership(unique: &[u32]) -> Vec<u8> {
2933 let mut out = Vec::with_capacity(unique.len().saturating_mul(2).saturating_add(5));
2934 put_varint(&mut out, u32::try_from(unique.len()).unwrap_or(u32::MAX));
2935 let mut previous = 0;
2936 for (at, &code) in unique.iter().enumerate() {
2937 put_varint(&mut out, if at == 0 { code } else { code - previous });
2938 previous = code;
2939 }
2940 out
2941}
2942
2943fn take_varint(bytes: &[u8], at: &mut usize) -> Result<u32> {
2944 let mut value = 0_u32;
2945 for shift in (0..35).step_by(7) {
2946 let byte = *bytes.get(*at).ok_or_else(|| invalid("membership varint is truncated"))?;
2947 *at += 1;
2948 let part = u32::from(byte & 0x7f);
2949 if shift == 28 && part > 0x0f {
2950 return Err(invalid("membership varint overflow"));
2951 }
2952 value = value
2953 .checked_add(
2954 part.checked_shl(shift).ok_or_else(|| invalid("membership varint overflow"))?,
2955 )
2956 .ok_or_else(|| invalid("membership varint overflow"))?;
2957 if byte & 0x80 == 0 {
2958 return Ok(value);
2959 }
2960 }
2961 Err(invalid("membership varint is too long"))
2962}
2963
2964fn decode_membership(bytes: &[u8]) -> Result<Vec<u32>> {
2965 let mut at = 0;
2966 let count = take_varint(bytes, &mut at)? as usize;
2967 let mut codes = Vec::with_capacity(count);
2968 let mut previous = 0_u32;
2969 for index in 0..count {
2970 let delta = take_varint(bytes, &mut at)?;
2971 let code = if index == 0 {
2972 delta
2973 } else {
2974 previous.checked_add(delta).ok_or_else(|| invalid("membership code overflow"))?
2975 };
2976 if index > 0 && code <= previous {
2977 return Err(invalid("membership codes are not increasing"));
2978 }
2979 codes.push(code);
2980 previous = code;
2981 }
2982 if at != bytes.len() {
2983 return Err(invalid("membership page has trailing bytes"));
2984 }
2985 Ok(codes)
2986}
2987
2988fn string_dictionary(vector: &Vector) -> Result<Option<Vec<u8>>> {
2989 let mut by_text = HashMap::new();
2990 let mut values = Vec::new();
2991 let mut codes = Vec::with_capacity(vector.len());
2992 let mut plain_bytes = 0_usize;
2993 for row in 0..vector.len() {
2994 let text = vector.text_at(row).unwrap_or("");
2995 plain_bytes = plain_bytes.saturating_add(text.len());
2996 let code = match by_text.get(text) {
2997 Some(&code) => code,
2998 None => {
2999 let code = u32::try_from(values.len())
3000 .map_err(|_| invalid("too many dictionary values"))?;
3001 by_text.insert(text, code);
3002 values.push(text);
3003 code
3004 }
3005 };
3006 codes.push(code);
3007 }
3008 let dictionary_bytes = values.iter().map(|value| value.len()).sum::<usize>();
3009 let encoded = 8_usize
3010 .saturating_add((values.len() + 1).saturating_mul(4))
3011 .saturating_add(dictionary_bytes)
3012 .saturating_add(codes.len().saturating_mul(4));
3013 let plain = (vector.len() + 1).saturating_mul(4).saturating_add(plain_bytes);
3014 if encoded >= plain {
3015 return Ok(None);
3016 }
3017 let mut out = Vec::with_capacity(encoded);
3018 put_u32(
3019 &mut out,
3020 u32::try_from(values.len()).map_err(|_| invalid("too many dictionary values"))?,
3021 );
3022 put_u32(
3023 &mut out,
3024 u32::try_from(dictionary_bytes).map_err(|_| invalid("dictionary payload exceeds 4GiB"))?,
3025 );
3026 let mut offset = 0_u32;
3027 put_u32(&mut out, offset);
3028 for value in &values {
3029 offset = offset
3030 .checked_add(
3031 u32::try_from(value.len()).map_err(|_| invalid("dictionary value is too long"))?,
3032 )
3033 .ok_or_else(|| invalid("dictionary payload exceeds 4GiB"))?;
3034 put_u32(&mut out, offset);
3035 }
3036 for value in values {
3037 out.extend_from_slice(value.as_bytes());
3038 }
3039 for code in codes {
3040 put_u32(&mut out, code);
3041 }
3042 Ok(Some(out))
3043}
3044
3045struct EncodedDictionary {
3046 index: Vec<u8>,
3047 ranks: Vec<u8>,
3048 payload: Vec<u8>,
3049}
3050
3051fn head(bytes: &[u8]) -> u64 {
3053 let mut word = [0; 8];
3054 let take = bytes.len().min(8);
3055 word[..take].copy_from_slice(&bytes[..take]);
3056 u64::from_be_bytes(word)
3057}
3058
3059fn rankings(dictionaries: &[Option<GlobalDictionary>]) -> Result<Vec<Vec<(u64, u32)>>> {
3067 let present =
3068 dictionaries.iter().enumerate().filter(|(_, held)| held.is_some()).map(|(at, _)| at);
3069 let present = present.collect::<Vec<_>>();
3070 let mut orders = vec![Vec::new(); dictionaries.len()];
3071 let workers = std::thread::available_parallelism()
3072 .map_or(1, usize::from)
3073 .min(MAX_FREQUENCY_WORKERS)
3074 .min(present.len());
3075 if workers <= 1 {
3076 for at in present {
3077 if let Some(dictionary) = &dictionaries[at] {
3078 orders[at] = dictionary.ranked();
3079 }
3080 }
3081 return Ok(orders);
3082 }
3083 let width = present.len().div_ceil(workers);
3084 let pieces = std::thread::scope(|scope| {
3085 present
3086 .chunks(width)
3087 .map(|columns| {
3088 scope.spawn(|| {
3089 columns
3090 .iter()
3091 .filter_map(|&at| dictionaries[at].as_ref().map(|held| (at, held.ranked())))
3092 .collect::<Vec<_>>()
3093 })
3094 })
3095 .collect::<Vec<_>>()
3096 .into_iter()
3097 .map(|handle| {
3098 handle.join().map_err(|_| Error::internal("a dictionary sort worker panicked"))
3099 })
3100 .collect::<Result<Vec<_>>>()
3101 })?;
3102 for piece in pieces {
3103 for (at, order) in piece {
3104 orders[at] = order;
3105 }
3106 }
3107 Ok(orders)
3108}
3109
3110fn encode_global_dictionary(
3111 dictionary: GlobalDictionary,
3112 order: &[(u64, u32)],
3113) -> Result<EncodedDictionary> {
3114 let values = dictionary.offsets.len() - 1;
3115 if order.len() != values {
3116 return Err(invalid("global dictionary order does not cover its values"));
3117 }
3118 let payload_len = dictionary.payload.len();
3119 let blocks = payload_len.div_ceil(TEXT_PAYLOAD_BLOCK);
3120 let ranks = encode_ranks(order);
3121 let rank_blocks = values.div_ceil(TEXT_RANK_BLOCK);
3122 let mut index = Vec::with_capacity(12 + (values + 1) * 4 + (blocks + rank_blocks) * 8);
3123 put_u32(
3124 &mut index,
3125 u32::try_from(values).map_err(|_| invalid("global dictionary has too many values"))?,
3126 );
3127 put_u32(&mut index, TEXT_PAYLOAD_BLOCK as u32);
3128 put_u32(
3129 &mut index,
3130 u32::try_from(blocks).map_err(|_| invalid("global dictionary has too many blocks"))?,
3131 );
3132 for offset in dictionary.offsets {
3133 put_u32(&mut index, offset);
3134 }
3135 for block in dictionary.payload.chunks(TEXT_PAYLOAD_BLOCK) {
3136 put_u64(&mut index, checksum(block));
3137 }
3138 for block in ranks.chunks(TEXT_RANK_BLOCK * RANK_ENTRY) {
3139 put_u64(&mut index, checksum(block));
3140 }
3141 Ok(EncodedDictionary { index, ranks, payload: dictionary.payload })
3142}
3143
3144fn encode_ranks(order: &[(u64, u32)]) -> Vec<u8> {
3151 let mut out = Vec::with_capacity(order.len() * RANK_ENTRY);
3152 for block in order.chunks(TEXT_RANK_BLOCK) {
3153 for &(head, _) in block {
3154 put_u64(&mut out, head);
3155 }
3156 for &(_, code) in block {
3157 put_u32(&mut out, code);
3158 }
3159 }
3160 out
3161}
3162
3163fn open_global_dictionary(file: Arc<File>, page: Page, ty: &LogicalType) -> Result<Vector> {
3164 if ty != &LogicalType::Varchar {
3165 return Err(invalid("global dictionary belongs to a non-string column"));
3166 }
3167 let mut header = [0; 12];
3168 read_at(&file, page.offset, &mut header)?;
3169 let count = u32::from_le_bytes(header[0..4].try_into().expect("four bytes")) as usize;
3170 let block_size = u32::from_le_bytes(header[4..8].try_into().expect("four bytes")) as usize;
3171 let blocks = u32::from_le_bytes(header[8..12].try_into().expect("four bytes")) as usize;
3172 if block_size != TEXT_PAYLOAD_BLOCK {
3173 return Err(invalid("global dictionary block width differs"));
3174 }
3175 let offset_len = (count + 1)
3176 .checked_mul(4)
3177 .ok_or_else(|| invalid("global dictionary offset count overflow"))?;
3178 let ranks = count;
3183 let rank_blocks = ranks.div_ceil(TEXT_RANK_BLOCK);
3184 let rank_len =
3185 ranks.checked_mul(RANK_ENTRY).ok_or_else(|| invalid("global dictionary rank overflow"))?;
3186 let hash_len = blocks
3187 .checked_add(rank_blocks)
3188 .and_then(|count| count.checked_mul(8))
3189 .ok_or_else(|| invalid("global dictionary block count overflow"))?;
3190 let index_len = 12usize
3191 .checked_add(offset_len)
3192 .and_then(|len| len.checked_add(hash_len))
3193 .ok_or_else(|| invalid("global dictionary header overflow"))?;
3194 let body_len = index_len
3195 .checked_add(rank_len)
3196 .ok_or_else(|| invalid("global dictionary header overflow"))?;
3197 if body_len > page.length as usize {
3198 return Err(invalid("global dictionary offset index exceeds its page"));
3199 }
3200 let mut index = vec![0; index_len];
3201 index[..12].copy_from_slice(&header);
3202 read_at(&file, page.offset + 12, &mut index[12..])?;
3203 if checksum(&index) != page.hash {
3204 return Err(invalid("global dictionary index checksum differs"));
3205 }
3206 let offsets = index[12..12 + offset_len]
3207 .chunks_exact(4)
3208 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
3209 .collect::<Vec<_>>();
3210 let mut hashes = index[12 + offset_len..]
3211 .chunks_exact(8)
3212 .map(|part| u64::from_le_bytes(part.try_into().expect("eight bytes")))
3213 .collect::<Vec<_>>();
3214 let rank_hashes = hashes.split_off(blocks);
3215 let payload_len = page.length as usize - body_len;
3216 if blocks != payload_len.div_ceil(TEXT_PAYLOAD_BLOCK) {
3217 return Err(invalid("global dictionary block count differs from its payload"));
3218 }
3219 if offsets.first() != Some(&0)
3220 || offsets.last().copied().map(|last| last as usize) != Some(payload_len)
3221 || offsets.windows(2).any(|pair| pair[0] > pair[1])
3222 {
3223 return Err(invalid("global dictionary offsets do not bound the payload"));
3224 }
3225 let payload_extents = (0..payload_len.div_ceil(TEXT_PAYLOAD_BLOCK * TEXT_PAYLOAD_EXTENT))
3226 .map(|_| OnceLock::new())
3227 .collect();
3228 let crossing = (0..count.div_ceil(TEXT_CROSSING_BLOCK)).map(|_| OnceLock::new()).collect();
3229 Vector::external_text(
3230 LogicalType::Varchar,
3231 Arc::new(NativeText {
3232 file,
3233 offsets,
3234 ranks,
3235 rank_at: page.offset + index_len as u64,
3236 rank_hashes,
3237 rank_blocks: (0..rank_blocks).map(|_| OnceLock::new()).collect(),
3238 payload: page.offset + body_len as u64,
3239 payload_len,
3240 hashes,
3241 payload_extents,
3242 crossing,
3243 }),
3244 )
3245}
3246
3247fn decode(
3248 ty: &LogicalType,
3249 rows: usize,
3250 bytes: &[u8],
3251 global: Option<Arc<Vector>>,
3252) -> Result<Vector> {
3253 let mut cur = Cursor { bytes, at: 0 };
3254 let codec = cur.u8()?;
3255 let flag = cur.u8()?;
3256 let validity = match flag {
3257 0 => Validity::AllValid,
3258 1 => Validity::AllInvalid,
3259 2 => {
3260 let mask = cur.take(rows.div_ceil(8))?;
3261 Validity::from_iter(rows, |row| mask[row / 8] >> (row % 8) & 1 == 1)
3262 }
3263 _ => return Err(invalid("page validity tag differs")),
3264 };
3265 if codec == 1 {
3266 if ty != &LogicalType::Varchar {
3267 return Err(invalid("dictionary codec belongs to a non-string page"));
3268 }
3269 let count = cur.u32()? as usize;
3270 let payload_len = cur.u32()? as usize;
3271 let offset_bytes = cur.take(
3272 (count + 1)
3273 .checked_mul(4)
3274 .ok_or_else(|| invalid("dictionary offset count overflow"))?,
3275 )?;
3276 let offsets = offset_bytes
3277 .chunks_exact(4)
3278 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
3279 .collect::<Vec<_>>();
3280 let payload = cur.take(payload_len)?.to_vec();
3281 if offsets.first() != Some(&0)
3282 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
3283 || offsets.windows(2).any(|pair| pair[0] > pair[1])
3284 {
3285 return Err(invalid("dictionary offsets do not bound the payload"));
3286 }
3287 let mut strings = StringColumn::over(Buffer::from_vec(payload));
3288 for pair in offsets.windows(2) {
3289 strings.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
3290 }
3291 let mut codes = Vec::with_capacity(rows);
3292 for _ in 0..rows {
3293 codes.push(cur.u32()?);
3294 }
3295 if codes.iter().any(|code| *code as usize >= count) {
3296 return Err(invalid("dictionary code is out of range"));
3297 }
3298 if cur.at != bytes.len() {
3299 return Err(invalid("dictionary page has trailing bytes"));
3300 }
3301 let dictionary = Vector::flat(LogicalType::Varchar, Data::Varlen(strings))?;
3302 return Ok(Vector::dictionary(codes, dictionary)?.with_validity(validity));
3303 }
3304 if codec == 3 {
3305 let dictionary = global.ok_or_else(|| invalid("global code page has no dictionary"))?;
3306 let mut codes = Vec::with_capacity(rows);
3307 let mut highest = None;
3308 for _ in 0..rows {
3309 let code = cur.u32()?;
3310 highest = Some(highest.map_or(code, |old: u32| old.max(code)));
3311 codes.push(code);
3312 }
3313 if cur.at != bytes.len() {
3314 return Err(invalid("global code page has trailing bytes"));
3315 }
3316 return Ok(Vector::stable_dictionary_validated(codes, dictionary, highest)?
3317 .with_validity(validity));
3318 }
3319 if codec == 2 {
3320 let width = u32::from(cur.u8()?);
3321 let base = i128::from_le_bytes(cur.take(16)?.try_into().expect("sixteen bytes"));
3322 let count = cur.u32()? as usize;
3323 let mut words = Vec::with_capacity(count);
3324 for _ in 0..count {
3325 words.push(cur.u64()?);
3326 }
3327 if cur.at != bytes.len() {
3328 return Err(invalid("packed page has trailing bytes"));
3329 }
3330 return Ok(Vector::packed(ty.clone(), words, width, base, rows)?.with_validity(validity));
3331 }
3332 if codec != 0 {
3333 return Err(invalid("page codec is unknown"));
3334 }
3335 let data = match ty {
3336 LogicalType::TinyInt => {
3337 let values = cur.take(rows)?;
3338 Data::Int8(values.iter().map(|item| *item as i8).collect::<Vec<_>>().into())
3339 }
3340 LogicalType::UTinyInt => Data::UInt8(cur.take(rows)?.to_vec().into()),
3341 LogicalType::SmallInt => {
3342 let values =
3343 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
3344 Data::Int16(
3345 values
3346 .chunks_exact(2)
3347 .map(|item| i16::from_le_bytes(item.try_into().expect("two bytes")))
3348 .collect::<Vec<_>>()
3349 .into(),
3350 )
3351 }
3352 LogicalType::USmallInt => {
3353 let values =
3354 cur.take(rows.checked_mul(2).ok_or_else(|| invalid("page size overflow"))?)?;
3355 Data::UInt16(
3356 values
3357 .chunks_exact(2)
3358 .map(|item| u16::from_le_bytes(item.try_into().expect("two bytes")))
3359 .collect::<Vec<_>>()
3360 .into(),
3361 )
3362 }
3363 LogicalType::UInteger => {
3364 let values =
3365 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
3366 Data::UInt32(
3367 values
3368 .chunks_exact(4)
3369 .map(|item| u32::from_le_bytes(item.try_into().expect("four bytes")))
3370 .collect::<Vec<_>>()
3371 .into(),
3372 )
3373 }
3374 LogicalType::UBigInt => {
3375 let values =
3376 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
3377 Data::UInt64(
3378 values
3379 .chunks_exact(8)
3380 .map(|item| u64::from_le_bytes(item.try_into().expect("eight bytes")))
3381 .collect::<Vec<_>>()
3382 .into(),
3383 )
3384 }
3385 LogicalType::Integer | LogicalType::Date => {
3386 let values =
3387 cur.take(rows.checked_mul(4).ok_or_else(|| invalid("page size overflow"))?)?;
3388 Data::Int32(
3389 values
3390 .chunks_exact(4)
3391 .map(|item| i32::from_le_bytes(item.try_into().expect("four bytes")))
3392 .collect::<Vec<_>>()
3393 .into(),
3394 )
3395 }
3396 LogicalType::BigInt | LogicalType::Timestamp => {
3397 let values =
3398 cur.take(rows.checked_mul(8).ok_or_else(|| invalid("page size overflow"))?)?;
3399 Data::Int64(
3400 values
3401 .chunks_exact(8)
3402 .map(|item| i64::from_le_bytes(item.try_into().expect("eight bytes")))
3403 .collect::<Vec<_>>()
3404 .into(),
3405 )
3406 }
3407 LogicalType::Boolean => {
3408 let values = cur.take(rows)?;
3409 if values.iter().any(|value| *value > 1) {
3410 return Err(invalid("boolean page has another value"));
3411 }
3412 Data::Bool(values.iter().map(|value| *value == 1).collect::<Vec<_>>().into())
3413 }
3414 LogicalType::Varchar => {
3415 let offset_bytes = cur
3416 .take((rows + 1).checked_mul(4).ok_or_else(|| invalid("offset count overflow"))?)?;
3417 let offsets = offset_bytes
3418 .chunks_exact(4)
3419 .map(|part| u32::from_le_bytes(part.try_into().expect("four bytes")))
3420 .collect::<Vec<_>>();
3421 let payload = cur.take(bytes.len() - cur.at)?.to_vec();
3422 if offsets.first() != Some(&0)
3423 || offsets.last().copied().map(|last| last as usize) != Some(payload.len())
3424 || offsets.windows(2).any(|pair| pair[0] > pair[1])
3425 {
3426 return Err(invalid("string offsets do not bound the payload"));
3427 }
3428 let mut values = StringColumn::over(Buffer::from_vec(payload));
3429 for pair in offsets.windows(2) {
3430 values.push_in_place(pair[0] as usize, (pair[1] - pair[0]) as usize)?;
3431 }
3432 Data::Varlen(values)
3433 }
3434 _ => return Err(Error::not_implemented(format!("native page for {ty}"))),
3435 };
3436 if cur.at != bytes.len() {
3437 return Err(invalid("page has trailing bytes"));
3438 }
3439 Ok(Vector::flat(ty.clone(), data)?.with_validity(validity))
3440}
3441
3442#[cfg(test)]
3443mod tests {
3444 use std::fs;
3445 use std::io::{Seek, SeekFrom, Write};
3446 use std::path::PathBuf;
3447 use std::time::{SystemTime, UNIX_EPOCH};
3448
3449 use rudb_common::Value;
3450 use rudb_common::bounds::Op;
3451
3452 use super::*;
3453
3454 #[test]
3455 fn checksum_matches_fixed_vectors() {
3456 assert_eq!(checksum(b""), 0xef46_db37_51d8_e999);
3457 assert_eq!(checksum(b"a"), 0xd24e_c4f1_a98c_6e5b);
3458 assert_eq!(checksum(b"abc"), 0x44bc_2cf5_ad77_0999);
3459 }
3460
3461 fn path(label: &str) -> PathBuf {
3462 let stamp = SystemTime::now().duration_since(UNIX_EPOCH).expect("time advances").as_nanos();
3463 std::env::temp_dir().join(format!("rudb-native-{label}-{}-{stamp}.rdb", std::process::id()))
3464 }
3465
3466 fn sample() -> Chunk {
3467 Chunk::new(vec![
3468 Vector::from_values(
3469 LogicalType::Integer,
3470 &[Value::Integer(4), Value::Integer(9), Value::Integer(-2)],
3471 )
3472 .expect("integers"),
3473 Vector::from_values(
3474 LogicalType::Varchar,
3475 &[
3476 Value::Varchar("alpha".into()),
3477 Value::Null,
3478 Value::Varchar("long text after a slash".into()),
3479 ],
3480 )
3481 .expect("strings"),
3482 ])
3483 .expect("matching rows")
3484 }
3485
3486 fn sample_ids() -> Chunk {
3487 Chunk::new(vec![
3488 Vector::flat(LogicalType::Integer, Data::Int32(vec![7, 8, 9].into()))
3489 .expect("integers"),
3490 ])
3491 .expect("one column")
3492 }
3493
3494 #[test]
3495 fn committed_file_reopens_and_reads_only_requested_columns() {
3496 let path = path("reopen");
3497 let mut writer = Writer::create(
3498 &path,
3499 "items",
3500 vec![
3501 Field::required("id", LogicalType::Integer),
3502 Field::new("text", LogicalType::Varchar),
3503 ],
3504 )
3505 .expect("new file");
3506 writer.append(&sample()).expect("first part");
3507 writer.append(&sample()).expect("second part");
3508 writer.finish().expect("commit");
3509 let reader = Reader::open(&path).expect("reopen from disk");
3510 assert_eq!(reader.table().rows(), 6);
3511 assert_eq!(reader.table().stripes().len(), 1);
3514 assert_eq!(reader.parts(), 2);
3515 assert_eq!(reader.part_rows(0), 3);
3516 assert_eq!(reader.part_rows(1), 3);
3517 let text = reader.read(1, &[1]).expect("only text page");
3518 assert_eq!(text.width(), 1);
3519 assert_eq!(text.value_at(1, 0), Value::Null);
3520 assert_eq!(text.value_at(2, 0), Value::Varchar("long text after a slash".into()));
3521 let sparse = reader.read_sparse(1, &[1]).expect("one part without its whole page");
3522 assert_eq!(sparse.width(), 1);
3523 assert_eq!(sparse.value_at(1, 0), Value::Null);
3524 assert_eq!(sparse.value_at(2, 0), Value::Varchar("long text after a slash".into()));
3525 assert!(!reader.skips_codes(0, 1, &[0]).expect("alpha is in the stripe"));
3526 assert!(!reader.skips_codes(0, 1, &[2]).expect("long text is in the stripe"));
3527 assert!(reader.skips_codes(0, 1, &[3]).expect("unknown code is absent"));
3528 let count = reader.read(0, &[]).expect("no page is needed for count");
3529 assert_eq!(count.len(), 3);
3530 assert!(reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }]));
3531 assert!(!reader.skips(0, &[Probe { column: 0, op: Op::Greater, value: Bound::Int(0) }]));
3532 let integers = reader.top_frequencies(0, 1).expect("valid integer synopsis").expect("kept");
3533 assert_eq!(
3534 integers,
3535 vec![(Value::Integer(-2), 2), (Value::Integer(4), 2), (Value::Integer(9), 2),]
3536 );
3537 let strings = reader.top_frequencies(1, 1).expect("valid string synopsis").expect("kept");
3538 assert_eq!(strings.len(), 3);
3539 assert!(strings.contains(&(Value::Null, 2)));
3540 assert!(strings.contains(&(Value::Varchar("alpha".into()), 2)));
3541 assert!(strings.contains(&(Value::Varchar("long text after a slash".into()), 2)));
3542 fs::remove_file(path).expect("remove scratch file");
3543 }
3544
3545 #[test]
3551 fn parts_past_the_stripe_bound_start_a_new_stripe() {
3552 let path = path("stripe-bound");
3553 let mut writer = Writer::create(
3554 &path,
3555 "items",
3556 vec![
3557 Field::required("id", LogicalType::Integer),
3558 Field::new("text", LogicalType::Varchar),
3559 ],
3560 )
3561 .expect("new file");
3562 let parts = STRIPE_PARTS * 2 + 3;
3563 for part in 0..parts {
3564 let id = part as i32;
3565 let chunk = Chunk::new(vec![
3566 Vector::from_values(
3567 LogicalType::Integer,
3568 &[Value::Integer(id), Value::Integer(-id)],
3569 )
3570 .expect("integers"),
3571 Vector::from_values(
3572 LogicalType::Varchar,
3573 &[Value::Varchar(format!("value {part}")), Value::Null],
3574 )
3575 .expect("strings"),
3576 ])
3577 .expect("matching rows");
3578 writer.append(&chunk).expect("one part");
3579 }
3580 writer.finish().expect("commit");
3581
3582 let reader = Reader::open(&path).expect("reopen from disk");
3583 assert_eq!(reader.parts(), parts);
3584 assert_eq!(reader.table().rows(), parts * 2);
3585 assert_eq!(reader.table().stripes().len(), parts.div_ceil(STRIPE_PARTS));
3586 assert_eq!(reader.table().stripes()[0].parts(), STRIPE_PARTS);
3587 assert_eq!(reader.table().stripes()[0].rows(), STRIPE_PARTS * 2);
3588 assert_eq!(reader.table().stripes()[2].parts(), 3);
3589 for part in (0..parts).rev() {
3592 let dense = reader.read(part, &[0, 1]).expect("a whole page read");
3593 let sparse = reader.read_sparse(part, &[0, 1]).expect("one part read");
3594 for chunk in [&dense, &sparse] {
3595 assert_eq!(chunk.len(), 2, "part {part} has its own row count");
3596 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
3597 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
3598 assert_eq!(chunk.value_at(0, 1), Value::Varchar(format!("value {part}")));
3599 assert_eq!(chunk.value_at(1, 1), Value::Null);
3600 }
3601 }
3602 let above = [Probe { column: 0, op: Op::Greater, value: Bound::Int(100) }];
3605 assert!(reader.skips(0, &above), "the first stripe stops at 63");
3606 assert!(!reader.skips(STRIPE_PARTS * 2, &above), "the third stripe reaches 130");
3607 fs::remove_file(path).expect("remove scratch file");
3608 }
3609
3610 fn scattered(n: i64) -> i64 {
3612 n.wrapping_mul(-7_046_029_254_386_353_131)
3613 }
3614
3615 #[test]
3621 fn a_part_is_skipped_when_its_sieve_does_not_hold_the_constant() {
3622 let path = path("sieve-skip");
3623 let mut writer =
3624 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
3625 .expect("new file");
3626 let parts = STRIPE_PARTS + 3;
3627 let per_part = 8;
3628 for part in 0..parts {
3629 let held: Vec<Value> = (0..per_part)
3630 .map(|row| Value::BigInt(scattered((part * per_part + row) as i64)))
3631 .collect();
3632 let chunk =
3633 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
3634 .expect("one column");
3635 writer.append(&chunk).expect("one part");
3636 }
3637 writer.finish().expect("commit");
3638
3639 let reader = Reader::open(&path).expect("reopen from disk");
3640 let probe = |value: i64| Probe {
3641 column: 0,
3642 op: Op::Equal,
3643 value: Bound::Int(i128::from(scattered(value))),
3644 };
3645 for wanted in [0_i64, 9, (parts * per_part - 1) as i64] {
3646 let tests = [probe(wanted)];
3647 let kept: Vec<usize> = (0..parts).filter(|&part| !reader.skips(part, &tests)).collect();
3648 let home = wanted as usize / per_part;
3649 assert_eq!(kept, vec![home], "only the part holding {wanted} is read");
3650 }
3651 let absent = [probe((parts * per_part) as i64 + 1)];
3652 assert!((0..parts).all(|part| reader.skips(part, &absent)), "no part holds it");
3653 let tests = [probe(0)];
3656 assert!(
3657 reader.table().stripes().iter().all(|stripe| !stripe.zone.skips(&tests)),
3658 "the bounds rule out no stripe at all"
3659 );
3660 fs::remove_file(path).expect("remove scratch file");
3661 }
3662
3663 #[test]
3669 fn a_damaged_sieve_page_is_read_through_rather_than_refused() {
3670 let path = path("sieve-damaged");
3671 let mut writer =
3672 Writer::create(&path, "hits", vec![Field::required("id", LogicalType::BigInt)])
3673 .expect("new file");
3674 let held: Vec<Value> = (0..8).map(|row| Value::BigInt(scattered(row))).collect();
3675 let chunk =
3676 Chunk::new(vec![Vector::from_values(LogicalType::BigInt, &held).expect("numbers")])
3677 .expect("one column");
3678 writer.append(&chunk).expect("one part");
3679 writer.finish().expect("commit");
3680
3681 let page =
3682 Reader::open(&path).expect("reopen").table.stripes[0].sieves[0].expect("a sieve page");
3683 let mut file = OpenOptions::new().write(true).open(&path).expect("open the sieve page");
3684 file.seek(SeekFrom::Start(page.offset + u64::from(page.length) - 1)).expect("seek");
3685 file.write_all(&[0xff]).expect("damage one byte");
3686 drop(file);
3687
3688 let reader = Reader::open(&path).expect("reopen the damaged file");
3689 let absent =
3690 [Probe { column: 0, op: Op::Equal, value: Bound::Int(i128::from(scattered(99))) }];
3691 assert!(!reader.skips(0, &absent), "a sieve that cannot be read skips nothing");
3692 assert_eq!(reader.read(0, &[0]).expect("the rows are untouched").len(), 8);
3693 fs::remove_file(path).expect("remove scratch file");
3694 }
3695
3696 #[test]
3707 fn workers_that_want_the_same_stripe_read_it_once() {
3708 let path = path("single-flight");
3709 let mut writer =
3710 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
3711 .expect("new file");
3712 for part in 0..STRIPE_PARTS {
3713 let id = part as i32;
3714 let chunk = Chunk::new(vec![
3715 Vector::from_values(
3716 LogicalType::Integer,
3717 &[Value::Integer(id), Value::Integer(-id)],
3718 )
3719 .expect("integers"),
3720 ])
3721 .expect("matching rows");
3722 writer.append(&chunk).expect("one part");
3723 }
3724 writer.finish().expect("commit");
3725
3726 let reader = Reader::open(&path).expect("reopen from disk");
3727 assert_eq!(reader.table().stripes().len(), 1, "one stripe is the point of the test");
3728 let barrier = std::sync::Barrier::new(8);
3729 std::thread::scope(|scope| {
3730 for worker in 0..8 {
3731 let reader = &reader;
3732 let barrier = &barrier;
3733 scope.spawn(move || {
3734 barrier.wait();
3735 for part in (worker..STRIPE_PARTS).step_by(8) {
3736 let chunk = reader.read(part, &[0]).expect("a whole page read");
3737 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
3738 assert_eq!(chunk.value_at(1, 0), Value::Integer(-(part as i32)));
3739 }
3740 });
3741 }
3742 });
3743 assert_eq!(reader.pages.load(Atomic::Relaxed), 1, "one stripe, one page read, whoever won");
3744 fs::remove_file(path).expect("remove scratch file");
3745 }
3746
3747 #[test]
3755 fn an_index_is_read_once_per_stripe_however_often_the_page_is_evicted() {
3756 let path = path("index-cache");
3757 let mut writer =
3758 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
3759 .expect("new file");
3760 let parts = STRIPE_PARTS * (CACHED_STRIPES_PER_COLUMN + 2);
3761 for part in 0..parts {
3762 let id = part as i32;
3763 let chunk = Chunk::new(vec![
3764 Vector::from_values(LogicalType::Integer, &[Value::Integer(id)]).expect("integers"),
3765 ])
3766 .expect("matching rows");
3767 writer.append(&chunk).expect("one part");
3768 }
3769 writer.finish().expect("commit");
3770
3771 let reader = Reader::open(&path).expect("reopen from disk");
3772 let stripes = reader.table().stripes().len();
3773 assert!(stripes > CACHED_STRIPES_PER_COLUMN, "the page cache has to be too small for this");
3774 for _ in 0..2 {
3776 for part in 0..parts {
3777 let chunk = reader.read(part, &[0]).expect("a part");
3778 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
3779 }
3780 }
3781 assert_eq!(reader.indexes.load(Atomic::Relaxed), stripes, "one index read per stripe");
3782 assert!(
3783 reader.pages.load(Atomic::Relaxed) > stripes,
3784 "the pages are the ones that get read again, which is what makes the index count mean \
3785 something"
3786 );
3787 fs::remove_file(path).expect("remove scratch file");
3788 }
3789
3790 #[test]
3799 fn a_worker_per_stripe_reads_its_page_once_when_the_cache_was_told_to_expect_it() {
3800 let workers = CACHED_STRIPES_PER_COLUMN + 4;
3801 let path = path("stripe-per-worker");
3802 let mut writer =
3803 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
3804 .expect("new file");
3805 for part in 0..STRIPE_PARTS * workers {
3806 let chunk = Chunk::new(vec![
3807 Vector::from_values(LogicalType::Integer, &[Value::Integer(part as i32)])
3808 .expect("integers"),
3809 ])
3810 .expect("matching rows");
3811 writer.append(&chunk).expect("one part");
3812 }
3813 writer.finish().expect("commit");
3814
3815 let read = |told: bool| {
3816 let reader = Reader::open(&path).expect("reopen from disk");
3817 assert_eq!(reader.table().stripes().len(), workers, "a stripe per worker");
3818 if told {
3819 reader.keep_stripes(workers);
3820 }
3821 let barrier = std::sync::Barrier::new(workers);
3822 std::thread::scope(|scope| {
3823 for (worker, run) in reader.stripe_parts().into_iter().enumerate() {
3824 let reader = &reader;
3825 let barrier = &barrier;
3826 scope.spawn(move || {
3827 for part in run {
3828 barrier.wait();
3829 let chunk = reader.read(part, &[0]).expect("a part of my own stripe");
3830 assert_eq!(chunk.value_at(0, 0), Value::Integer(part as i32));
3831 }
3832 assert!(worker < workers);
3833 });
3834 }
3835 });
3836 reader.pages.load(Atomic::Relaxed)
3837 };
3838
3839 assert_eq!(read(true), workers, "one page read per stripe and no more");
3840 assert!(read(false) > workers, "a cache that small is read again on every part");
3841 fs::remove_file(path).expect("remove scratch file");
3842 }
3843
3844 #[test]
3849 fn a_damaged_index_page_is_an_error() {
3850 let path = path("damaged-index");
3851 let mut writer =
3852 Writer::create(&path, "items", vec![Field::required("id", LogicalType::Integer)])
3853 .expect("new file");
3854 writer.append(&sample_ids()).expect("first part");
3855 writer.append(&sample_ids()).expect("second part");
3856 writer.finish().expect("commit");
3857
3858 let reader = Reader::open(&path).expect("valid directory");
3859 let index = reader.table.stripes[0].index;
3860 let mut byte = [0; 1];
3861 read_at(&reader.file, index.offset, &mut byte).expect("the first part length");
3862 let mut file = OpenOptions::new().write(true).open(&path).expect("open index page");
3863 file.seek(SeekFrom::Start(index.offset)).expect("index start");
3864 file.write_all(&[!byte[0]]).expect("damage the first part length");
3865 let error = reader.read(1, &[0]).expect_err("a damaged index must not be used");
3866 assert!(error.message().contains("index page section checksum differs"), "{error}");
3867 fs::remove_file(path).expect("remove scratch file");
3868 }
3869
3870 #[test]
3877 fn every_integer_width_round_trips_through_a_page() {
3878 let path = path("integer-widths");
3879 let columns = [
3880 (LogicalType::TinyInt, vec![Value::TinyInt(i8::MIN), Value::TinyInt(i8::MAX)]),
3881 (LogicalType::UTinyInt, vec![Value::UTinyInt(0), Value::UTinyInt(u8::MAX)]),
3882 (LogicalType::SmallInt, vec![Value::SmallInt(i16::MIN), Value::SmallInt(i16::MAX)]),
3883 (LogicalType::USmallInt, vec![Value::USmallInt(0), Value::USmallInt(u16::MAX)]),
3884 (LogicalType::Integer, vec![Value::Integer(i32::MIN), Value::Integer(i32::MAX)]),
3885 (LogicalType::UInteger, vec![Value::UInteger(0), Value::UInteger(u32::MAX)]),
3886 (LogicalType::BigInt, vec![Value::BigInt(i64::MIN), Value::BigInt(i64::MAX)]),
3887 (LogicalType::UBigInt, vec![Value::UBigInt(0), Value::UBigInt(u64::MAX)]),
3888 ];
3889 let fields = columns
3890 .iter()
3891 .enumerate()
3892 .map(|(at, (ty, _))| Field::required(format!("c{at}"), ty.clone()))
3893 .collect::<Vec<_>>();
3894 let vectors = columns
3895 .iter()
3896 .map(|(ty, values)| Vector::from_values(ty.clone(), values).expect("a vector"))
3897 .collect::<Vec<_>>();
3898 let mut writer = Writer::create(&path, "widths", fields).expect("new file");
3899 writer.append(&Chunk::new(vectors).expect("matching rows")).expect("one stripe");
3900 writer.finish().expect("commit");
3901
3902 let reader = Reader::open(&path).expect("reopen from disk");
3903 let wanted = (0..columns.len()).collect::<Vec<_>>();
3904 let read = reader.read(0, &wanted).expect("every column");
3905 assert_eq!(read.len(), 2);
3906 for (at, (ty, values)) in columns.iter().enumerate() {
3908 assert_eq!(read.value_at(0, at), values[0], "the low end of {ty}");
3909 assert_eq!(read.value_at(1, at), values[1], "the high end of {ty}");
3910 }
3911 fs::remove_file(path).expect("remove scratch file");
3912 }
3913
3914 #[test]
3915 fn numeric_frequency_candidates_keep_bounded_row_ordinals() {
3916 let path = path("frequency-ordinals");
3917 let mut writer =
3918 Writer::create(&path, "items", vec![Field::required("id", LogicalType::BigInt)])
3919 .expect("new file");
3920 let mut values = Vec::new();
3921 for leader in 0..10_i64 {
3922 values.extend(std::iter::repeat_n(leader, 100));
3923 }
3924 values.extend(1_000_i64..41_000);
3925 for part in values.chunks(1_024) {
3926 let vector = Vector::flat(LogicalType::BigInt, Data::Int64(part.to_vec().into()))
3927 .expect("big integers");
3928 writer.append(&Chunk::new(vec![vector]).expect("one column")).expect("one stripe");
3929 }
3930 writer.finish().expect("commit");
3931
3932 let reader = Reader::open(&path).expect("reopen from disk");
3933 let occurrences =
3934 reader.frequency_occurrences(0).expect("valid metadata").expect("bounded ordinals");
3935 assert!(occurrences.omitted_max < 100);
3936 assert!(occurrences.ordinals.len() <= FREQUENCY_ORDINALS);
3937 assert!(occurrences.ordinals.windows(2).all(|pair| pair[0] < pair[1]));
3938 assert_eq!(&occurrences.ordinals[..1_000], &(0_u64..1_000).collect::<Vec<_>>());
3939 fs::remove_file(path).expect("remove scratch file");
3940 }
3941
3942 #[test]
3943 fn an_unfinished_or_damaged_file_does_not_answer_with_partial_rows() {
3944 let unfinished = path("unfinished");
3945 let mut writer =
3946 Writer::create(&unfinished, "items", vec![Field::new("id", LogicalType::Integer)])
3947 .expect("new file");
3948 let chunk = Chunk::new(vec![
3949 Vector::flat(LogicalType::Integer, Data::Int32(vec![1, 2, 3].into()))
3950 .expect("integers"),
3951 ])
3952 .expect("chunk");
3953 writer.append(&chunk).expect("page written");
3954 drop(writer);
3955 assert!(Reader::open(&unfinished).is_err(), "no directory was committed");
3956 fs::remove_file(unfinished).expect("remove scratch file");
3957
3958 let damaged = path("damaged");
3959 let mut writer =
3960 Writer::create(&damaged, "items", vec![Field::new("id", LogicalType::Integer)])
3961 .expect("new file");
3962 writer.append(&chunk).expect("page written");
3963 writer.finish().expect("commit");
3964 let reader = Reader::open(&damaged).expect("valid directory");
3965 let mut file =
3966 OpenOptions::new().write(true).open(&damaged).expect("open for a damaged page");
3967 file.seek(SeekFrom::Start(HEADER + 1)).expect("inside first page");
3968 file.write_all(&[255]).expect("damage one byte");
3969 assert!(reader.read(0, &[0]).is_err(), "page checksum rejects corruption");
3970 fs::remove_file(damaged).expect("remove scratch file");
3971 }
3972
3973 #[test]
3974 fn damaged_lazy_dictionary_payload_is_an_error() {
3975 let path = path("damaged-dictionary");
3976 let mut writer = Writer::create(
3977 &path,
3978 "items",
3979 vec![
3980 Field::required("id", LogicalType::Integer),
3981 Field::new("text", LogicalType::Varchar),
3982 ],
3983 )
3984 .expect("new file");
3985 writer.append(&sample()).expect("stripe written");
3986 writer.finish().expect("commit");
3987
3988 let reader = Reader::open(&path).expect("valid directory");
3989 let dictionary = reader.table.dictionaries[1].expect("string dictionary page");
3990 let mut header = [0; 12];
3993 read_at(&reader.file, dictionary.offset, &mut header).expect("dictionary header");
3994 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
3995 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
3996 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
3997 let index_len =
3998 12 + (count + 1) * 4 + (blocks + rank_blocks) * 8 + count * RANK_ENTRY as u64;
3999 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
4000 file.seek(SeekFrom::Start(dictionary.offset + index_len))
4001 .expect("inside dictionary payload");
4002 file.write_all(&[255]).expect("damage dictionary payload");
4003
4004 let chunk = reader.read(0, &[1]).expect("code page and dictionary index remain valid");
4005 let error =
4006 chunk.validate_external().expect_err("payload corruption must reach the caller");
4007 assert!(error.message().contains("payload checksum differs"), "{error}");
4008 fs::remove_file(path).expect("remove scratch file");
4009 }
4010
4011 #[test]
4018 fn a_dictionary_over_one_extent_checks_every_block_of_it() {
4019 let path = path("dictionary-extents");
4020 let value =
4021 |row: usize| format!("{row:07} a value long enough to be worth a payload block");
4022 let parts = 30;
4023 let per_part = 1000;
4024 let mut writer =
4025 Writer::create(&path, "items", vec![Field::required("text", LogicalType::Varchar)])
4026 .expect("new file");
4027 for part in 0..parts {
4028 let values = (0..per_part)
4029 .map(|row| Value::Varchar(value(part * per_part + row)))
4030 .collect::<Vec<_>>();
4031 let chunk = Chunk::new(vec![
4032 Vector::from_values(LogicalType::Varchar, &values).expect("strings"),
4033 ])
4034 .expect("matching rows");
4035 writer.append(&chunk).expect("a part");
4036 }
4037 writer.finish().expect("commit");
4038
4039 let reader = Reader::open(&path).expect("reopen from disk");
4040 let dictionary = reader.table.dictionaries[0].expect("string dictionary page");
4041 assert!(
4042 dictionary.length as usize > TEXT_PAYLOAD_BLOCK * TEXT_PAYLOAD_EXTENT,
4043 "the dictionary has to be over one extent for this to be testing anything"
4044 );
4045 for part in [0, parts - 1] {
4046 let chunk = reader.read(part, &[0]).expect("a part");
4047 chunk.validate_external().expect("every payload block checks out");
4048 assert_eq!(chunk.value_at(0, 0), Value::Varchar(value(part * per_part)));
4049 }
4050
4051 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
4052 file.seek(SeekFrom::Start(dictionary.offset + u64::from(dictionary.length) - 4))
4053 .expect("the last bytes of the page are payload");
4054 file.write_all(&[255]).expect("damage the last payload block");
4055 let reader = Reader::open(&path).expect("the directory and the index are untouched");
4056 let chunk = reader.read(parts - 1, &[0]).expect("the code page remains valid");
4057 let error = chunk.validate_external().expect_err("the damage must reach the caller");
4058 assert!(error.message().contains("payload checksum differs"), "{error}");
4059 fs::remove_file(path).expect("remove scratch file");
4060 }
4061
4062 #[test]
4067 fn a_damaged_sorted_order_is_an_error() {
4068 let path = path("damaged-order");
4069 let mut writer = Writer::create(
4070 &path,
4071 "items",
4072 vec![
4073 Field::required("id", LogicalType::Integer),
4074 Field::new("text", LogicalType::Varchar),
4075 ],
4076 )
4077 .expect("new file");
4078 writer.append(&sample()).expect("stripe written");
4079 writer.finish().expect("commit");
4080
4081 let reader = Reader::open(&path).expect("valid directory");
4082 let page = reader.table.dictionaries[1].expect("string dictionary page");
4083 let mut header = [0; 12];
4084 read_at(&reader.file, page.offset, &mut header).expect("dictionary header");
4085 let count = u64::from(u32::from_le_bytes(header[0..4].try_into().expect("four bytes")));
4086 let blocks = u64::from(u32::from_le_bytes(header[8..12].try_into().expect("four bytes")));
4087 let rank_blocks = count.div_ceil(TEXT_RANK_BLOCK as u64);
4088 let index_len = 12 + (count + 1) * 4 + (blocks + rank_blocks) * 8;
4089 let mut file = OpenOptions::new().write(true).open(&path).expect("open dictionary page");
4090 file.seek(SeekFrom::Start(page.offset + index_len)).expect("the first head");
4091 file.write_all(&[255]).expect("damage the order");
4092
4093 let dictionary = reader.dictionary(1).expect("read").expect("a string column has one");
4094 let error = dictionary.compare_rank(0, b"anything").expect_err("a damaged order is caught");
4095 assert!(error.message().contains("rank checksum differs"), "{error}");
4096 fs::remove_file(path).expect("remove scratch file");
4097 }
4098
4099 #[test]
4103 fn a_global_dictionary_carries_the_sorted_order_of_its_values() {
4104 let spellings = ["overlong1z", "b", "", "overlong1a", "overlong", "ab", "a", "overlong1"];
4107 let path = path("dictionary-order");
4108 let mut writer =
4109 Writer::create(&path, "items", vec![Field::new("text", LogicalType::Varchar)])
4110 .expect("new file");
4111 writer
4112 .append(
4113 &Chunk::new(vec![
4114 Vector::from_values(
4115 LogicalType::Varchar,
4116 &spellings.map(|text| Value::Varchar(text.into())),
4117 )
4118 .expect("strings"),
4119 ])
4120 .expect("one column"),
4121 )
4122 .expect("stripe written");
4123 writer.finish().expect("commit");
4124
4125 let reader = Reader::open(&path).expect("valid directory");
4126 let dictionary = reader.dictionary(0).expect("read").expect("a string column has one");
4127 let count = dictionary.ranks().expect("a v10 file stores one");
4128 assert_eq!(count, spellings.len(), "every distinct value has a rank");
4129 let order = (0..count)
4130 .map(|rank| dictionary.code_at_rank(rank).expect("a code"))
4131 .collect::<Vec<_>>();
4132 let mut seen = order.clone();
4133 seen.sort_unstable();
4134 assert_eq!(seen, (0..spellings.len() as u32).collect::<Vec<_>>(), "a permutation of codes");
4135
4136 let ranked = order
4137 .iter()
4138 .map(|&code| {
4139 dictionary.try_bytes_at(code as usize).expect("read").expect("a value").to_vec()
4140 })
4141 .collect::<Vec<_>>();
4142 let mut expected = spellings.map(|text| text.as_bytes().to_vec()).to_vec();
4143 expected.sort();
4144 assert_eq!(ranked, expected, "rank order is value order");
4145
4146 for (rank, value) in expected.iter().enumerate() {
4149 assert_eq!(
4150 dictionary.compare_rank(rank, value).expect("compare"),
4151 Ordering::Equal,
4152 "rank {rank} is its own value"
4153 );
4154 if rank > 0 {
4155 assert_eq!(
4156 dictionary.compare_rank(rank - 1, value).expect("compare"),
4157 Ordering::Less,
4158 "rank {rank} follows the one before it"
4159 );
4160 }
4161 }
4162 fs::remove_file(path).expect("remove scratch file");
4163 }
4164
4165 #[test]
4166 fn damaged_membership_cannot_skip_a_string_page() {
4167 let path = path("damaged-membership");
4168 let mut writer = Writer::create(
4169 &path,
4170 "items",
4171 vec![
4172 Field::required("id", LogicalType::Integer),
4173 Field::new("text", LogicalType::Varchar),
4174 ],
4175 )
4176 .expect("new file");
4177 writer.append(&sample()).expect("stripe written");
4178 writer.finish().expect("commit");
4179
4180 let reader = Reader::open(&path).expect("valid directory");
4181 let membership = reader.table.stripes[0].memberships[1].expect("string membership");
4182 let mut file = OpenOptions::new().write(true).open(&path).expect("open membership page");
4183 file.seek(SeekFrom::Start(membership.offset)).expect("membership start");
4184 file.write_all(&[255]).expect("damage membership");
4185 let error = reader.skips_codes(0, 1, &[3]).expect_err("corruption must not skip rows");
4186 assert!(error.message().contains("membership page checksum differs"), "{error}");
4187 fs::remove_file(path).expect("remove scratch file");
4188 }
4189
4190 #[test]
4191 fn membership_delta_stream_is_sorted_exact_and_bounded() {
4192 let unique = unique_codes(&[900, 4, 4, 72, 9, u32::MAX]);
4193 assert_eq!(unique, [4, 9, 72, 900, u32::MAX]);
4194 let encoded = encode_membership(&unique);
4195 assert_eq!(
4196 decode_membership(&encoded).expect("valid membership"),
4197 [4, 9, 72, 900, u32::MAX]
4198 );
4199 let merged = merged_codes(vec![vec![4, 900], vec![9, 900, u32::MAX], vec![72]]);
4202 assert_eq!(merged, [4, 9, 72, 900, u32::MAX]);
4203 assert_eq!(
4204 decode_membership(&encode_membership(&merged)).expect("valid membership"),
4205 unique
4206 );
4207 assert!(decode_membership(&[1, 0x80]).is_err(), "a truncated varint is invalid");
4208 assert!(
4209 decode_membership(&[1, 0xff, 0xff, 0xff, 0xff, 0x10]).is_err(),
4210 "a value past u32 is invalid"
4211 );
4212 }
4213
4214 #[test]
4215 fn a_global_dictionary_may_be_larger_than_one_column_page() {
4216 let dictionary = Page {
4217 offset: HEADER,
4218 length: u32::try_from(MAX_PAGE + 1).expect("the page bound fits on disk"),
4219 hash: 0,
4220 };
4221 let table = Table {
4222 name: "items".to_owned(),
4223 fields: vec![Field::new("text", LogicalType::Varchar)],
4224 stripes: Vec::new(),
4225 rows: 0,
4226 dictionaries: vec![Some(dictionary)],
4227 frequencies: vec![None],
4228 };
4229 let directory = encode_directory(&table).expect("directory");
4230 let file_size = dictionary.offset + u64::from(dictionary.length) + 1;
4231
4232 let decoded = decode_directory(&directory, file_size).expect("large lazy dictionary");
4233 assert_eq!(decoded.dictionaries[0].expect("dictionary").length, dictionary.length);
4234 }
4235}