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