1use std::fs::File;
4use std::os::fd::AsRawFd;
5use std::path::Path;
6#[cfg(any(test, feature = "test-support"))]
7use std::path::PathBuf;
8use std::ptr::NonNull;
9use std::sync::atomic::{AtomicU64, Ordering};
10use std::sync::{Arc, Mutex, OnceLock};
11
12use xxhash_rust::xxh3::xxh3_64;
13
14use crate::format::frame::{
15 FILE_HEADER_LEN, FILE_TRAILER_LEN, FormatCheck, FormatError, decode_header, read_u16, read_u32,
16 read_u64,
17};
18use crate::format::{FormatFamily, FormatRegistry};
19use crate::graph::block::{GraphNodeBlocks, ValidatedGraphNodeBlocks, decode_node_blocks};
20use crate::ingest::{DocId, DocumentVersion, Revision};
21use crate::meta::{AliveSet, ColumnStore};
22use crate::quant::Bit4Factors;
23use crate::vfs::Vfs;
24
25use super::layout::{
26 CHECKSUM_CHUNK_BYTES, Cursor, Int8Factors, REGION_ALIGNMENT, REGION_ENTRY_LEN, RegionEntry,
27 RegionKind, SEGMENT_PREFIX_LEN, VECTOR_HEADER_LEN, align_up, decode_alive, decode_columns,
28 decode_vector_header,
29};
30use super::{SegmentError, SegmentId, SegmentMeta};
31
32const MAX_SEGMENT_HEADER_BYTES: usize = 1024 * 1024;
33
34thread_local! {
35 static DATA_READ_AUDIT: std::cell::RefCell<Option<Arc<AtomicU64>>> = const {
36 std::cell::RefCell::new(None)
37 };
38}
39
40#[cfg(any(test, feature = "test-support"))]
41thread_local! {
42 static SEGMENT_COST_AUDIT: std::cell::RefCell<Option<Arc<SegmentCostCounters>>> = const {
43 std::cell::RefCell::new(None)
44 };
45}
46
47#[cfg(any(test, feature = "test-support"))]
48#[derive(Default)]
49struct SegmentCostCounters {
50 identity_hash_bytes: AtomicU64,
51 rescore_hash_bytes: AtomicU64,
52 int8_factor_decode_bytes: AtomicU64,
53 postings_hash_bytes: AtomicU64,
54 postings_decode_bytes: AtomicU64,
55 columns_hash_bytes: AtomicU64,
56 columns_decode_bytes: AtomicU64,
57 alive_hash_bytes: AtomicU64,
58 alive_decode_bytes: AtomicU64,
59}
60
61#[cfg(any(test, feature = "test-support"))]
63#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
64pub struct SegmentCostSnapshot {
65 pub identity_hash_bytes: u64,
67 pub rescore_hash_bytes: u64,
69 pub int8_factor_decode_bytes: u64,
71 pub postings_hash_bytes: u64,
73 pub postings_decode_bytes: u64,
75 pub columns_hash_bytes: u64,
77 pub columns_decode_bytes: u64,
79 pub alive_hash_bytes: u64,
81 pub alive_decode_bytes: u64,
83}
84
85#[cfg(any(test, feature = "test-support"))]
87#[derive(Clone, Default)]
88pub struct SegmentCostAudit {
89 counters: Arc<SegmentCostCounters>,
90}
91
92#[cfg(any(test, feature = "test-support"))]
93struct SegmentCostAuditGuard {
94 previous: Option<Arc<SegmentCostCounters>>,
95}
96
97#[cfg(any(test, feature = "test-support"))]
98impl Drop for SegmentCostAuditGuard {
99 fn drop(&mut self) {
100 let previous = self.previous.take();
101 SEGMENT_COST_AUDIT.with(|slot| {
102 slot.replace(previous);
103 });
104 }
105}
106
107#[cfg(any(test, feature = "test-support"))]
108impl SegmentCostAudit {
109 #[must_use]
111 pub fn new() -> Self {
112 Self::default()
113 }
114
115 pub fn measure<T>(&self, operation: impl FnOnce() -> T) -> T {
117 let previous =
118 SEGMENT_COST_AUDIT.with(|slot| slot.replace(Some(Arc::clone(&self.counters))));
119 let _guard = SegmentCostAuditGuard { previous };
120 operation()
121 }
122
123 #[must_use]
125 pub fn snapshot(&self) -> SegmentCostSnapshot {
126 SegmentCostSnapshot {
127 identity_hash_bytes: self.counters.identity_hash_bytes.load(Ordering::Relaxed),
128 rescore_hash_bytes: self.counters.rescore_hash_bytes.load(Ordering::Relaxed),
129 int8_factor_decode_bytes: self
130 .counters
131 .int8_factor_decode_bytes
132 .load(Ordering::Relaxed),
133 postings_hash_bytes: self.counters.postings_hash_bytes.load(Ordering::Relaxed),
134 postings_decode_bytes: self.counters.postings_decode_bytes.load(Ordering::Relaxed),
135 columns_hash_bytes: self.counters.columns_hash_bytes.load(Ordering::Relaxed),
136 columns_decode_bytes: self.counters.columns_decode_bytes.load(Ordering::Relaxed),
137 alive_hash_bytes: self.counters.alive_hash_bytes.load(Ordering::Relaxed),
138 alive_decode_bytes: self.counters.alive_decode_bytes.load(Ordering::Relaxed),
139 }
140 }
141}
142
143#[cfg(any(test, feature = "test-support"))]
144fn account_region_hash(kind: RegionKind, bytes: usize) {
145 let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
146 SEGMENT_COST_AUDIT.with(|slot| {
147 let Some(counters) = slot.borrow().as_ref().cloned() else {
148 return;
149 };
150 let counter = match kind {
151 RegionKind::DocumentVersions => &counters.identity_hash_bytes,
152 RegionKind::VectorRescore => &counters.rescore_hash_bytes,
153 RegionKind::Postings => &counters.postings_hash_bytes,
154 RegionKind::Columns => &counters.columns_hash_bytes,
155 RegionKind::Alive => &counters.alive_hash_bytes,
156 _ => return,
157 };
158 let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
159 Some(current.saturating_add(bytes))
160 });
161 });
162}
163
164#[cfg(any(test, feature = "test-support"))]
165fn account_region_decode(kind: RegionKind, bytes: usize) {
166 let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
167 SEGMENT_COST_AUDIT.with(|slot| {
168 let Some(counters) = slot.borrow().as_ref().cloned() else {
169 return;
170 };
171 let counter = match kind {
172 RegionKind::VectorFactors => &counters.int8_factor_decode_bytes,
173 RegionKind::Postings => &counters.postings_decode_bytes,
174 RegionKind::Columns => &counters.columns_decode_bytes,
175 RegionKind::Alive => &counters.alive_decode_bytes,
176 _ => return,
177 };
178 let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
179 Some(current.saturating_add(bytes))
180 });
181 });
182}
183
184#[cfg(any(test, feature = "test-support"))]
185pub(crate) fn account_active_postings_decode(bytes: usize) {
186 let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
187 SEGMENT_COST_AUDIT.with(|slot| {
188 if let Some(counters) = slot.borrow().as_ref() {
189 let _ = counters.postings_decode_bytes.fetch_update(
190 Ordering::Relaxed,
191 Ordering::Relaxed,
192 |current| Some(current.saturating_add(bytes)),
193 );
194 }
195 });
196}
197
198fn region_hash(_kind: RegionKind, bytes: &[u8]) -> u64 {
199 #[cfg(any(test, feature = "test-support"))]
200 account_region_hash(_kind, bytes.len());
201 xxh3_64(bytes)
202}
203
204pub(crate) struct DataReadAuditGuard {
205 previous: Option<Arc<AtomicU64>>,
206}
207
208impl Drop for DataReadAuditGuard {
209 fn drop(&mut self) {
210 let previous = self.previous.take();
211 DATA_READ_AUDIT.with(|slot| {
212 slot.replace(previous);
213 });
214 }
215}
216
217pub(crate) fn install_data_read_audit(counter: Option<Arc<AtomicU64>>) -> DataReadAuditGuard {
218 let previous = DATA_READ_AUDIT.with(|slot| slot.replace(counter));
219 DataReadAuditGuard { previous }
220}
221
222fn account_data_read(bytes: usize) {
223 let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
224 DATA_READ_AUDIT.with(|slot| {
225 if let Some(counter) = slot.borrow().as_ref() {
226 let _ = counter.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
227 Some(current.saturating_add(bytes))
228 });
229 }
230 });
231}
232
233struct MappedFile {
234 _file: File,
235 pointer: NonNull<u8>,
236 length: usize,
237}
238
239impl MappedFile {
240 fn open(file: File, path: &Path) -> Result<Self, SegmentError> {
241 let length = usize::try_from(
242 file.metadata()
243 .map_err(|error| SegmentError::io(path, error))?
244 .len(),
245 )
246 .map_err(|_| SegmentError::Geometry("mapped file length exceeds usize".to_owned()))?;
247 if length == 0 {
248 return Err(FormatError::new(
249 path.display().to_string(),
250 FormatCheck::Length,
251 "empty segment file",
252 )
253 .into());
254 }
255 let mapped = unsafe {
259 libc::mmap(
260 std::ptr::null_mut(),
261 length,
262 libc::PROT_READ,
263 libc::MAP_PRIVATE,
264 file.as_raw_fd(),
265 0,
266 )
267 };
268 if mapped == libc::MAP_FAILED {
269 return Err(SegmentError::io(path, std::io::Error::last_os_error()));
270 }
271 let pointer = NonNull::new(mapped.cast::<u8>()).ok_or_else(|| {
272 SegmentError::Geometry("mmap returned a null non-failure pointer".to_owned())
273 })?;
274 Ok(Self {
275 _file: file,
276 pointer,
277 length,
278 })
279 }
280
281 fn as_bytes(&self) -> &[u8] {
282 unsafe { std::slice::from_raw_parts(self.pointer.as_ptr(), self.length) }
285 }
286
287 #[cfg(any(test, feature = "test-support"))]
293 fn corrupt_byte(&mut self, offset: usize, mask: u8) -> std::io::Result<()> {
294 if offset >= self.length {
295 return Err(std::io::Error::other(
296 "corruption offset is outside the mapping",
297 ));
298 }
299 let page_size_raw = unsafe {
300 libc::sysconf(libc::_SC_PAGESIZE)
302 };
303 let page_size = usize::try_from(page_size_raw)
304 .ok()
305 .filter(|size| *size != 0)
306 .ok_or_else(|| std::io::Error::other("sysconf returned an invalid page size"))?;
307 let page_offset = offset - offset % page_size;
308 let page = self
309 .pointer
310 .as_ptr()
311 .wrapping_add(page_offset)
312 .cast::<libc::c_void>();
313 if unsafe { libc::mprotect(page, page_size, libc::PROT_READ | libc::PROT_WRITE) } != 0 {
316 return Err(std::io::Error::last_os_error());
317 }
318 unsafe {
321 let byte = self.pointer.as_ptr().add(offset);
322 byte.write(byte.read() ^ mask);
323 }
324 if unsafe { libc::mprotect(page, page_size, libc::PROT_READ) } != 0 {
326 return Err(std::io::Error::last_os_error());
327 }
328 Ok(())
329 }
330}
331
332impl Drop for MappedFile {
333 fn drop(&mut self) {
334 let _ = unsafe { libc::munmap(self.pointer.as_ptr().cast(), self.length) };
337 }
338}
339
340unsafe impl Send for MappedFile {}
342unsafe impl Sync for MappedFile {}
344
345impl crate::graph::search::RescoreValidator for SegmentReader {
346 fn validate_rows(&self, rows: &[u32]) -> Result<(), String> {
347 self.validate_rescore_rows(rows).map_err(|source| {
348 format!(
349 "exact scores unavailable for segment {}: {source}",
350 self.meta.id
351 )
352 })
353 }
354}
355
356struct ParsedHeader {
357 meta: SegmentMeta,
358 header_length: usize,
359 entries: Vec<RegionEntry>,
360}
361
362pub struct SegmentReader {
364 mapping: MappedFile,
365 meta: SegmentMeta,
366 #[cfg(any(test, feature = "test-support"))]
367 storage_store_directory: PathBuf,
368 header_length: usize,
369 entries: Vec<RegionEntry>,
370 document_versions_validation: OnceLock<Result<(), DocumentVersionValidationError>>,
371 stored_text_validated: OnceLock<()>,
375 vector_codes_validated: OnceLock<()>,
376 vector_factors_validated: OnceLock<()>,
377 rescore_valid_chunks: Mutex<Box<[u64]>>,
378 query_accounting: Option<Arc<crate::lifecycle::stats::Accounting>>,
379 int8_factors_cache: OnceLock<CachedQueryValue<Arc<Vec<crate::scan::Int8Factors>>>>,
380 postings_cache: OnceLock<CachedQueryValue<Arc<crate::fts::sealed::SealedSegment>>>,
381 columns_cache: OnceLock<CachedQueryValue<Arc<ColumnStore>>>,
382 alive_cache: OnceLock<CachedQueryValue<Arc<AliveSet>>>,
383 document_version_index_cache: OnceLock<CachedQueryValue<Arc<Box<[u32]>>>>,
386 pub(crate) graph_search_cache: crate::lifecycle::graph_cache::SegmentGraphSearchCache,
389 vector_ceiling_norm_range: OnceLock<crate::graph::search::GraphSegmentNormRange>,
394}
395
396struct CachedQueryValue<T> {
397 value: T,
398 _accounting: Option<crate::lifecycle::stats::AccountedCounter>,
399}
400
401#[derive(Clone, Debug)]
402enum DocumentVersionValidationError {
403 Geometry(String),
404 Checksum,
405}
406
407impl DocumentVersionValidationError {
408 fn into_segment_error(self, segment_id: SegmentId) -> SegmentError {
409 match self {
410 Self::Geometry(detail) => SegmentError::Geometry(detail),
411 Self::Checksum => FormatError::new(
412 format!("segment:{segment_id}:document-versions"),
413 FormatCheck::BlockChecksum,
414 "document-version region checksum mismatch",
415 )
416 .into(),
417 }
418 }
419}
420
421#[derive(Clone, Copy, Debug)]
423pub struct StoredMetadataRows<'a> {
424 offsets: &'a [u8],
425 bytes: &'a [u8],
426 row_count: usize,
427}
428
429impl<'a> StoredMetadataRows<'a> {
430 #[must_use]
432 pub const fn row_count(self) -> usize {
433 self.row_count
434 }
435
436 #[must_use]
438 pub fn row(self, row: usize) -> Option<&'a [u8]> {
439 if row >= self.row_count {
440 return None;
441 }
442 let start_index = row.checked_mul(8)?;
443 let end_index = row.checked_add(1)?.checked_mul(8)?;
444 let start = self
445 .offsets
446 .get(start_index..start_index.checked_add(8)?)?
447 .try_into()
448 .ok()
449 .map(u64::from_le_bytes)
450 .and_then(|value| usize::try_from(value).ok())?;
451 let end = self
452 .offsets
453 .get(end_index..end_index.checked_add(8)?)?
454 .try_into()
455 .ok()
456 .map(u64::from_le_bytes)
457 .and_then(|value| usize::try_from(value).ok())?;
458 self.bytes.get(start..end)
459 }
460}
461
462#[derive(Clone, Copy, Debug)]
464pub struct StoredTextRows<'a> {
465 present: &'a [u8],
466 offsets: &'a [u8],
467 bytes: &'a [u8],
468 row_count: usize,
469}
470
471impl<'a> StoredTextRows<'a> {
472 #[must_use]
474 pub const fn row_count(self) -> usize {
475 self.row_count
476 }
477
478 #[must_use]
481 pub fn row(self, row: usize) -> Option<Option<&'a str>> {
482 if row >= self.row_count {
483 return None;
484 }
485 let end_index = row.checked_mul(8)?;
486 let end = self
487 .offsets
488 .get(end_index..end_index.checked_add(8)?)?
489 .try_into()
490 .ok()
491 .map(u64::from_le_bytes)
492 .and_then(|value| usize::try_from(value).ok())?;
493 let start = if row == 0 {
494 0
495 } else {
496 let start_index = row.checked_sub(1)?.checked_mul(8)?;
497 self.offsets
498 .get(start_index..start_index.checked_add(8)?)?
499 .try_into()
500 .ok()
501 .map(u64::from_le_bytes)
502 .and_then(|value| usize::try_from(value).ok())?
503 };
504 let byte = *self.present.get(row / 8)?;
505 if byte & (1 << (row % 8)) == 0 {
506 return (start == end).then_some(None);
507 }
508 let text = std::str::from_utf8(self.bytes.get(start..end)?).ok()?;
509 Some(Some(text))
510 }
511}
512
513static QUERY_CHECKSUMS: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
516
517#[cfg(any(test, feature = "test-support"))]
522pub fn set_query_checksum_verification(enabled: bool) {
523 QUERY_CHECKSUMS.store(
524 if enabled { 2 } else { 1 },
525 std::sync::atomic::Ordering::Relaxed,
526 );
527}
528
529impl SegmentReader {
530 pub fn open(vfs: &dyn Vfs, path: &Path, expected_id: SegmentId) -> Result<Self, SegmentError> {
532 let file = vfs
533 .open_for_map(path)
534 .map_err(|error| SegmentError::io(path, error))?;
535 let mapping = MappedFile::open(file, path)?;
536 let artifact = path.display().to_string();
537 let parsed = parse_segment_header(
538 &artifact,
539 mapping.as_bytes(),
540 mapping.length as u64,
541 expected_id,
542 )?;
543 let rescore_valid_chunks = rescore_validation_words(&parsed.entries)?;
544 Ok(Self {
545 mapping,
546 meta: parsed.meta,
547 #[cfg(any(test, feature = "test-support"))]
548 storage_store_directory: path.parent().unwrap_or(Path::new("")).to_path_buf(),
549 header_length: parsed.header_length,
550 entries: parsed.entries,
551 document_versions_validation: OnceLock::new(),
552 stored_text_validated: OnceLock::new(),
553 vector_codes_validated: OnceLock::new(),
554 vector_factors_validated: OnceLock::new(),
555 rescore_valid_chunks: Mutex::new(vec![0_u64; rescore_valid_chunks].into_boxed_slice()),
556 query_accounting: None,
557 int8_factors_cache: OnceLock::new(),
558 postings_cache: OnceLock::new(),
559 columns_cache: OnceLock::new(),
560 alive_cache: OnceLock::new(),
561 document_version_index_cache: OnceLock::new(),
562 graph_search_cache: crate::lifecycle::graph_cache::SegmentGraphSearchCache::new(),
563 vector_ceiling_norm_range: OnceLock::new(),
564 })
565 }
566
567 pub(crate) fn open_accounted(
568 vfs: &dyn Vfs,
569 path: &Path,
570 expected: &SegmentMeta,
571 accounting: &Arc<crate::lifecycle::stats::Accounting>,
572 mut before_reader_allocation: impl FnMut(usize) -> Result<(), crate::lifecycle::StoreError>,
573 ) -> Result<Self, crate::lifecycle::StoreError> {
574 let file = vfs
575 .open_for_map(path)
576 .map_err(|error| SegmentError::io(path, error))
577 .map_err(crate::lifecycle::StoreError::Segment)?;
578 let mapping =
579 MappedFile::open(file, path).map_err(crate::lifecycle::StoreError::Segment)?;
580 let artifact = path.display().to_string();
581 let region_count =
582 preflight_region_count(&artifact, mapping.as_bytes(), mapping.length as u64)
583 .map_err(crate::lifecycle::StoreError::Segment)?;
584 let directory_bytes = region_count
585 .checked_mul(std::mem::size_of::<RegionEntry>())
586 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
587 needed: u64::MAX,
588 budget: u64::MAX,
589 component: "snapshot",
590 })?;
591 before_reader_allocation(directory_bytes)?;
592 let parsed = parse_segment_header(
593 &artifact,
594 mapping.as_bytes(),
595 mapping.length as u64,
596 expected.id,
597 )
598 .map_err(crate::lifecycle::StoreError::Segment)?;
599 if !parsed.meta.same_segment_file(expected) {
600 return Err(crate::lifecycle::StoreError::Segment(
601 SegmentError::Geometry(format!(
602 "manifest metadata {expected:?}, mapped header metadata {:?}",
603 parsed.meta
604 )),
605 ));
606 }
607 let rescore_valid_chunks = rescore_validation_words(&parsed.entries)
608 .map_err(crate::lifecycle::StoreError::Segment)?;
609 let validation_bytes = rescore_valid_chunks
610 .checked_mul(std::mem::size_of::<u64>())
611 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
612 needed: u64::MAX,
613 budget: u64::MAX,
614 component: "snapshot",
615 })?;
616 before_reader_allocation(validation_bytes)?;
617 Ok(Self {
618 mapping,
619 meta: expected.clone(),
620 #[cfg(any(test, feature = "test-support"))]
621 storage_store_directory: path.parent().unwrap_or(Path::new("")).to_path_buf(),
622 header_length: parsed.header_length,
623 entries: parsed.entries,
624 document_versions_validation: OnceLock::new(),
625 stored_text_validated: OnceLock::new(),
626 vector_codes_validated: OnceLock::new(),
627 vector_factors_validated: OnceLock::new(),
628 rescore_valid_chunks: Mutex::new(vec![0_u64; rescore_valid_chunks].into_boxed_slice()),
629 query_accounting: Some(Arc::clone(accounting)),
630 int8_factors_cache: OnceLock::new(),
631 postings_cache: OnceLock::new(),
632 columns_cache: OnceLock::new(),
633 alive_cache: OnceLock::new(),
634 document_version_index_cache: OnceLock::new(),
635 graph_search_cache: crate::lifecycle::graph_cache::SegmentGraphSearchCache::new(),
636 vector_ceiling_norm_range: OnceLock::new(),
637 })
638 }
639
640 #[must_use]
642 pub const fn meta(&self) -> &SegmentMeta {
643 &self.meta
644 }
645
646 #[must_use]
648 pub const fn header_length(&self) -> usize {
649 self.header_length
650 }
651
652 #[must_use]
654 pub(crate) const fn mapped_bytes(&self) -> usize {
655 self.mapping.length
656 }
657
658 pub(crate) fn mapped_resident_bytes(&self) -> std::io::Result<u64> {
660 crate::sys::memory::mincore_resident_bytes(self.mapping.as_bytes())
661 }
662
663 #[cfg(test)]
664 pub(crate) fn retained_validation_bytes(&self) -> usize {
665 self.rescore_valid_chunks
666 .lock()
667 .map_or(0, |words| std::mem::size_of_val(words.as_ref()))
668 }
669
670 #[cfg(any(test, feature = "test-support"))]
674 #[doc(hidden)]
675 pub fn corrupt_mapped_byte_for_test(&mut self, offset: usize, mask: u8) -> std::io::Result<()> {
676 self.mapping.corrupt_byte(offset, mask)
677 }
678
679 #[cfg(any(test, feature = "test-support"))]
680 #[doc(hidden)]
681 pub fn retained_query_view_bytes(&self) -> u64 {
682 let int8_factors = self.int8_factors_cache.get().and_then(|cached| {
683 cached
684 .value
685 .capacity()
686 .checked_mul(std::mem::size_of::<crate::scan::Int8Factors>())
687 .and_then(arc_resident_bytes::<Vec<crate::scan::Int8Factors>>)
688 });
689 let postings = self.postings_cache.get().and_then(|cached| {
690 cached
691 .value
692 .resident_bytes()
693 .ok()
694 .and_then(arc_resident_bytes::<crate::fts::sealed::SealedSegment>)
695 });
696 let columns = self.columns_cache.get().and_then(|cached| {
697 cached
698 .value
699 .resident_bytes()
700 .and_then(arc_resident_bytes::<ColumnStore>)
701 });
702 let alive = self.alive_cache.get().and_then(|cached| {
703 cached
704 .value
705 .resident_bytes()
706 .and_then(arc_resident_bytes::<AliveSet>)
707 });
708 let document_version_index = self.document_version_index_cache.get().and_then(|cached| {
709 cached
710 .value
711 .len()
712 .checked_mul(std::mem::size_of::<u32>())
713 .and_then(arc_resident_bytes::<Box<[u32]>>)
714 });
715 int8_factors
716 .into_iter()
717 .chain(postings)
718 .chain(columns)
719 .chain(alive)
720 .chain(document_version_index)
721 .fold(0_u64, |total, bytes| {
722 total.saturating_add(u64::try_from(bytes).unwrap_or(u64::MAX))
723 })
724 }
725
726 #[must_use]
728 pub fn directory(&self) -> &[RegionEntry] {
729 &self.entries
730 }
731
732 #[must_use]
734 pub fn unknown_region_ids(&self) -> Vec<u16> {
735 self.entries
736 .iter()
737 .filter_map(|entry| {
738 RegionKind::from_id(entry.kind)
739 .is_none()
740 .then_some(entry.kind)
741 })
742 .collect()
743 }
744
745 pub fn region(&self, kind: RegionKind) -> Result<&[u8], SegmentError> {
747 let entry = self.entry(kind)?;
748 let bytes = self.region_slice(entry)?;
749 let actual = region_hash(kind, bytes);
750 if actual != entry.checksum {
751 #[cfg(any(test, feature = "test-support"))]
752 crate::lifecycle::record_storage_segment_checksum_fault(
753 &self.storage_store_directory,
754 self.meta.id,
755 kind,
756 0,
757 entry.checksum,
758 actual,
759 );
760 return Err(FormatError::checksum_mismatch(
761 format!("segment:{}:{kind:?}", self.meta.id),
762 FormatCheck::BlockChecksum,
763 entry.checksum,
764 actual,
765 )
766 .into());
767 }
768 Ok(bytes)
769 }
770
771 pub(crate) fn region_by_id(&self, kind: u16) -> Result<&[u8], SegmentError> {
773 let entry = self
774 .entries
775 .iter()
776 .find(|entry| entry.kind == kind)
777 .ok_or_else(|| {
778 SegmentError::Geometry(format!("segment is missing region kind {kind}"))
779 })?;
780 let bytes = self.region_slice(entry)?;
781 let actual = xxh3_64(bytes);
782 if actual != entry.checksum {
783 return Err(FormatError::checksum_mismatch(
784 format!("segment:{}:region-{kind}", self.meta.id),
785 FormatCheck::BlockChecksum,
786 entry.checksum,
787 actual,
788 )
789 .into());
790 }
791 Ok(bytes)
792 }
793
794 pub fn region_chunk(&self, kind: RegionKind, chunk_index: u32) -> Result<&[u8], SegmentError> {
796 let entry = self.entry(kind)?;
797 let region = self.region_slice(entry)?;
798 let start = (chunk_index as usize)
799 .checked_mul(CHECKSUM_CHUNK_BYTES)
800 .ok_or_else(|| SegmentError::Geometry("chunk offset overflow".to_owned()))?;
801 let end = start.saturating_add(CHECKSUM_CHUNK_BYTES).min(region.len());
802 let chunk = region.get(start..end).ok_or_else(|| {
803 SegmentError::Geometry(format!("chunk {chunk_index} is outside {kind:?}"))
804 })?;
805 if chunk.is_empty() {
806 return Err(SegmentError::Geometry(format!(
807 "chunk {chunk_index} is outside {kind:?}"
808 )));
809 }
810 let expected = self.chunk_checksum(kind, chunk_index)?;
811 let actual = region_hash(kind, chunk);
812 if actual != expected {
813 #[cfg(any(test, feature = "test-support"))]
814 crate::lifecycle::record_storage_segment_checksum_fault(
815 &self.storage_store_directory,
816 self.meta.id,
817 kind,
818 chunk_index,
819 expected,
820 actual,
821 );
822 return Err(FormatError::checksum_mismatch(
823 format!("segment:{}:{kind:?}:chunk-{chunk_index}", self.meta.id),
824 FormatCheck::BlockChecksum,
825 expected,
826 actual,
827 )
828 .into());
829 }
830 Ok(chunk)
831 }
832
833 pub fn bit4_codes(&self) -> Result<&[u8], SegmentError> {
835 let (header, payload) = self.vector_payload(RegionKind::VectorCodes)?;
836 self.bit4_codes_from(header, payload)
837 }
838
839 fn bit4_codes_unchecked(&self) -> Result<&[u8], SegmentError> {
840 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorCodes)?;
841 self.bit4_codes_from(header, payload)
842 }
843
844 fn bit4_codes_from<'a>(
845 &'a self,
846 header: super::layout::VectorHeader,
847 payload: &'a [u8],
848 ) -> Result<&'a [u8], SegmentError> {
849 if self.meta.scheme != 4 {
850 return Err(SegmentError::Geometry(format!(
851 "Bit4 codes requested for scheme {}",
852 self.meta.scheme
853 )));
854 }
855 let expected = (self.meta.dims as usize)
856 .div_ceil(2)
857 .checked_mul(self.meta.row_count as usize)
858 .ok_or_else(|| SegmentError::Geometry("Bit4 code length overflow".to_owned()))?;
859 if header.row_stride_bytes as usize != (self.meta.dims as usize).div_ceil(2)
860 || payload.len() != expected
861 {
862 return Err(SegmentError::Geometry(format!(
863 "Bit4 code stride/length {}/{}, expected {}/{}",
864 header.row_stride_bytes,
865 payload.len(),
866 (self.meta.dims as usize).div_ceil(2),
867 expected
868 )));
869 }
870 Ok(payload)
871 }
872
873 pub fn f32_codes(&self) -> Result<&[f32], SegmentError> {
875 let (header, payload) = self.vector_payload(RegionKind::VectorCodes)?;
876 self.f32_codes_from(header, payload)
877 }
878
879 pub(crate) fn query_f32_codes(&self) -> Result<&[f32], SegmentError> {
881 self.validate_vector_region_once(RegionKind::VectorCodes, &self.vector_codes_validated)?;
882 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorCodes)?;
883 self.f32_codes_from(header, payload)
884 }
885
886 fn f32_codes_from<'a>(
887 &'a self,
888 header: super::layout::VectorHeader,
889 payload: &'a [u8],
890 ) -> Result<&'a [f32], SegmentError> {
891 if self.meta.scheme != 0 {
892 return Err(SegmentError::Geometry(format!(
893 "F32 codes requested for scheme {}",
894 self.meta.scheme
895 )));
896 }
897 let expected = (self.meta.dims as usize)
898 .checked_mul(self.meta.row_count as usize)
899 .ok_or_else(|| SegmentError::Geometry("F32 code length overflow".to_owned()))?;
900 let expected_bytes = expected
901 .checked_mul(std::mem::size_of::<f32>())
902 .ok_or_else(|| SegmentError::Geometry("F32 code byte length overflow".to_owned()))?;
903 let expected_stride = (self.meta.dims as usize)
904 .checked_mul(std::mem::size_of::<f32>())
905 .ok_or_else(|| SegmentError::Geometry("F32 code stride overflow".to_owned()))?;
906 if header.row_stride_bytes as usize != expected_stride || payload.len() != expected_bytes {
907 return Err(SegmentError::Geometry(format!(
908 "F32 code stride/length {}/{}, expected {expected_stride}/{expected_bytes}",
909 header.row_stride_bytes,
910 payload.len()
911 )));
912 }
913 cast_slice::<f32>(payload, expected, "F32 codes")
914 }
915
916 pub fn int8_codes(&self) -> Result<&[i8], SegmentError> {
918 let (header, payload) = self.vector_payload(RegionKind::VectorCodes)?;
919 self.int8_codes_from(header, payload)
920 }
921
922 fn int8_codes_unchecked(&self) -> Result<&[i8], SegmentError> {
923 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorCodes)?;
924 self.int8_codes_from(header, payload)
925 }
926
927 fn int8_codes_from<'a>(
928 &'a self,
929 header: super::layout::VectorHeader,
930 payload: &'a [u8],
931 ) -> Result<&'a [i8], SegmentError> {
932 if self.meta.scheme != 2 {
933 return Err(SegmentError::Geometry(format!(
934 "Int8 codes requested for scheme {}",
935 self.meta.scheme
936 )));
937 }
938 let expected = (self.meta.dims as usize)
939 .checked_mul(self.meta.row_count as usize)
940 .ok_or_else(|| SegmentError::Geometry("Int8 code length overflow".to_owned()))?;
941 if header.row_stride_bytes as usize != self.meta.dims as usize || payload.len() != expected
942 {
943 return Err(SegmentError::Geometry(format!(
944 "Int8 code stride/length {}/{}, expected {}/{}",
945 header.row_stride_bytes,
946 payload.len(),
947 self.meta.dims,
948 expected
949 )));
950 }
951 cast_slice::<i8>(payload, expected, "Int8 codes")
952 }
953
954 pub fn bit4_factors(&self) -> Result<&[Bit4Factors], SegmentError> {
956 let (header, payload) = self.vector_payload(RegionKind::VectorFactors)?;
957 self.bit4_factors_from(header, payload)
958 }
959
960 fn bit4_factors_unchecked(&self) -> Result<&[Bit4Factors], SegmentError> {
961 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorFactors)?;
962 self.bit4_factors_from(header, payload)
963 }
964
965 fn bit4_factors_from<'a>(
966 &'a self,
967 header: super::layout::VectorHeader,
968 payload: &'a [u8],
969 ) -> Result<&'a [Bit4Factors], SegmentError> {
970 if self.meta.scheme != 4 {
971 return Err(SegmentError::Geometry(format!(
972 "Bit4 factors requested for scheme {}",
973 self.meta.scheme
974 )));
975 }
976 if header.factor_stride_bytes != 12 {
977 return Err(SegmentError::Geometry(format!(
978 "Bit4 factor stride {}, expected 12",
979 header.factor_stride_bytes
980 )));
981 }
982 cast_slice::<Bit4Factors>(payload, self.meta.row_count as usize, "Bit4 factors")
983 }
984
985 pub fn int8_factors(&self) -> Result<&[Int8Factors], SegmentError> {
987 let (header, payload) = self.vector_payload(RegionKind::VectorFactors)?;
988 self.int8_factors_from(header, payload)
989 }
990
991 fn int8_factors_unchecked(&self) -> Result<&[Int8Factors], SegmentError> {
992 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorFactors)?;
993 self.int8_factors_from(header, payload)
994 }
995
996 fn int8_factors_from<'a>(
997 &'a self,
998 header: super::layout::VectorHeader,
999 payload: &'a [u8],
1000 ) -> Result<&'a [Int8Factors], SegmentError> {
1001 if self.meta.scheme != 2 {
1002 return Err(SegmentError::Geometry(format!(
1003 "Int8 factors requested for scheme {}",
1004 self.meta.scheme
1005 )));
1006 }
1007 if header.factor_stride_bytes != 8 {
1008 return Err(SegmentError::Geometry(format!(
1009 "Int8 factor stride {}, expected 8",
1010 header.factor_stride_bytes
1011 )));
1012 }
1013 #[cfg(any(test, feature = "test-support"))]
1014 account_region_decode(RegionKind::VectorFactors, payload.len());
1015 cast_slice::<Int8Factors>(payload, self.meta.row_count as usize, "Int8 factors")
1016 }
1017
1018 fn query_checksums_enabled() -> bool {
1028 match QUERY_CHECKSUMS.load(std::sync::atomic::Ordering::Relaxed) {
1029 0 => {
1030 let enabled =
1031 std::env::var_os("ZE_VERIFY_QUERY_CHECKSUMS").is_some_and(|value| value != "0");
1032 QUERY_CHECKSUMS.store(
1033 if enabled { 2 } else { 1 },
1034 std::sync::atomic::Ordering::Relaxed,
1035 );
1036 enabled
1037 }
1038 1 => false,
1039 _ => true,
1040 }
1041 }
1042
1043 fn validate_vector_region_once(
1053 &self,
1054 kind: RegionKind,
1055 gate: &OnceLock<()>,
1056 ) -> Result<(), SegmentError> {
1057 if gate.get().is_some() || !Self::query_checksums_enabled() {
1058 return Ok(());
1059 }
1060 let _ = self.region(kind)?;
1061 let _ = gate.set(());
1062 Ok(())
1063 }
1064
1065 pub(crate) fn query_bit4_codes(&self) -> Result<&[u8], SegmentError> {
1067 self.validate_vector_region_once(RegionKind::VectorCodes, &self.vector_codes_validated)?;
1068 self.bit4_codes_unchecked()
1069 }
1070
1071 pub(crate) fn query_int8_codes(&self) -> Result<&[i8], SegmentError> {
1073 self.validate_vector_region_once(RegionKind::VectorCodes, &self.vector_codes_validated)?;
1074 self.int8_codes_unchecked()
1075 }
1076
1077 pub(crate) fn query_int8_factors_slice(&self) -> Result<&[Int8Factors], SegmentError> {
1079 self.validate_vector_region_once(
1080 RegionKind::VectorFactors,
1081 &self.vector_factors_validated,
1082 )?;
1083 self.int8_factors_unchecked()
1084 }
1085
1086 pub(crate) fn cached_vector_ceiling_norm_range(
1092 &self,
1093 ) -> Option<crate::graph::search::GraphSegmentNormRange> {
1094 self.vector_ceiling_norm_range.get().copied()
1095 }
1096
1097 pub(crate) fn remember_vector_ceiling_norm_range(
1099 &self,
1100 range: crate::graph::search::GraphSegmentNormRange,
1101 ) {
1102 if self.vector_ceiling_norm_range.set(range).is_err() {
1103 }
1106 }
1107
1108 pub(crate) fn query_bit4_factors(&self) -> Result<&[Bit4Factors], SegmentError> {
1110 self.validate_vector_region_once(
1111 RegionKind::VectorFactors,
1112 &self.vector_factors_validated,
1113 )?;
1114 self.bit4_factors_unchecked()
1115 }
1116
1117 pub(crate) fn query_int8_factors(
1118 &self,
1119 ) -> Result<Arc<Vec<crate::scan::Int8Factors>>, crate::lifecycle::StoreError> {
1120 if let Some(cached) = self.int8_factors_cache.get() {
1121 return Ok(Arc::clone(&cached.value));
1122 }
1123 let factors = self
1124 .int8_factors()
1125 .map_err(crate::lifecycle::StoreError::Segment)?
1126 .iter()
1127 .map(|factor| crate::scan::Int8Factors::new(factor.scale, factor.offset))
1128 .collect::<Result<Vec<_>, _>>()
1129 .map_err(|_| {
1130 crate::lifecycle::StoreError::Segment(SegmentError::Geometry(
1131 "Int8 factor is not finite and non-negative".to_owned(),
1132 ))
1133 })?;
1134 let resident_bytes = factors
1135 .capacity()
1136 .checked_mul(std::mem::size_of::<crate::scan::Int8Factors>())
1137 .and_then(arc_resident_bytes::<Vec<crate::scan::Int8Factors>>)
1138 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1139 needed: u64::MAX,
1140 budget: u64::MAX,
1141 component: "snapshot",
1142 })?;
1143 let cached = CachedQueryValue {
1144 value: Arc::new(factors),
1145 _accounting: self.account_query_cache(resident_bytes)?,
1146 };
1147 if self.int8_factors_cache.set(cached).is_err() {
1148 }
1150 self.int8_factors_cache
1151 .get()
1152 .map(|cached| Arc::clone(&cached.value))
1153 .ok_or(crate::lifecycle::StoreError::Synchronization {
1154 component: "Int8 factors query cache",
1155 })
1156 }
1157
1158 pub fn rescore_f32(&self) -> Result<&[f32], SegmentError> {
1160 let (header, payload) = self.vector_payload(RegionKind::VectorRescore)?;
1161 if header.scheme != 0 {
1162 return Err(SegmentError::Geometry(format!(
1163 "v1 rescore scheme must be F32/0, got {}",
1164 header.scheme
1165 )));
1166 }
1167 let count = (self.meta.row_count as usize)
1168 .checked_mul(self.meta.dims as usize)
1169 .ok_or_else(|| SegmentError::Geometry("rescore count overflow".to_owned()))?;
1170 cast_slice::<f32>(payload, count, "f32 rescore")
1171 }
1172
1173 pub(crate) fn query_rescore_f32(&self) -> Result<&[f32], SegmentError> {
1174 self.validate_rescore_byte_range(0, VECTOR_HEADER_LEN)?;
1175 let (header, payload) = self.vector_payload_unchecked(RegionKind::VectorRescore)?;
1176 if header.scheme != 0 {
1177 return Err(SegmentError::Geometry(format!(
1178 "v1 rescore scheme must be F32/0, got {}",
1179 header.scheme
1180 )));
1181 }
1182 let count = (self.meta.row_count as usize)
1183 .checked_mul(self.meta.dims as usize)
1184 .ok_or_else(|| SegmentError::Geometry("rescore count overflow".to_owned()))?;
1185 cast_slice::<f32>(payload, count, "f32 rescore")
1186 }
1187
1188 pub(crate) fn validate_rescore_rows(&self, rows: &[u32]) -> Result<(), SegmentError> {
1189 let row_bytes = (self.meta.dims as usize)
1190 .checked_mul(std::mem::size_of::<f32>())
1191 .ok_or_else(|| SegmentError::Geometry("rescore row byte length overflow".to_owned()))?;
1192 for row in rows {
1193 if *row >= self.meta.row_count {
1194 return Err(SegmentError::Geometry(format!(
1195 "rescore row {row} is outside {} rows",
1196 self.meta.row_count
1197 )));
1198 }
1199 let start = (*row as usize)
1200 .checked_mul(row_bytes)
1201 .and_then(|offset| offset.checked_add(VECTOR_HEADER_LEN))
1202 .ok_or_else(|| SegmentError::Geometry("rescore row offset overflow".to_owned()))?;
1203 self.validate_rescore_byte_range(start, row_bytes)?;
1204 }
1205 Ok(())
1206 }
1207
1208 pub fn columns(&self) -> Result<ColumnStore, SegmentError> {
1210 let region = self.region(RegionKind::Columns)?;
1211 #[cfg(any(test, feature = "test-support"))]
1212 account_region_decode(RegionKind::Columns, region.len());
1213 let columns = decode_columns(region)?;
1214 if columns.row_count() != self.meta.row_count {
1215 return Err(SegmentError::Geometry(format!(
1216 "column rows {}, header rows {}",
1217 columns.row_count(),
1218 self.meta.row_count
1219 )));
1220 }
1221 Ok(columns)
1222 }
1223
1224 pub(crate) fn query_columns(&self) -> Result<Arc<ColumnStore>, crate::lifecycle::StoreError> {
1225 if let Some(cached) = self.columns_cache.get() {
1226 return Ok(Arc::clone(&cached.value));
1227 }
1228 let region = self
1229 .region(RegionKind::Columns)
1230 .map_err(crate::lifecycle::StoreError::Segment)?;
1231 #[cfg(any(test, feature = "test-support"))]
1232 account_region_decode(RegionKind::Columns, region.len());
1233 let mut columns = decode_columns(region).map_err(crate::lifecycle::StoreError::Segment)?;
1234 if columns.row_count() != self.meta.row_count {
1235 return Err(crate::lifecycle::StoreError::Segment(
1236 SegmentError::Geometry(format!(
1237 "column rows {}, header rows {}",
1238 columns.row_count(),
1239 self.meta.row_count
1240 )),
1241 ));
1242 }
1243 columns.compact_for_cache();
1244 let resident_bytes = columns
1245 .resident_bytes()
1246 .and_then(arc_resident_bytes::<ColumnStore>)
1247 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1248 needed: u64::MAX,
1249 budget: u64::MAX,
1250 component: "snapshot",
1251 })?;
1252 let cached = CachedQueryValue {
1253 value: Arc::new(columns),
1254 _accounting: self.account_query_cache(resident_bytes)?,
1255 };
1256 if self.columns_cache.set(cached).is_err() {
1257 }
1259 self.columns_cache
1260 .get()
1261 .map(|cached| Arc::clone(&cached.value))
1262 .ok_or(crate::lifecycle::StoreError::Synchronization {
1263 component: "columns query cache",
1264 })
1265 }
1266
1267 pub fn postings(&self) -> Result<Option<crate::fts::sealed::SealedSegment>, SegmentError> {
1269 if !self
1270 .entries
1271 .iter()
1272 .any(|entry| entry.kind == RegionKind::Postings.id())
1273 {
1274 return Ok(None);
1275 }
1276 let region = self.region(RegionKind::Postings)?;
1277 #[cfg(any(test, feature = "test-support"))]
1278 account_region_decode(RegionKind::Postings, region.len());
1279 let postings = crate::fts::sealed::SealedSegment::decode_region(region)?;
1280 if postings.row_count() != self.meta.row_count {
1281 return Err(SegmentError::Geometry(format!(
1282 "postings rows {}, header rows {}",
1283 postings.row_count(),
1284 self.meta.row_count
1285 )));
1286 }
1287 Ok(Some(postings))
1288 }
1289
1290 pub(crate) fn lexical_postings_identity(&self) -> Result<(u64, u64), SegmentError> {
1293 let entry = self.entry(RegionKind::Postings)?;
1294 Ok((entry.length, entry.checksum))
1295 }
1296
1297 pub(crate) fn query_postings(
1298 &self,
1299 ) -> Result<Option<Arc<crate::fts::sealed::SealedSegment>>, crate::lifecycle::StoreError> {
1300 if !self
1301 .entries
1302 .iter()
1303 .any(|entry| entry.kind == RegionKind::Postings.id())
1304 {
1305 return Ok(None);
1306 }
1307 if let Some(cached) = self.postings_cache.get() {
1308 return Ok(Some(Arc::clone(&cached.value)));
1309 }
1310 let region = self
1311 .region(RegionKind::Postings)
1312 .map_err(crate::lifecycle::StoreError::Segment)?;
1313 #[cfg(any(test, feature = "test-support"))]
1314 account_region_decode(RegionKind::Postings, region.len());
1315 let postings = crate::fts::sealed::SealedSegment::decode_region(region)
1316 .map_err(SegmentError::from)
1317 .map_err(crate::lifecycle::StoreError::Segment)?;
1318 if postings.row_count() != self.meta.row_count {
1319 return Err(crate::lifecycle::StoreError::Segment(
1320 SegmentError::Geometry(format!(
1321 "postings rows {}, header rows {}",
1322 postings.row_count(),
1323 self.meta.row_count
1324 )),
1325 ));
1326 }
1327 let resident_bytes = postings
1328 .resident_bytes()
1329 .map_err(SegmentError::from)
1330 .map_err(crate::lifecycle::StoreError::Segment)?
1331 .checked_add(std::mem::size_of::<crate::fts::sealed::SealedSegment>())
1332 .and_then(|bytes| bytes.checked_add(2 * std::mem::size_of::<usize>()))
1333 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1334 needed: u64::MAX,
1335 budget: u64::MAX,
1336 component: "snapshot",
1337 })?;
1338 let mut cache_accounting = self
1339 .query_accounting
1340 .as_ref()
1341 .map(|accounting| {
1342 crate::lifecycle::stats::AccountedCounter::new(
1343 accounting,
1344 crate::lifecycle::stats::AllocationComponent::Snapshot,
1345 )
1346 })
1347 .transpose()?;
1348 if let Some(counter) = cache_accounting.as_mut() {
1349 counter.set(resident_bytes)?;
1350 }
1351 let cached = CachedQueryValue {
1352 value: Arc::new(postings),
1353 _accounting: cache_accounting,
1354 };
1355 if self.postings_cache.set(cached).is_err() {
1356 }
1358 self.postings_cache
1359 .get()
1360 .map(|cached| Some(Arc::clone(&cached.value)))
1361 .ok_or(crate::lifecycle::StoreError::Synchronization {
1362 component: "postings query cache",
1363 })
1364 }
1365
1366 pub fn alive(&self) -> Result<AliveSet, SegmentError> {
1368 let region = self.region(RegionKind::Alive)?;
1369 #[cfg(any(test, feature = "test-support"))]
1370 account_region_decode(RegionKind::Alive, region.len());
1371 let alive = decode_alive(region)?;
1372 if alive.row_count() != self.meta.row_count {
1373 return Err(SegmentError::Geometry(format!(
1374 "alive rows {}, header rows {}",
1375 alive.row_count(),
1376 self.meta.row_count
1377 )));
1378 }
1379 Ok(alive)
1380 }
1381
1382 pub(crate) fn query_alive(&self) -> Result<Arc<AliveSet>, crate::lifecycle::StoreError> {
1383 if let Some(cached) = self.alive_cache.get() {
1384 return Ok(Arc::clone(&cached.value));
1385 }
1386 let region = self
1387 .region(RegionKind::Alive)
1388 .map_err(crate::lifecycle::StoreError::Segment)?;
1389 #[cfg(any(test, feature = "test-support"))]
1390 account_region_decode(RegionKind::Alive, region.len());
1391 let mut alive = decode_alive(region).map_err(crate::lifecycle::StoreError::Segment)?;
1392 if alive.row_count() != self.meta.row_count {
1393 return Err(crate::lifecycle::StoreError::Segment(
1394 SegmentError::Geometry(format!(
1395 "alive rows {}, header rows {}",
1396 alive.row_count(),
1397 self.meta.row_count
1398 )),
1399 ));
1400 }
1401 alive.compact_for_cache();
1402 let resident_bytes = alive
1403 .resident_bytes()
1404 .and_then(arc_resident_bytes::<AliveSet>)
1405 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1406 needed: u64::MAX,
1407 budget: u64::MAX,
1408 component: "snapshot",
1409 })?;
1410 let cached = CachedQueryValue {
1411 value: Arc::new(alive),
1412 _accounting: self.account_query_cache(resident_bytes)?,
1413 };
1414 if self.alive_cache.set(cached).is_err() {
1415 }
1417 self.alive_cache
1418 .get()
1419 .map(|cached| Arc::clone(&cached.value))
1420 .ok_or(crate::lifecycle::StoreError::Synchronization {
1421 component: "alive query cache",
1422 })
1423 }
1424
1425 fn document_versions_region(&self) -> Result<Option<&[u8]>, SegmentError> {
1428 let Some(entry) = self
1429 .entries
1430 .iter()
1431 .find(|entry| entry.kind == RegionKind::DocumentVersions.id())
1432 else {
1433 return Ok(None);
1434 };
1435 let validation = self.document_versions_validation.get_or_init(|| {
1436 let bytes = self
1437 .region_slice(entry)
1438 .map_err(|error| DocumentVersionValidationError::Geometry(error.to_string()))?;
1439 if region_hash(RegionKind::DocumentVersions, bytes) != entry.checksum {
1440 return Err(DocumentVersionValidationError::Checksum);
1441 }
1442 let expected = (self.meta.row_count as usize)
1443 .checked_mul(24)
1444 .ok_or_else(|| {
1445 DocumentVersionValidationError::Geometry(
1446 "document-version length overflow".to_owned(),
1447 )
1448 })?;
1449 if bytes.len() != expected {
1450 return Err(DocumentVersionValidationError::Geometry(format!(
1451 "document-version bytes {}, expected {expected}",
1452 bytes.len()
1453 )));
1454 }
1455 Ok(())
1456 });
1457 if let Err(error) = validation {
1458 return Err(error.clone().into_segment_error(self.meta.id));
1459 }
1460 self.region_slice_unaccounted(entry).map(Some)
1461 }
1462
1463 pub fn document_version(&self, row: usize) -> Result<Option<DocumentVersion>, SegmentError> {
1465 let Some(bytes) = self.document_versions_region()? else {
1466 return Ok(None);
1467 };
1468 let start = row
1469 .checked_mul(24)
1470 .ok_or_else(|| SegmentError::Geometry("document-version row overflow".to_owned()))?;
1471 let doc_end = start
1472 .checked_add(16)
1473 .ok_or_else(|| SegmentError::Geometry("document id end overflow".to_owned()))?;
1474 let revision_end = doc_end
1475 .checked_add(8)
1476 .ok_or_else(|| SegmentError::Geometry("revision end overflow".to_owned()))?;
1477 let doc_id = bytes
1478 .get(start..doc_end)
1479 .and_then(|value| value.try_into().ok())
1480 .map(u128::from_le_bytes)
1481 .ok_or_else(|| SegmentError::Geometry(format!("document id row {row} is missing")))?;
1482 let revision = bytes
1483 .get(doc_end..revision_end)
1484 .and_then(|value| value.try_into().ok())
1485 .map(u64::from_le_bytes)
1486 .ok_or_else(|| SegmentError::Geometry(format!("revision row {row} is missing")))?;
1487 Ok(Some(DocumentVersion::new(
1488 DocId::new(doc_id),
1489 Revision::new(revision),
1490 )))
1491 }
1492
1493 pub fn row_for_document_version(
1498 &self,
1499 version: DocumentVersion,
1500 ) -> Result<Option<usize>, SegmentError> {
1501 let Some(bytes) = self.document_versions_region()? else {
1502 return Ok(None);
1503 };
1504 let needle = encode_document_version(version);
1505 Ok(bytes
1506 .chunks_exact(DOCUMENT_VERSION_BYTES)
1507 .position(|entry| entry == needle))
1508 }
1509
1510 pub(crate) fn query_row_for_document_version(
1515 &self,
1516 version: DocumentVersion,
1517 ) -> Result<Option<usize>, crate::lifecycle::StoreError> {
1518 let Some(bytes) = self
1519 .document_versions_region()
1520 .map_err(crate::lifecycle::StoreError::Segment)?
1521 else {
1522 return Ok(None);
1523 };
1524 let Some(index) = self.query_document_version_index(bytes)? else {
1525 return Ok(None);
1526 };
1527 let needle = encode_document_version(version);
1528 let position =
1529 index.partition_point(|&row| document_version_entry(bytes, row) < needle.as_slice());
1530 Ok(index
1531 .get(position)
1532 .copied()
1533 .filter(|&row| document_version_entry(bytes, row) == needle)
1534 .map(|row| row as usize))
1535 }
1536
1537 fn query_document_version_index(
1538 &self,
1539 bytes: &[u8],
1540 ) -> Result<Option<Arc<Box<[u32]>>>, crate::lifecycle::StoreError> {
1541 if let Some(cached) = self.document_version_index_cache.get() {
1542 return Ok(Some(Arc::clone(&cached.value)));
1543 }
1544 let mut rows: Vec<u32> = (0..self.meta.row_count).collect();
1545 rows.sort_unstable_by(|&left, &right| {
1546 document_version_entry(bytes, left)
1547 .cmp(document_version_entry(bytes, right))
1548 .then(left.cmp(&right))
1549 });
1550 let index = rows.into_boxed_slice();
1551 let resident_bytes = index
1552 .len()
1553 .checked_mul(std::mem::size_of::<u32>())
1554 .and_then(arc_resident_bytes::<Box<[u32]>>)
1555 .ok_or(crate::lifecycle::StoreError::BudgetExceeded {
1556 needed: u64::MAX,
1557 budget: u64::MAX,
1558 component: "snapshot",
1559 })?;
1560 let cached = CachedQueryValue {
1561 value: Arc::new(index),
1562 _accounting: self.account_query_cache(resident_bytes)?,
1563 };
1564 if self.document_version_index_cache.set(cached).is_err() {
1565 }
1567 self.document_version_index_cache
1568 .get()
1569 .map(|cached| Some(Arc::clone(&cached.value)))
1570 .ok_or(crate::lifecycle::StoreError::Synchronization {
1571 component: "document-version index cache",
1572 })
1573 }
1574
1575 pub fn stored_metadata(&self) -> Result<Option<StoredMetadataRows<'_>>, SegmentError> {
1577 let Some(entry) = self
1578 .entries
1579 .iter()
1580 .find(|entry| entry.kind == RegionKind::StoredMetadata.id())
1581 else {
1582 return Ok(None);
1583 };
1584 let region = self.region(RegionKind::StoredMetadata)?;
1585 let declared_rows = region
1586 .get(..4)
1587 .and_then(|value| value.try_into().ok())
1588 .map(u32::from_le_bytes)
1589 .ok_or_else(|| {
1590 SegmentError::Geometry("stored-metadata header is truncated".to_owned())
1591 })?;
1592 let reserved = region
1593 .get(4..8)
1594 .and_then(|value| value.try_into().ok())
1595 .map(u32::from_le_bytes)
1596 .ok_or_else(|| {
1597 SegmentError::Geometry("stored-metadata reserved field is truncated".to_owned())
1598 })?;
1599 if declared_rows != self.meta.row_count || reserved != 0 {
1600 return Err(SegmentError::Geometry(format!(
1601 "stored-metadata rows/reserved {declared_rows}/{reserved}, expected {}/0",
1602 self.meta.row_count
1603 )));
1604 }
1605 let row_count = declared_rows as usize;
1606 let offset_bytes = row_count
1607 .checked_add(1)
1608 .and_then(|count| count.checked_mul(8))
1609 .ok_or_else(|| SegmentError::Geometry("stored-metadata offsets overflow".to_owned()))?;
1610 let offsets_end = 8_usize.checked_add(offset_bytes).ok_or_else(|| {
1611 SegmentError::Geometry("stored-metadata offset end overflow".to_owned())
1612 })?;
1613 let offsets = region.get(8..offsets_end).ok_or_else(|| {
1614 SegmentError::Geometry("stored-metadata offsets are truncated".to_owned())
1615 })?;
1616 let bytes = region.get(offsets_end..).ok_or_else(|| {
1617 SegmentError::Geometry("stored-metadata payload is truncated".to_owned())
1618 })?;
1619 let view = StoredMetadataRows {
1620 offsets,
1621 bytes,
1622 row_count,
1623 };
1624 let first = view
1625 .offsets
1626 .get(..8)
1627 .and_then(|value| value.try_into().ok())
1628 .map(u64::from_le_bytes)
1629 .ok_or_else(|| {
1630 SegmentError::Geometry("stored-metadata first offset is missing".to_owned())
1631 })?;
1632 if first != 0 {
1633 return Err(SegmentError::Geometry(format!(
1634 "stored-metadata first offset is {first}, expected zero"
1635 )));
1636 }
1637 let mut previous = 0_usize;
1638 for row in 0..row_count {
1639 let value = view.row(row).ok_or_else(|| {
1640 SegmentError::Geometry(format!("stored-metadata row {row} is invalid"))
1641 })?;
1642 previous = previous.checked_add(value.len()).ok_or_else(|| {
1643 SegmentError::Geometry("stored-metadata length overflow".to_owned())
1644 })?;
1645 }
1646 if previous != bytes.len() {
1647 return Err(SegmentError::Geometry(format!(
1648 "stored-metadata final offset {previous}, bytes {}",
1649 bytes.len()
1650 )));
1651 }
1652 let _ = entry;
1653 Ok(Some(view))
1654 }
1655
1656 pub fn stored_text(&self) -> Result<Option<StoredTextRows<'_>>, SegmentError> {
1660 let Some(_) = self
1661 .entries
1662 .iter()
1663 .find(|entry| entry.kind == RegionKind::StoredText.id())
1664 else {
1665 return Ok(None);
1666 };
1667 let region = self.region(RegionKind::StoredText)?;
1668 let view = stored_text_view(region, self.meta.row_count)?;
1669 let StoredTextRows {
1670 offsets,
1671 bytes,
1672 row_count,
1673 ..
1674 } = view;
1675 let mut previous = 0_usize;
1676 for row in 0..row_count {
1677 let value = view.row(row).ok_or_else(|| {
1678 SegmentError::Geometry(format!("stored-text row {row} is invalid"))
1679 })?;
1680 let end_index = row.checked_mul(8).ok_or_else(|| {
1681 SegmentError::Geometry("stored-text offset index overflow".to_owned())
1682 })?;
1683 let end = offsets
1684 .get(end_index..end_index + 8)
1685 .and_then(|value| value.try_into().ok())
1686 .map(u64::from_le_bytes)
1687 .and_then(|value| usize::try_from(value).ok())
1688 .ok_or_else(|| {
1689 SegmentError::Geometry(format!("stored-text offset row {row} is invalid"))
1690 })?;
1691 if end < previous || end > bytes.len() {
1692 return Err(SegmentError::Geometry(format!(
1693 "stored-text offset row {row} is outside {previous}..={} bytes",
1694 bytes.len()
1695 )));
1696 }
1697 if value.is_none() && end != previous {
1698 return Err(SegmentError::Geometry(format!(
1699 "stored-text absent row {row} owns bytes {previous}..{end}"
1700 )));
1701 }
1702 previous = end;
1703 }
1704 if previous != bytes.len() {
1705 return Err(SegmentError::Geometry(format!(
1706 "stored-text final offset {previous}, bytes {}",
1707 bytes.len()
1708 )));
1709 }
1710 Ok(Some(view))
1711 }
1712
1713 pub(crate) fn query_stored_text(&self) -> Result<Option<StoredTextRows<'_>>, SegmentError> {
1719 if self.stored_text_validated.get().is_some() || !Self::query_checksums_enabled() {
1720 let Some(entry) = self
1721 .entries
1722 .iter()
1723 .find(|entry| entry.kind == RegionKind::StoredText.id())
1724 else {
1725 return Ok(None);
1726 };
1727 let region = self.region_slice(entry)?;
1728 return stored_text_view(region, self.meta.row_count).map(Some);
1729 }
1730 let view = self.stored_text()?;
1731 if view.is_some() {
1732 let _ = self.stored_text_validated.set(());
1734 }
1735 Ok(view)
1736 }
1737}
1738
1739fn stored_text_view(region: &[u8], header_rows: u32) -> Result<StoredTextRows<'_>, SegmentError> {
1741 {
1742 let declared_rows = region
1743 .get(..4)
1744 .and_then(|value| value.try_into().ok())
1745 .map(u32::from_le_bytes)
1746 .ok_or_else(|| SegmentError::Geometry("stored-text header is truncated".to_owned()))?;
1747 let reserved = region
1748 .get(4..8)
1749 .and_then(|value| value.try_into().ok())
1750 .map(u32::from_le_bytes)
1751 .ok_or_else(|| {
1752 SegmentError::Geometry("stored-text reserved field is truncated".to_owned())
1753 })?;
1754 if declared_rows != header_rows || reserved != 0 {
1755 return Err(SegmentError::Geometry(format!(
1756 "stored-text rows/reserved {declared_rows}/{reserved}, expected {header_rows}/0"
1757 )));
1758 }
1759 let row_count = declared_rows as usize;
1760 let bitmap_len = row_count.div_ceil(8);
1761 let bitmap_end = 8_usize
1762 .checked_add(bitmap_len)
1763 .ok_or_else(|| SegmentError::Geometry("stored-text bitmap overflow".to_owned()))?;
1764 let offsets_len = row_count
1765 .checked_mul(8)
1766 .ok_or_else(|| SegmentError::Geometry("stored-text offsets overflow".to_owned()))?;
1767 let offsets_end = bitmap_end
1768 .checked_add(offsets_len)
1769 .ok_or_else(|| SegmentError::Geometry("stored-text offset end overflow".to_owned()))?;
1770 let present = region.get(8..bitmap_end).ok_or_else(|| {
1771 SegmentError::Geometry("stored-text presence bitmap is truncated".to_owned())
1772 })?;
1773 let offsets = region.get(bitmap_end..offsets_end).ok_or_else(|| {
1774 SegmentError::Geometry("stored-text offsets are truncated".to_owned())
1775 })?;
1776 let bytes = region
1777 .get(offsets_end..)
1778 .ok_or_else(|| SegmentError::Geometry("stored-text payload is truncated".to_owned()))?;
1779 if !row_count.is_multiple_of(8) {
1780 let used = row_count % 8;
1781 let padding_mask = !((1_u8 << used) - 1);
1782 if present.last().is_some_and(|byte| byte & padding_mask != 0) {
1783 return Err(SegmentError::Geometry(
1784 "stored-text presence padding bits are nonzero".to_owned(),
1785 ));
1786 }
1787 }
1788 Ok(StoredTextRows {
1789 present,
1790 offsets,
1791 bytes,
1792 row_count,
1793 })
1794 }
1795}
1796
1797impl SegmentReader {
1798 pub fn graph_node_blocks(&self) -> Result<GraphNodeBlocks<'_>, SegmentError> {
1800 let blocks = decode_node_blocks(self.region(RegionKind::GraphNodeBlocks)?)?;
1801 if blocks.layout().dims() != self.meta.dims {
1802 return Err(SegmentError::Geometry(format!(
1803 "graph dimensions {}, segment dimensions {}",
1804 blocks.layout().dims(),
1805 self.meta.dims
1806 )));
1807 }
1808 if blocks.node_count() != self.meta.row_count {
1809 return Err(SegmentError::Geometry(format!(
1810 "graph nodes {}, segment rows {}",
1811 blocks.node_count(),
1812 self.meta.row_count
1813 )));
1814 }
1815 Ok(blocks)
1816 }
1817
1818 pub(crate) fn bind_validated_graph_node_blocks(
1819 &self,
1820 descriptor: ValidatedGraphNodeBlocks,
1821 ) -> Result<GraphNodeBlocks<'_>, SegmentError> {
1822 let entry = self.entry(RegionKind::GraphNodeBlocks)?;
1823 let bytes = self.region_slice(entry)?;
1824 descriptor.bind(bytes).map_err(SegmentError::Graph)
1825 }
1826
1827 pub fn validate_all(&self) -> Result<(), SegmentError> {
1829 for entry in &self.entries {
1830 let bytes = self.region_slice(entry)?;
1831 let actual = xxh3_64(bytes);
1832 if actual != entry.checksum {
1833 return Err(FormatError::new(
1834 format!("segment:{}:region-{}", self.meta.id, entry.kind),
1835 FormatCheck::BlockChecksum,
1836 format!("expected {:#018x}, computed {actual:#018x}", entry.checksum),
1837 )
1838 .into());
1839 }
1840 }
1841 let bytes = self.mapping.as_bytes();
1842 let trailer_start = bytes.len().checked_sub(FILE_TRAILER_LEN).ok_or_else(|| {
1843 FormatError::new(
1844 format!("segment: {}", self.meta.id),
1845 FormatCheck::Length,
1846 "missing file trailer",
1847 )
1848 })?;
1849 let expected = read_u64("segment", bytes, trailer_start)?;
1850 let checksummed = bytes.get(..trailer_start).ok_or_else(|| {
1851 FormatError::new("segment", FormatCheck::Length, "invalid trailer position")
1852 })?;
1853 let actual = xxh3_64(checksummed);
1854 if actual != expected {
1855 return Err(FormatError::new(
1856 format!("segment: {}", self.meta.id),
1857 FormatCheck::FileChecksum,
1858 format!("expected {expected:#018x}, computed {actual:#018x}"),
1859 )
1860 .into());
1861 }
1862 Ok(())
1863 }
1864
1865 fn vector_payload(
1866 &self,
1867 kind: RegionKind,
1868 ) -> Result<(super::layout::VectorHeader, &[u8]), SegmentError> {
1869 let region = self.region(kind)?;
1870 let header_bytes = region.get(..VECTOR_HEADER_LEN).ok_or_else(|| {
1871 SegmentError::Geometry(format!("{kind:?} is shorter than vector header"))
1872 })?;
1873 let header = decode_vector_header(header_bytes)?;
1874 if header.dims != self.meta.dims
1875 || header.row_count != self.meta.row_count
1876 || (kind != RegionKind::VectorRescore && header.scheme != self.meta.scheme)
1877 {
1878 return Err(SegmentError::Geometry(format!(
1879 "{kind:?} header scheme/dims/rows {}/{}/{}, segment {}/{}/{}",
1880 header.scheme,
1881 header.dims,
1882 header.row_count,
1883 self.meta.scheme,
1884 self.meta.dims,
1885 self.meta.row_count
1886 )));
1887 }
1888 let payload = region
1889 .get(VECTOR_HEADER_LEN..)
1890 .ok_or_else(|| SegmentError::Geometry(format!("{kind:?} payload offset is invalid")))?;
1891 Ok((header, payload))
1892 }
1893
1894 fn vector_payload_unchecked(
1895 &self,
1896 kind: RegionKind,
1897 ) -> Result<(super::layout::VectorHeader, &[u8]), SegmentError> {
1898 let entry = self.entry(kind)?;
1899 let region = self.region_slice_unaccounted(entry)?;
1900 let header_bytes = region.get(..VECTOR_HEADER_LEN).ok_or_else(|| {
1901 SegmentError::Geometry(format!("{kind:?} is shorter than vector header"))
1902 })?;
1903 let header = decode_vector_header(header_bytes)?;
1904 if header.dims != self.meta.dims
1905 || header.row_count != self.meta.row_count
1906 || (kind != RegionKind::VectorRescore && header.scheme != self.meta.scheme)
1907 {
1908 return Err(SegmentError::Geometry(format!(
1909 "{kind:?} header scheme/dims/rows {}/{}/{}, segment {}/{}/{}",
1910 header.scheme,
1911 header.dims,
1912 header.row_count,
1913 self.meta.scheme,
1914 self.meta.dims,
1915 self.meta.row_count
1916 )));
1917 }
1918 let payload = region
1919 .get(VECTOR_HEADER_LEN..)
1920 .ok_or_else(|| SegmentError::Geometry(format!("{kind:?} payload offset is invalid")))?;
1921 Ok((header, payload))
1922 }
1923
1924 fn validate_rescore_byte_range(&self, start: usize, length: usize) -> Result<(), SegmentError> {
1925 let entry = self.entry(RegionKind::VectorRescore)?;
1926 let region_length = usize::try_from(entry.length).map_err(|_| {
1927 SegmentError::Geometry("rescore region length exceeds usize".to_owned())
1928 })?;
1929 let end = start
1930 .checked_add(length)
1931 .ok_or_else(|| SegmentError::Geometry("rescore byte range overflow".to_owned()))?;
1932 if end > region_length {
1933 return Err(SegmentError::Geometry(format!(
1934 "rescore byte range {start}..{end} exceeds {region_length}"
1935 )));
1936 }
1937 if length == 0 || !Self::query_checksums_enabled() {
1938 return Ok(());
1939 }
1940 let first_chunk = start / CHECKSUM_CHUNK_BYTES;
1941 let last_chunk = end.saturating_sub(1) / CHECKSUM_CHUNK_BYTES;
1942 let mut validated = self.rescore_valid_chunks.lock().map_err(|_| {
1943 SegmentError::Geometry("rescore validation state is poisoned".to_owned())
1944 })?;
1945 for chunk in first_chunk..=last_chunk {
1946 let word = chunk / u64::BITS as usize;
1947 let bit = chunk % u64::BITS as usize;
1948 let mask = 1_u64 << bit;
1949 let state = validated.get_mut(word).ok_or_else(|| {
1950 SegmentError::Geometry(format!("rescore checksum chunk {chunk} is untracked"))
1951 })?;
1952 if *state & mask != 0 {
1953 continue;
1954 }
1955 let chunk_index = u32::try_from(chunk).map_err(|_| {
1956 SegmentError::Geometry("rescore checksum chunk exceeds u32".to_owned())
1957 })?;
1958 self.region_chunk(RegionKind::VectorRescore, chunk_index)?;
1959 *state |= mask;
1960 }
1961 Ok(())
1962 }
1963
1964 fn account_query_cache(
1965 &self,
1966 bytes: usize,
1967 ) -> Result<Option<crate::lifecycle::stats::AccountedCounter>, crate::lifecycle::StoreError>
1968 {
1969 let Some(accounting) = &self.query_accounting else {
1970 return Ok(None);
1971 };
1972 let mut counter = crate::lifecycle::stats::AccountedCounter::new(
1973 accounting,
1974 crate::lifecycle::stats::AllocationComponent::Snapshot,
1975 )?;
1976 counter.set(bytes)?;
1977 Ok(Some(counter))
1978 }
1979
1980 fn entry(&self, kind: RegionKind) -> Result<&RegionEntry, SegmentError> {
1981 self.entries
1982 .iter()
1983 .find(|entry| entry.kind == kind.id())
1984 .ok_or(SegmentError::MissingRegion(kind))
1985 }
1986
1987 fn region_slice(&self, entry: &RegionEntry) -> Result<&[u8], SegmentError> {
1988 let bytes = self.region_slice_unaccounted(entry)?;
1989 account_data_read(bytes.len());
1990 Ok(bytes)
1991 }
1992
1993 fn region_slice_unaccounted(&self, entry: &RegionEntry) -> Result<&[u8], SegmentError> {
1994 let start = usize::try_from(entry.offset)
1995 .map_err(|_| SegmentError::Geometry("region offset exceeds usize".to_owned()))?;
1996 let length = usize::try_from(entry.length)
1997 .map_err(|_| SegmentError::Geometry("region length exceeds usize".to_owned()))?;
1998 let end = start
1999 .checked_add(length)
2000 .ok_or_else(|| SegmentError::Geometry("region end overflow".to_owned()))?;
2001 let bytes = self.mapping.as_bytes().get(start..end).ok_or_else(|| {
2002 SegmentError::Geometry(format!(
2003 "region {} range {start}..{end} exceeds file {}",
2004 entry.kind, self.mapping.length
2005 ))
2006 })?;
2007 Ok(bytes)
2008 }
2009
2010 fn chunk_checksum(&self, kind: RegionKind, chunk_index: u32) -> Result<u64, SegmentError> {
2011 let table = self.region(RegionKind::ChecksumTable)?;
2012 let mut cursor = Cursor::new("checksum-table", table);
2013 let count = cursor.u32()?;
2014 let chunk_size = cursor.u32()?;
2015 if chunk_size as usize != CHECKSUM_CHUNK_BYTES {
2016 return Err(SegmentError::Geometry(format!(
2017 "checksum chunk size {chunk_size}, expected {CHECKSUM_CHUNK_BYTES}"
2018 )));
2019 }
2020 let mut found = None;
2021 for _ in 0..count {
2022 let entry_kind = cursor.u16()?;
2023 let reserved = cursor.u16()?;
2024 let entry_chunk = cursor.u32()?;
2025 let checksum = cursor.u64()?;
2026 if reserved != 0 {
2027 return Err(SegmentError::Geometry(
2028 "checksum-table reserved field is non-zero".to_owned(),
2029 ));
2030 }
2031 if entry_kind == kind.id() && entry_chunk == chunk_index {
2032 found = Some(checksum);
2033 }
2034 }
2035 cursor.finish().map_err(SegmentError::Geometry)?;
2036 found.ok_or_else(|| {
2037 SegmentError::Geometry(format!(
2038 "checksum table has no {kind:?} chunk {chunk_index}"
2039 ))
2040 })
2041 }
2042}
2043
2044const DOCUMENT_VERSION_BYTES: usize = 24;
2047
2048fn encode_document_version(version: DocumentVersion) -> [u8; DOCUMENT_VERSION_BYTES] {
2049 let doc_id = version.doc_id().get().to_le_bytes();
2050 let revision = version.revision().get().to_le_bytes();
2051 let mut bytes = [0_u8; DOCUMENT_VERSION_BYTES];
2052 for (slot, byte) in bytes.iter_mut().zip(doc_id.into_iter().chain(revision)) {
2053 *slot = byte;
2054 }
2055 bytes
2056}
2057
2058fn document_version_entry(bytes: &[u8], row: u32) -> &[u8] {
2062 (row as usize)
2063 .checked_mul(DOCUMENT_VERSION_BYTES)
2064 .and_then(|start| bytes.get(start..start.checked_add(DOCUMENT_VERSION_BYTES)?))
2065 .unwrap_or(&[])
2066}
2067
2068fn arc_resident_bytes<T>(deep_bytes: usize) -> Option<usize> {
2069 deep_bytes
2070 .checked_add(std::mem::size_of::<T>())?
2071 .checked_add(2 * std::mem::size_of::<usize>())
2072}
2073
2074fn rescore_validation_words(entries: &[RegionEntry]) -> Result<usize, SegmentError> {
2075 let Some(entry) = entries
2076 .iter()
2077 .find(|entry| entry.kind == RegionKind::VectorRescore.id())
2078 else {
2079 return Ok(0);
2080 };
2081 let length = usize::try_from(entry.length)
2082 .map_err(|_| SegmentError::Geometry("rescore region length exceeds usize".to_owned()))?;
2083 let chunks = length.div_ceil(CHECKSUM_CHUNK_BYTES);
2084 Ok(chunks.div_ceil(u64::BITS as usize))
2085}
2086
2087pub fn validate_header_with_vfs(
2089 vfs: &dyn Vfs,
2090 path: &Path,
2091 expected: &SegmentMeta,
2092) -> Result<usize, SegmentError> {
2093 let actual_length = vfs
2094 .open(path)
2095 .map_err(|error| SegmentError::io(path, error))?;
2096 let mut header_bytes = vfs
2097 .read_range(path, 0, FILE_HEADER_LEN)
2098 .map_err(|error| SegmentError::io(path, error))?;
2099 let artifact = path.display().to_string();
2100 #[cfg(any(test, feature = "test-support"))]
2101 let actual_family = header_bytes
2102 .get(8..10)
2103 .and_then(|family| <[u8; 2]>::try_from(family).ok())
2104 .map(u16::from_le_bytes);
2105 let fixed =
2106 decode_header(&artifact, FormatFamily::Segment, &header_bytes).map_err(|error| {
2107 #[cfg(any(test, feature = "test-support"))]
2108 crate::lifecycle::record_storage_segment_format_fault(
2109 &error,
2110 actual_family,
2111 expected.id,
2112 );
2113 SegmentError::from(error)
2114 })?;
2115 let header_length = usize::try_from(fixed.header_length)
2116 .map_err(|_| SegmentError::Geometry("segment header length exceeds usize".to_owned()))?;
2117 validate_bounded_header_length(&artifact, header_length, actual_length)?;
2118 let remaining = header_length.saturating_sub(FILE_HEADER_LEN);
2119 if remaining > 0 {
2120 let tail = vfs
2121 .read_range(path, FILE_HEADER_LEN as u64, remaining)
2122 .map_err(|error| SegmentError::io(path, error))?;
2123 header_bytes.extend_from_slice(&tail);
2124 }
2125 let parsed = parse_segment_header(&artifact, &header_bytes, actual_length, expected.id)?;
2126 if !parsed.meta.same_segment_file(expected) {
2127 return Err(SegmentError::Geometry(format!(
2128 "manifest metadata {expected:?}, header metadata {:?}",
2129 parsed.meta
2130 )));
2131 }
2132 Ok(header_length)
2133}
2134
2135pub fn validate_segment_bytes(bytes: &[u8]) -> Result<SegmentMeta, SegmentError> {
2137 let id_bytes: [u8; 16] = bytes
2138 .get(FILE_HEADER_LEN..FILE_HEADER_LEN.saturating_add(16))
2139 .ok_or_else(|| {
2140 FormatError::new(
2141 "segment-bytes",
2142 FormatCheck::Length,
2143 format!("missing segment id in {} bytes", bytes.len()),
2144 )
2145 })?
2146 .try_into()
2147 .map_err(|_| {
2148 FormatError::new("segment-bytes", FormatCheck::Length, "invalid segment id")
2149 })?;
2150 let id = SegmentId::from_bytes(id_bytes);
2151 let actual_length = u64::try_from(bytes.len())
2152 .map_err(|_| SegmentError::Geometry("byte slice length exceeds u64".to_owned()))?;
2153 let parsed = parse_segment_header("segment-bytes", bytes, actual_length, id)?;
2154 for entry in &parsed.entries {
2155 let start = usize::try_from(entry.offset)
2156 .map_err(|_| SegmentError::Geometry("region offset exceeds usize".to_owned()))?;
2157 let length = usize::try_from(entry.length)
2158 .map_err(|_| SegmentError::Geometry("region length exceeds usize".to_owned()))?;
2159 let end = start
2160 .checked_add(length)
2161 .ok_or_else(|| SegmentError::Geometry("region end overflow".to_owned()))?;
2162 let region = bytes.get(start..end).ok_or_else(|| {
2163 FormatError::new(
2164 "segment-bytes",
2165 FormatCheck::BlockLength,
2166 format!("region {} range {start}..{end} is truncated", entry.kind),
2167 )
2168 })?;
2169 let actual = xxh3_64(region);
2170 if actual != entry.checksum {
2171 return Err(FormatError::new(
2172 format!("segment-bytes:region-{}", entry.kind),
2173 FormatCheck::BlockChecksum,
2174 format!("expected {:#018x}, computed {actual:#018x}", entry.checksum),
2175 )
2176 .into());
2177 }
2178 }
2179 let trailer = bytes.len().checked_sub(FILE_TRAILER_LEN).ok_or_else(|| {
2180 FormatError::new("segment-bytes", FormatCheck::Length, "missing file trailer")
2181 })?;
2182 let expected = read_u64("segment-bytes", bytes, trailer)?;
2183 let checksummed = bytes.get(..trailer).ok_or_else(|| {
2184 FormatError::new(
2185 "segment-bytes",
2186 FormatCheck::Length,
2187 "invalid trailer offset",
2188 )
2189 })?;
2190 let actual = xxh3_64(checksummed);
2191 if actual != expected {
2192 return Err(FormatError::new(
2193 "segment-bytes",
2194 FormatCheck::FileChecksum,
2195 format!("expected {expected:#018x}, computed {actual:#018x}"),
2196 )
2197 .into());
2198 }
2199 Ok(parsed.meta)
2200}
2201
2202fn preflight_region_count(
2203 artifact: &str,
2204 bytes: &[u8],
2205 actual_file_length: u64,
2206) -> Result<usize, SegmentError> {
2207 let fixed = decode_header(artifact, FormatFamily::Segment, bytes)?;
2208 if fixed.file_length != actual_file_length {
2209 return Err(FormatError::new(
2210 artifact,
2211 FormatCheck::FileLength,
2212 format!(
2213 "declared {}, actual {actual_file_length}",
2214 fixed.file_length
2215 ),
2216 )
2217 .into());
2218 }
2219 let header_length = usize::try_from(fixed.header_length)
2220 .map_err(|_| SegmentError::Geometry("header length exceeds usize".to_owned()))?;
2221 validate_bounded_header_length(artifact, header_length, actual_file_length)?;
2222 let header_bytes = bytes.get(..header_length).ok_or_else(|| {
2223 FormatError::new(
2224 artifact,
2225 FormatCheck::Length,
2226 format!("declared header {header_length}, available {}", bytes.len()),
2227 )
2228 })?;
2229 Ok(usize::from(read_u16(
2230 artifact,
2231 header_bytes,
2232 FILE_HEADER_LEN.saturating_add(20),
2233 )?))
2234}
2235
2236fn parse_segment_header(
2237 artifact: &str,
2238 bytes: &[u8],
2239 actual_file_length: u64,
2240 expected_id: SegmentId,
2241) -> Result<ParsedHeader, SegmentError> {
2242 #[cfg(any(test, feature = "test-support"))]
2243 let actual_family = bytes
2244 .get(8..10)
2245 .and_then(|family| <[u8; 2]>::try_from(family).ok())
2246 .map(u16::from_le_bytes);
2247 let fixed = decode_header(artifact, FormatFamily::Segment, bytes).map_err(|error| {
2248 #[cfg(any(test, feature = "test-support"))]
2249 crate::lifecycle::record_storage_segment_format_fault(&error, actual_family, expected_id);
2250 SegmentError::from(error)
2251 })?;
2252 if fixed.file_length != actual_file_length {
2253 return Err(FormatError::new(
2254 artifact,
2255 FormatCheck::FileLength,
2256 format!(
2257 "declared {}, actual {actual_file_length}",
2258 fixed.file_length
2259 ),
2260 )
2261 .into());
2262 }
2263 let header_length = usize::try_from(fixed.header_length)
2264 .map_err(|_| SegmentError::Geometry("header length exceeds usize".to_owned()))?;
2265 validate_bounded_header_length(artifact, header_length, actual_file_length)?;
2266 let header_bytes = bytes.get(..header_length).ok_or_else(|| {
2267 FormatError::new(
2268 artifact,
2269 FormatCheck::Length,
2270 format!("declared header {header_length}, available {}", bytes.len()),
2271 )
2272 })?;
2273 let id_bytes: [u8; 16] = header_bytes
2274 .get(FILE_HEADER_LEN..FILE_HEADER_LEN.saturating_add(16))
2275 .ok_or_else(|| FormatError::new(artifact, FormatCheck::Length, "missing segment id"))?
2276 .try_into()
2277 .map_err(|_| FormatError::new(artifact, FormatCheck::Length, "invalid segment id"))?;
2278 let actual_id = SegmentId::from_bytes(id_bytes);
2279 if actual_id != expected_id {
2280 #[cfg(any(test, feature = "test-support"))]
2281 crate::lifecycle::record_storage_segment_identity_fault(artifact, expected_id, actual_id);
2282 return Err(SegmentError::WrongObject {
2283 artifact: artifact.to_owned(),
2284 expected: expected_id,
2285 actual: actual_id,
2286 });
2287 }
2288 let row_count = read_u32(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(16))?;
2289 let region_count = usize::from(read_u16(
2290 artifact,
2291 header_bytes,
2292 FILE_HEADER_LEN.saturating_add(20),
2293 )?);
2294 if read_u16(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(22))? != 0 {
2295 return Err(SegmentError::Geometry(
2296 "segment prefix reserved field is non-zero".to_owned(),
2297 ));
2298 }
2299 let scheme = read_u16(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(24))?;
2300 FormatRegistry::require_scheme(scheme)
2301 .map_err(|error| SegmentError::Geometry(error.to_string()))?;
2302 if read_u16(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(26))? != 0 {
2303 return Err(SegmentError::Geometry(
2304 "segment scheme padding is non-zero".to_owned(),
2305 ));
2306 }
2307 let dims = read_u32(artifact, header_bytes, FILE_HEADER_LEN.saturating_add(28))?;
2308 let expected_header_length = FILE_HEADER_LEN
2309 .checked_add(SEGMENT_PREFIX_LEN)
2310 .and_then(|value| value.checked_add(region_count.saturating_mul(REGION_ENTRY_LEN)))
2311 .and_then(|value| value.checked_add(8))
2312 .ok_or_else(|| SegmentError::Geometry("header length overflow".to_owned()))?;
2313 if header_length != expected_header_length {
2314 return Err(FormatError::new(
2315 artifact,
2316 FormatCheck::HeaderLength,
2317 format!("declared {header_length}, directory implies {expected_header_length}"),
2318 )
2319 .into());
2320 }
2321 let checksum_offset = header_length.saturating_sub(8);
2322 let expected_checksum = read_u64(artifact, header_bytes, checksum_offset)?;
2323 let checksum_bytes = header_bytes.get(..checksum_offset).ok_or_else(|| {
2324 FormatError::new(
2325 artifact,
2326 FormatCheck::Length,
2327 "invalid header checksum offset",
2328 )
2329 })?;
2330 let actual_checksum = xxh3_64(checksum_bytes);
2331 if actual_checksum != expected_checksum {
2332 return Err(FormatError::new(
2333 artifact,
2334 FormatCheck::BlockChecksum,
2335 format!("header expected {expected_checksum:#018x}, computed {actual_checksum:#018x}"),
2336 )
2337 .into());
2338 }
2339
2340 let mut entries: Vec<RegionEntry> = Vec::with_capacity(region_count);
2341 let directory_start = FILE_HEADER_LEN.saturating_add(SEGMENT_PREFIX_LEN);
2342 for position in 0..region_count {
2343 let offset = directory_start
2344 .checked_add(position.saturating_mul(REGION_ENTRY_LEN))
2345 .ok_or_else(|| SegmentError::Geometry("directory offset overflow".to_owned()))?;
2346 let entry = RegionEntry {
2347 kind: read_u16(artifact, header_bytes, offset)?,
2348 version: read_u16(artifact, header_bytes, offset.saturating_add(2))?,
2349 reserved: read_u32(artifact, header_bytes, offset.saturating_add(4))?,
2350 offset: read_u64(artifact, header_bytes, offset.saturating_add(8))?,
2351 length: read_u64(artifact, header_bytes, offset.saturating_add(16))?,
2352 checksum: read_u64(artifact, header_bytes, offset.saturating_add(24))?,
2353 };
2354 if entry.reserved != 0 {
2355 return Err(SegmentError::Geometry(format!(
2356 "region {} reserved field is non-zero",
2357 entry.kind
2358 )));
2359 }
2360 if entries.iter().any(|existing| existing.kind == entry.kind) {
2361 return Err(SegmentError::Geometry(format!(
2362 "duplicate region kind {}",
2363 entry.kind
2364 )));
2365 }
2366 validate_entry(artifact, &entry, header_length, actual_file_length)?;
2367 if let Some(kind) = RegionKind::from_id(entry.kind)
2368 && let Some(family) = kind.family()
2369 {
2370 FormatRegistry::require(family.id(), entry.version).map_err(|error| {
2371 FormatError::new(artifact, FormatCheck::Version, error.to_string())
2372 })?;
2373 }
2374 entries.push(entry);
2375 }
2376 validate_non_overlapping(&entries)?;
2377 Ok(ParsedHeader {
2378 meta: SegmentMeta {
2379 id: actual_id,
2380 row_count,
2381 scheme,
2382 dims,
2383 file_size: actual_file_length,
2384 epoch_id: None,
2385 clustering_key_range: super::ClusteringKeyRange::Unstamped,
2386 },
2387 header_length,
2388 entries,
2389 })
2390}
2391
2392fn validate_bounded_header_length(
2393 artifact: &str,
2394 header_length: usize,
2395 actual_file_length: u64,
2396) -> Result<(), SegmentError> {
2397 if !(FILE_HEADER_LEN + SEGMENT_PREFIX_LEN + 8..=MAX_SEGMENT_HEADER_BYTES)
2398 .contains(&header_length)
2399 {
2400 return Err(FormatError::new(
2401 artifact,
2402 FormatCheck::HeaderLength,
2403 format!("header length {header_length} is outside bounded v1 range"),
2404 )
2405 .into());
2406 }
2407 let file_length = usize::try_from(actual_file_length)
2408 .map_err(|_| SegmentError::Geometry("file length exceeds usize".to_owned()))?;
2409 if header_length > file_length.saturating_sub(FILE_TRAILER_LEN) {
2410 return Err(FormatError::new(
2411 artifact,
2412 FormatCheck::HeaderLength,
2413 format!("header {header_length} exceeds file {file_length}"),
2414 )
2415 .into());
2416 }
2417 Ok(())
2418}
2419
2420fn validate_entry(
2421 artifact: &str,
2422 entry: &RegionEntry,
2423 header_length: usize,
2424 file_length: u64,
2425) -> Result<(), SegmentError> {
2426 let offset = usize::try_from(entry.offset)
2427 .map_err(|_| SegmentError::Geometry("region offset exceeds usize".to_owned()))?;
2428 let length = usize::try_from(entry.length)
2429 .map_err(|_| SegmentError::Geometry("region length exceeds usize".to_owned()))?;
2430 let minimum = align_up(header_length, REGION_ALIGNMENT)?;
2431 if offset < minimum || !offset.is_multiple_of(REGION_ALIGNMENT) {
2432 return Err(SegmentError::Geometry(format!(
2433 "region {} offset {offset} is below/aligned from {minimum}",
2434 entry.kind
2435 )));
2436 }
2437 let end = offset
2438 .checked_add(length)
2439 .ok_or_else(|| SegmentError::Geometry("region end overflow".to_owned()))?;
2440 let maximum = usize::try_from(file_length)
2441 .map_err(|_| SegmentError::Geometry("file length exceeds usize".to_owned()))?
2442 .checked_sub(FILE_TRAILER_LEN)
2443 .ok_or_else(|| SegmentError::Geometry("file is shorter than trailer".to_owned()))?;
2444 if end > maximum {
2445 return Err(FormatError::new(
2446 artifact,
2447 FormatCheck::BlockLength,
2448 format!("region {} ends at {end}, maximum {maximum}", entry.kind),
2449 )
2450 .into());
2451 }
2452 Ok(())
2453}
2454
2455fn validate_non_overlapping(entries: &[RegionEntry]) -> Result<(), SegmentError> {
2456 for (position, entry) in entries.iter().enumerate() {
2457 let entry_end = entry
2458 .offset
2459 .checked_add(entry.length)
2460 .ok_or_else(|| SegmentError::Geometry("region u64 end overflow".to_owned()))?;
2461 for other in entries.iter().skip(position.saturating_add(1)) {
2462 let other_end = other
2463 .offset
2464 .checked_add(other.length)
2465 .ok_or_else(|| SegmentError::Geometry("region u64 end overflow".to_owned()))?;
2466 if entry.offset < other_end && other.offset < entry_end {
2467 let (previous, current, previous_end) = if entry.offset <= other.offset {
2468 (entry, other, entry_end)
2469 } else {
2470 (other, entry, other_end)
2471 };
2472 return Err(SegmentError::Geometry(format!(
2473 "region {} begins {} before previous end {previous_end} for region {}",
2474 current.kind, current.offset, previous.kind
2475 )));
2476 }
2477 }
2478 }
2479 Ok(())
2480}
2481
2482fn cast_slice<'a, T>(
2483 bytes: &'a [u8],
2484 count: usize,
2485 artifact: &str,
2486) -> Result<&'a [T], SegmentError> {
2487 #[cfg(target_endian = "big")]
2488 {
2489 let _ = (bytes, count);
2490 return Err(SegmentError::Geometry(format!(
2491 "{artifact} zero-copy requires little-endian target"
2492 )));
2493 }
2494 #[cfg(target_endian = "little")]
2495 {
2496 let expected = count
2497 .checked_mul(std::mem::size_of::<T>())
2498 .ok_or_else(|| SegmentError::Geometry(format!("{artifact} byte length overflow")))?;
2499 if bytes.len() != expected {
2500 return Err(SegmentError::Geometry(format!(
2501 "{artifact} bytes {}, expected {expected}",
2502 bytes.len()
2503 )));
2504 }
2505 if !(bytes.as_ptr() as usize).is_multiple_of(std::mem::align_of::<T>()) {
2506 return Err(SegmentError::Geometry(format!(
2507 "{artifact} address is not {}-byte aligned",
2508 std::mem::align_of::<T>()
2509 )));
2510 }
2511 Ok(unsafe { std::slice::from_raw_parts(bytes.as_ptr().cast::<T>(), count) })
2515 }
2516}