1pub(crate) mod bmp;
4pub(crate) mod loader;
5mod types;
6
7pub use bmp::{BmpDimStats, BmpIndex};
8#[cfg(feature = "native")]
9pub(crate) use types::DimRawData;
10pub use types::{SparseIndex, VectorIndex, VectorSearchResult};
11
12const MAX_PREFIX_TERMS: usize = 1_024;
16const MAX_PREFIX_POSTINGS: u64 = 5_000_000;
17const MAX_DENSE_CANDIDATES_PER_SEGMENT: usize = 20_000;
21const DENSE_SCORE_BATCH: usize = 4_096;
23const BINARY_SCORE_BATCH: usize = 8_192;
24const MAX_VECTOR_SCORE_BATCH_BYTES: usize = 8 * 1024 * 1024;
25
26#[derive(Debug, Clone, Default)]
32pub struct SegmentMemoryStats {
33 pub segment_id: u128,
35 pub num_docs: u32,
37 pub term_dict_cache_bytes: usize,
39 pub store_cache_bytes: usize,
41 pub sparse_heap_bytes: usize,
43 pub dense_heap_bytes: usize,
45 pub term_bloom_file_bytes: u64,
47 pub sparse_file_backed_bytes: u64,
49 pub dense_file_backed_bytes: u64,
51 pub pinned_metadata_bytes: u64,
53 pub pin_intended_bytes: u64,
56 pub sparse_pinned_metadata_bytes: u64,
58 pub sparse_pin_intended_bytes: u64,
60 pub dense_pinned_metadata_bytes: u64,
62 pub dense_pin_intended_bytes: u64,
64}
65
66impl SegmentMemoryStats {
67 pub fn estimated_heap_bytes(&self) -> usize {
69 self.term_dict_cache_bytes
70 + self.store_cache_bytes
71 + self.sparse_heap_bytes
72 + self.dense_heap_bytes
73 }
74
75 pub fn file_backed_bytes(&self) -> u64 {
79 self.term_bloom_file_bytes
80 .saturating_add(self.sparse_file_backed_bytes)
81 .saturating_add(self.dense_file_backed_bytes)
82 }
83}
84
85use std::cmp::Ordering;
86use std::collections::BinaryHeap;
87use std::sync::Arc;
88
89use rustc_hash::{FxHashMap, FxHashSet};
90
91use super::vector_data::LazyFlatVectorData;
92use crate::directories::{Directory, FileHandle};
93use crate::dsl::{DenseVectorQuantization, Document, Field, Schema};
94use crate::query::{MAX_DENSE_NPROBE, MAX_DENSE_RERANK_FACTOR};
95use crate::structures::{
96 AsyncSSTableReader, BlockPostingList, CoarseCentroids, SSTableStats, TermInfo,
97};
98use crate::{DocId, Error, Result};
99
100use super::store::{AsyncStoreReader, RawStoreBlock};
101use super::types::{SegmentFiles, SegmentId, SegmentMeta};
102
103pub(crate) fn combine_ordinal_results(
109 raw: impl IntoIterator<Item = (u32, u16, f32)>,
110 combiner: crate::query::MultiValueCombiner,
111 limit: usize,
112) -> Vec<VectorSearchResult> {
113 let collected: Vec<(u32, u16, f32)> = raw.into_iter().collect();
114
115 let num_raw = collected.len();
116 if log::log_enabled!(log::Level::Debug) {
117 let mut ids: Vec<u32> = collected.iter().map(|(d, _, _)| *d).collect();
118 ids.sort_unstable();
119 ids.dedup();
120 log::debug!(
121 "combine_ordinal_results: {} raw entries, {} unique docs, combiner={:?}, limit={}",
122 num_raw,
123 ids.len(),
124 combiner,
125 limit
126 );
127 }
128
129 let all_single = collected.iter().all(|&(_, ord, _)| ord == 0);
131 if all_single {
132 let mut results: Vec<VectorSearchResult> = collected
133 .into_iter()
134 .map(|(doc_id, _, score)| VectorSearchResult::new(doc_id, score, vec![(0, score)]))
135 .collect();
136 results.sort_unstable_by(|a, b| {
137 b.score
138 .total_cmp(&a.score)
139 .then_with(|| a.doc_id.cmp(&b.doc_id))
140 });
141 results.truncate(limit);
142 return results;
143 }
144
145 let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
147 rustc_hash::FxHashMap::default();
148 for (doc_id, ordinal, score) in collected {
149 doc_ordinals
150 .entry(doc_id as DocId)
151 .or_default()
152 .push((ordinal as u32, score));
153 }
154 let mut results: Vec<VectorSearchResult> = doc_ordinals
155 .into_iter()
156 .map(|(doc_id, ordinals)| {
157 let combined_score = combiner.combine(&ordinals);
158 VectorSearchResult::new(doc_id, combined_score, ordinals)
159 })
160 .collect();
161 results.sort_unstable_by(|a, b| {
162 b.score
163 .total_cmp(&a.score)
164 .then_with(|| a.doc_id.cmp(&b.doc_id))
165 });
166 results.truncate(limit);
167 results
168}
169
170struct HeapVectorResult(VectorSearchResult);
175
176impl PartialEq for HeapVectorResult {
177 fn eq(&self, other: &Self) -> bool {
178 self.0.score.to_bits() == other.0.score.to_bits() && self.0.doc_id == other.0.doc_id
179 }
180}
181
182impl Eq for HeapVectorResult {}
183
184impl Ord for HeapVectorResult {
185 fn cmp(&self, other: &Self) -> Ordering {
186 other
189 .0
190 .score
191 .total_cmp(&self.0.score)
192 .then_with(|| self.0.doc_id.cmp(&other.0.doc_id))
193 }
194}
195
196impl PartialOrd for HeapVectorResult {
197 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
198 Some(self.cmp(other))
199 }
200}
201
202struct FlatDocumentCollector {
206 heap: BinaryHeap<HeapVectorResult>,
207 limit: usize,
208 combiner: crate::query::MultiValueCombiner,
209 current_doc: Option<DocId>,
210 current_ordinals: Vec<(u32, f32)>,
211}
212
213impl FlatDocumentCollector {
214 fn new(limit: usize, combiner: crate::query::MultiValueCombiner) -> Self {
215 Self {
216 heap: BinaryHeap::with_capacity(limit.min(8 * 1024)),
217 limit,
218 combiner,
219 current_doc: None,
220 current_ordinals: Vec::new(),
221 }
222 }
223
224 fn push(&mut self, doc_id: DocId, ordinal: u16, score: f32) {
225 if self.current_doc.is_some_and(|current| current != doc_id) {
226 self.finish_current();
227 }
228 self.current_doc = Some(doc_id);
229 self.current_ordinals.push((ordinal as u32, score));
230 }
231
232 fn finish_current(&mut self) {
233 let Some(doc_id) = self.current_doc.take() else {
234 return;
235 };
236 let score = self.combiner.combine(&self.current_ordinals);
237 let should_retain = self.heap.len() < self.limit
238 || self.heap.peek().is_some_and(|worst| {
239 HeapVectorResult(VectorSearchResult::new(doc_id, score, Vec::new()))
240 .cmp(worst)
241 .is_lt()
242 });
243
244 if !should_retain {
245 self.current_ordinals.clear();
249 return;
250 }
251
252 let ordinals = std::mem::take(&mut self.current_ordinals);
253 let entry = HeapVectorResult(VectorSearchResult::new(doc_id, score, ordinals));
254 if self.heap.len() < self.limit {
255 self.heap.push(entry);
256 } else if let Some(mut worst) = self.heap.peek_mut() {
257 let mut evicted = std::mem::replace(&mut worst.0, entry.0);
260 evicted.ordinals.clear();
261 self.current_ordinals = evicted.ordinals;
262 }
263 }
264
265 fn into_results(mut self) -> Vec<VectorSearchResult> {
266 self.finish_current();
267 let mut results: Vec<_> = self.heap.into_iter().map(|entry| entry.0).collect();
268 results.sort_unstable_by(|a, b| {
269 b.score
270 .total_cmp(&a.score)
271 .then_with(|| a.doc_id.cmp(&b.doc_id))
272 });
273 results
274 }
275}
276
277fn combine_grouped_ordinal_results(
280 raw: impl IntoIterator<Item = RawVectorCandidate>,
281 combiner: crate::query::MultiValueCombiner,
282 limit: usize,
283) -> Vec<VectorSearchResult> {
284 let mut collector = FlatDocumentCollector::new(limit, combiner);
285 for (doc_id, ordinal, score) in raw {
286 collector.push(doc_id, ordinal, score);
287 }
288 collector.into_results()
289}
290
291#[derive(Clone, Copy)]
292struct DenseSearchParams {
293 dim: usize,
294 nprobe: usize,
295 unit_norm: bool,
296}
297
298struct PreparedDenseScoreQuery<'a> {
305 query: &'a [f32],
306 query_f16: Vec<u16>,
307 inv_norm_q: f32,
308 quantization: DenseVectorQuantization,
309 dim: usize,
310 unit_norm: bool,
311}
312
313impl<'a> PreparedDenseScoreQuery<'a> {
314 fn new(
315 query: &'a [f32],
316 quantization: DenseVectorQuantization,
317 dim: usize,
318 unit_norm: bool,
319 ) -> Result<Self> {
320 use crate::structures::simd;
321
322 if query.len() != dim {
323 return Err(Error::Query(format!(
324 "dense SIMD query dimension {} does not match vector dimension {dim}",
325 query.len()
326 )));
327 }
328 if quantization == DenseVectorQuantization::Binary {
329 return Err(Error::InvalidFieldType {
330 expected: "non-binary dense vector".to_string(),
331 got: "binary dense vector".to_string(),
332 });
333 }
334
335 let norm_q_sq = simd::dot_product_f32(query, query, dim);
336 let inv_norm_q = if norm_q_sq < f32::EPSILON {
337 0.0
338 } else {
339 simd::fast_inv_sqrt(norm_q_sq)
340 };
341 let query_f16 = if quantization == DenseVectorQuantization::F16 {
342 query.iter().map(|&value| simd::f32_to_f16(value)).collect()
343 } else {
344 Vec::new()
345 };
346
347 Ok(Self {
348 query,
349 query_f16,
350 inv_norm_q,
351 quantization,
352 dim,
353 unit_norm,
354 })
355 }
356
357 fn score_batch(&self, raw: &[u8], scores: &mut [f32]) -> Result<()> {
358 use crate::structures::simd;
359
360 let element_size = match self.quantization {
361 DenseVectorQuantization::F32 => std::mem::size_of::<f32>(),
362 DenseVectorQuantization::F16 => std::mem::size_of::<u16>(),
363 DenseVectorQuantization::UInt8 => 1,
364 DenseVectorQuantization::Binary => {
365 return Err(Error::InvalidFieldType {
366 expected: "non-binary dense vector".to_string(),
367 got: "binary dense vector".to_string(),
368 });
369 }
370 };
371 let required_bytes = scores
372 .len()
373 .checked_mul(self.dim)
374 .and_then(|elements| elements.checked_mul(element_size))
375 .ok_or_else(|| Error::Corruption("dense vector batch byte length overflow".into()))?;
376 if raw.len() < required_bytes {
377 return Err(Error::Corruption(format!(
378 "dense vector batch is truncated: need {required_bytes} bytes, got {}",
379 raw.len()
380 )));
381 }
382 if self.quantization == DenseVectorQuantization::F16
383 && required_bytes > 0
384 && !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<u16>())
385 {
386 return Err(Error::Corruption(
387 "f16 vector data is not 2-byte aligned".to_string(),
388 ));
389 }
390 if self.quantization == DenseVectorQuantization::F32
391 && !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>())
392 {
393 return Err(Error::Corruption(
394 "f32 vector data is not 4-byte aligned".to_string(),
395 ));
396 }
397
398 if self.dim == 0 || scores.is_empty() {
402 return Ok(());
403 }
404
405 match (self.quantization, self.unit_norm) {
406 (DenseVectorQuantization::F32, false) => {
407 let num_floats = scores.len() * self.dim;
408 let vectors: &[f32] =
409 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
410 simd::batch_cosine_scores_precomp(
411 self.query,
412 vectors,
413 self.dim,
414 scores,
415 self.inv_norm_q,
416 );
417 }
418 (DenseVectorQuantization::F32, true) => {
419 let num_floats = scores.len() * self.dim;
420 let vectors: &[f32] =
421 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
422 simd::batch_dot_scores_precomp(
423 self.query,
424 vectors,
425 self.dim,
426 scores,
427 self.inv_norm_q,
428 );
429 }
430 (DenseVectorQuantization::F16, false) => {
431 simd::batch_cosine_scores_f16_precomp(
432 &self.query_f16,
433 raw,
434 self.dim,
435 scores,
436 self.inv_norm_q,
437 );
438 }
439 (DenseVectorQuantization::F16, true) => {
440 simd::batch_dot_scores_f16_precomp(
441 &self.query_f16,
442 raw,
443 self.dim,
444 scores,
445 self.inv_norm_q,
446 );
447 }
448 (DenseVectorQuantization::UInt8, false) => {
449 simd::batch_cosine_scores_u8_precomp(
450 self.query,
451 raw,
452 self.dim,
453 scores,
454 self.inv_norm_q,
455 );
456 }
457 (DenseVectorQuantization::UInt8, true) => {
458 simd::batch_dot_scores_u8_precomp(
459 self.query,
460 raw,
461 self.dim,
462 scores,
463 self.inv_norm_q,
464 );
465 }
466 (DenseVectorQuantization::Binary, _) => unreachable!("validated during preparation"),
467 }
468 Ok(())
469 }
470}
471
472fn checked_dense_fetch_k(k: usize, rerank_factor: f32) -> Result<usize> {
474 if !rerank_factor.is_finite() || !(1.0..=MAX_DENSE_RERANK_FACTOR).contains(&rerank_factor) {
475 return Err(Error::Query(format!(
476 "dense rerank_factor must be finite and in [1, {MAX_DENSE_RERANK_FACTOR}], got {rerank_factor}"
477 )));
478 }
479
480 let fetch = (k as f64) * (rerank_factor as f64);
481 if !fetch.is_finite()
482 || fetch > usize::MAX as f64
483 || fetch > MAX_DENSE_CANDIDATES_PER_SEGMENT as f64
484 {
485 return Err(Error::Query(format!(
486 "dense candidate count exceeds the per-segment maximum of \
487 {MAX_DENSE_CANDIDATES_PER_SEGMENT}: k={k}, rerank_factor={rerank_factor}"
488 )));
489 }
490 Ok(fetch.ceil() as usize)
491}
492
493#[inline]
499fn checked_binary_combined_fetch_k(k: usize) -> Result<usize> {
500 if k > MAX_DENSE_CANDIDATES_PER_SEGMENT {
501 return Err(Error::Query(format!(
502 "binary dense result count exceeds the per-segment maximum of \
503 {MAX_DENSE_CANDIDATES_PER_SEGMENT}: k={k}"
504 )));
505 }
506 Ok(crate::query::max_candidate_limit(k).min(MAX_DENSE_CANDIDATES_PER_SEGMENT))
507}
508
509#[inline]
510fn bounded_vector_score_batch(vector_byte_size: usize, preferred: usize) -> usize {
511 preferred.min((MAX_VECTOR_SCORE_BATCH_BYTES / vector_byte_size.max(1)).max(1))
512}
513
514#[inline]
515fn bounded_rerank_batch(vector_byte_size: usize, preferred: usize, vector_count: usize) -> usize {
516 bounded_vector_score_batch(vector_byte_size, preferred).min(vector_count.max(1))
517}
518
519fn checked_file_range(
520 offset: u64,
521 length: u64,
522 file_length: u64,
523 description: &str,
524) -> Result<std::ops::Range<u64>> {
525 let end = offset
526 .checked_add(length)
527 .ok_or_else(|| Error::Corruption(format!("{description} byte range overflows u64")))?;
528 if end > file_length {
529 return Err(Error::Corruption(format!(
530 "{description} byte range {offset}..{end} exceeds file length {file_length}"
531 )));
532 }
533 Ok(offset..end)
534}
535
536type RawVectorCandidate = (u32, u16, f32);
537type CandidateVectorRef = (DocId, u16, usize); #[derive(Clone, Copy)]
540struct CandidateDocumentRange {
541 doc_id: DocId,
542 start: usize,
543 end: usize,
544}
545
546struct AnnCandidateDocuments {
547 ranges: Vec<CandidateDocumentRange>,
548 vector_count: usize,
549}
550
551fn ann_candidate_document_ranges(
560 ann_results: &[RawVectorCandidate],
561 flat: &LazyFlatVectorData,
562) -> Result<AnnCandidateDocuments> {
563 ann_candidate_document_ranges_from_ids(ann_results.iter().map(|candidate| candidate.0), flat)
564}
565
566fn ann_candidate_document_ranges_from_ids(
567 doc_ids: impl IntoIterator<Item = DocId>,
568 flat: &LazyFlatVectorData,
569) -> Result<AnnCandidateDocuments> {
570 let mut candidate_docs: Vec<DocId> = doc_ids.into_iter().collect();
571 candidate_docs.sort_unstable();
572 candidate_docs.dedup();
573
574 let mut ranges = Vec::with_capacity(candidate_docs.len());
575 let mut vector_count = 0usize;
576 for doc_id in candidate_docs {
577 let (start, count) = flat.flat_indexes_for_doc_range(doc_id);
578 if count == 0 {
579 return Err(Error::Corruption(format!(
580 "ANN candidate document {doc_id} is missing from flat vector storage"
581 )));
582 }
583 vector_count = vector_count
584 .checked_add(count)
585 .ok_or_else(|| Error::Query("ANN candidate vector expansion overflow".to_string()))?;
586 let end = start
587 .checked_add(count)
588 .ok_or_else(|| Error::Corruption("flat vector range overflow".to_string()))?;
589 if end > flat.num_vectors {
590 return Err(Error::Corruption(format!(
591 "flat vector range {start}..{end} for document {doc_id} exceeds {} vectors",
592 flat.num_vectors
593 )));
594 }
595 ranges.push(CandidateDocumentRange { doc_id, start, end });
596 }
597 Ok(AnnCandidateDocuments {
598 ranges,
599 vector_count,
600 })
601}
602
603fn validate_binary_single_value_ann_results(
611 ann_results: Vec<RawVectorCandidate>,
612 flat: &LazyFlatVectorData,
613) -> Result<Vec<RawVectorCandidate>> {
614 let mut seen_docs = FxHashSet::default();
615 let mut validated = Vec::with_capacity(ann_results.len());
616 for (doc_id, ordinal, score) in ann_results {
617 let (start, count) = flat.flat_indexes_for_doc_range(doc_id);
618 if count == 0 {
619 return Err(Error::Corruption(format!(
620 "ANN candidate document {doc_id} is missing from flat vector storage"
621 )));
622 }
623 if count != 1 {
624 return Err(Error::Corruption(format!(
625 "binary ANN single-valued candidate document {doc_id} has {count} flat vectors"
626 )));
627 }
628 let (stored_doc_id, stored_ordinal) = flat.get_doc_id(start);
629 if stored_doc_id != doc_id {
630 return Err(Error::Corruption(format!(
631 "flat vector doc map is not contiguous for document {doc_id}"
632 )));
633 }
634 if stored_ordinal != ordinal {
635 return Err(Error::Corruption(format!(
636 "binary ANN candidate document {doc_id} ordinal {ordinal} is missing from flat vector storage"
637 )));
638 }
639 if seen_docs.insert(doc_id) {
640 validated.push((doc_id, ordinal, score));
641 }
642 }
643 Ok(validated)
644}
645
646struct CandidateVectorCursor<'a> {
647 ranges: &'a [CandidateDocumentRange],
648 range_index: usize,
649 flat_index: usize,
650}
651
652impl<'a> CandidateVectorCursor<'a> {
653 fn new(ranges: &'a [CandidateDocumentRange]) -> Self {
654 Self {
655 ranges,
656 range_index: 0,
657 flat_index: ranges.first().map_or(0, |range| range.start),
658 }
659 }
660
661 fn fill_batch(
665 &mut self,
666 flat: &LazyFlatVectorData,
667 batch: &mut Vec<CandidateVectorRef>,
668 limit: usize,
669 ) -> Result<bool> {
670 batch.clear();
671 while batch.len() < limit && self.range_index < self.ranges.len() {
672 let range = self.ranges[self.range_index];
673 if self.flat_index == range.end {
674 self.range_index += 1;
675 if let Some(next) = self.ranges.get(self.range_index) {
676 self.flat_index = next.start;
677 }
678 continue;
679 }
680 let (stored_doc_id, ordinal) = flat.get_doc_id(self.flat_index);
681 if stored_doc_id != range.doc_id {
682 return Err(Error::Corruption(format!(
683 "flat vector doc map is not contiguous for document {}",
684 range.doc_id
685 )));
686 }
687 batch.push((range.doc_id, ordinal, self.flat_index));
688 self.flat_index += 1;
689 }
690 Ok(!batch.is_empty())
691 }
692}
693
694#[derive(Clone, Copy)]
695struct VectorReadRun {
696 buffer_start: usize,
697 flat_start: usize,
698 count: usize,
699}
700
701fn plan_vector_read_runs(indexes: &[usize], runs: &mut Vec<VectorReadRun>) -> Result<()> {
706 runs.clear();
707 for (buffer_index, &flat_index) in indexes.iter().enumerate() {
708 if let Some(run) = runs.last_mut()
709 && run
710 .flat_start
711 .checked_add(run.count)
712 .is_some_and(|next| next == flat_index)
713 {
714 run.count += 1;
715 continue;
716 }
717 if buffer_index > 0 && flat_index <= indexes[buffer_index - 1] {
718 return Err(Error::Corruption(
719 "candidate flat-vector indexes are not strictly ordered".into(),
720 ));
721 }
722 runs.push(VectorReadRun {
723 buffer_start: buffer_index,
724 flat_start: flat_index,
725 count: 1,
726 });
727 }
728 Ok(())
729}
730
731fn prepare_vector_read_runs(
735 flat: &LazyFlatVectorData,
736 indexes: &[usize],
737 runs: &mut Vec<VectorReadRun>,
738) -> Result<()> {
739 plan_vector_read_runs(indexes, runs)?;
740 #[cfg(feature = "native")]
741 flat.prefetch_vectors(indexes.iter().copied());
742 #[cfg(not(feature = "native"))]
743 let _ = flat;
744 Ok(())
745}
746
747async fn read_vector_runs(
748 flat: &LazyFlatVectorData,
749 indexes: &[usize],
750 runs: &mut Vec<VectorReadRun>,
751 output: &mut [u8],
752) -> Result<()> {
753 prepare_vector_read_runs(flat, indexes, runs)?;
754 let vector_byte_size = flat.vector_byte_size();
755 for run in runs {
756 let bytes = flat
757 .read_vectors_batch(run.flat_start, run.count)
758 .await
759 .map_err(Error::Io)?;
760 let start = run
761 .buffer_start
762 .checked_mul(vector_byte_size)
763 .ok_or_else(|| Error::Query("dense rerank buffer offset overflow".into()))?;
764 let end = start
765 .checked_add(bytes.len())
766 .ok_or_else(|| Error::Query("dense rerank buffer range overflow".into()))?;
767 let destination = output
768 .get_mut(start..end)
769 .ok_or_else(|| Error::Corruption("dense rerank buffer is too short".into()))?;
770 destination.copy_from_slice(bytes.as_slice());
771 }
772 Ok(())
773}
774
775#[cfg(feature = "sync")]
776fn read_vector_runs_sync(
777 flat: &LazyFlatVectorData,
778 indexes: &[usize],
779 runs: &mut Vec<VectorReadRun>,
780 output: &mut [u8],
781) -> Result<()> {
782 prepare_vector_read_runs(flat, indexes, runs)?;
783 let vector_byte_size = flat.vector_byte_size();
784 for run in runs {
785 let bytes = flat
786 .read_vectors_batch_sync(run.flat_start, run.count)
787 .map_err(Error::Io)?;
788 let start = run
789 .buffer_start
790 .checked_mul(vector_byte_size)
791 .ok_or_else(|| Error::Query("dense rerank buffer offset overflow".into()))?;
792 let end = start
793 .checked_add(bytes.len())
794 .ok_or_else(|| Error::Query("dense rerank buffer range overflow".into()))?;
795 let destination = output
796 .get_mut(start..end)
797 .ok_or_else(|| Error::Corruption("dense rerank buffer is too short".into()))?;
798 destination.copy_from_slice(bytes.as_slice());
799 }
800 Ok(())
801}
802
803#[derive(Default)]
804struct DenseRerankStats {
805 vector_count: usize,
806 resolve_elapsed: std::time::Duration,
807 read_elapsed: std::time::Duration,
808 score_elapsed: std::time::Duration,
809}
810
811async fn exact_score_dense_candidate_documents(
812 ann_results: &[RawVectorCandidate],
813 flat: &LazyFlatVectorData,
814 query: &[f32],
815 unit_norm: bool,
816 combiner: crate::query::MultiValueCombiner,
817 limit: usize,
818) -> Result<(Vec<VectorSearchResult>, DenseRerankStats)> {
819 let resolve_started = std::time::Instant::now();
820 let documents = ann_candidate_document_ranges(ann_results, flat)?;
821 let mut stats = DenseRerankStats {
822 vector_count: documents.vector_count,
823 resolve_elapsed: resolve_started.elapsed(),
824 ..Default::default()
825 };
826 let vector_byte_size = flat.vector_byte_size();
827 let batch_len =
828 bounded_rerank_batch(vector_byte_size, DENSE_SCORE_BATCH, documents.vector_count);
829 let raw_capacity = batch_len
830 .checked_mul(vector_byte_size)
831 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".to_string()))?;
832 let mut raw = vec![0u8; raw_capacity];
833 let mut scores = vec![0.0f32; batch_len];
834 let mut batch = Vec::with_capacity(batch_len);
835 let mut flat_indexes = Vec::with_capacity(batch_len);
836 let mut read_runs = Vec::new();
837 let mut cursor = CandidateVectorCursor::new(&documents.ranges);
838 let mut collector = FlatDocumentCollector::new(limit, combiner);
839 let mut scored = 0usize;
840 let prepared_query =
841 PreparedDenseScoreQuery::new(query, flat.quantization, flat.dim, unit_norm)?;
842
843 while cursor.fill_batch(flat, &mut batch, batch_len)? {
844 flat_indexes.clear();
845 flat_indexes.extend(batch.iter().map(|&(_, _, flat_index)| flat_index));
846 let raw_len = batch
847 .len()
848 .checked_mul(vector_byte_size)
849 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".to_string()))?;
850 let raw = &mut raw[..raw_len];
851
852 let read_started = std::time::Instant::now();
853 read_vector_runs(flat, &flat_indexes, &mut read_runs, raw).await?;
854 stats.read_elapsed += read_started.elapsed();
855
856 let score_started = std::time::Instant::now();
857 prepared_query.score_batch(raw, &mut scores[..batch.len()])?;
858 stats.score_elapsed += score_started.elapsed();
859 for (buffer_index, &(doc_id, ordinal, _)) in batch.iter().enumerate() {
860 collector.push(doc_id, ordinal, scores[buffer_index]);
861 }
862 scored += batch.len();
863 }
864 debug_assert_eq!(scored, documents.vector_count);
865 Ok((collector.into_results(), stats))
866}
867
868#[cfg(feature = "sync")]
869fn exact_score_dense_candidate_documents_sync(
870 ann_results: &[RawVectorCandidate],
871 flat: &LazyFlatVectorData,
872 query: &[f32],
873 unit_norm: bool,
874 combiner: crate::query::MultiValueCombiner,
875 limit: usize,
876) -> Result<Vec<VectorSearchResult>> {
877 let documents = ann_candidate_document_ranges(ann_results, flat)?;
878 let vector_byte_size = flat.vector_byte_size();
879 let batch_len =
880 bounded_rerank_batch(vector_byte_size, DENSE_SCORE_BATCH, documents.vector_count);
881 let raw_capacity = batch_len
882 .checked_mul(vector_byte_size)
883 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".to_string()))?;
884 let mut raw = vec![0u8; raw_capacity];
885 let mut scores = vec![0.0f32; batch_len];
886 let mut batch = Vec::with_capacity(batch_len);
887 let mut flat_indexes = Vec::with_capacity(batch_len);
888 let mut read_runs = Vec::new();
889 let mut cursor = CandidateVectorCursor::new(&documents.ranges);
890 let mut collector = FlatDocumentCollector::new(limit, combiner);
891 let mut scored = 0usize;
892 let prepared_query =
893 PreparedDenseScoreQuery::new(query, flat.quantization, flat.dim, unit_norm)?;
894
895 while cursor.fill_batch(flat, &mut batch, batch_len)? {
896 flat_indexes.clear();
897 flat_indexes.extend(batch.iter().map(|&(_, _, flat_index)| flat_index));
898 let raw_len = batch
899 .len()
900 .checked_mul(vector_byte_size)
901 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".to_string()))?;
902 let raw = &mut raw[..raw_len];
903 read_vector_runs_sync(flat, &flat_indexes, &mut read_runs, raw)?;
904 prepared_query.score_batch(raw, &mut scores[..batch.len()])?;
905 for (buffer_index, &(doc_id, ordinal, _)) in batch.iter().enumerate() {
906 collector.push(doc_id, ordinal, scores[buffer_index]);
907 }
908 scored += batch.len();
909 }
910 debug_assert_eq!(scored, documents.vector_count);
911 Ok(collector.into_results())
912}
913
914async fn exact_score_binary_candidate_documents(
915 ann_results: &[RawVectorCandidate],
916 flat: &LazyFlatVectorData,
917 query: &[u8],
918 dim_bits: usize,
919 combiner: crate::query::MultiValueCombiner,
920 limit: usize,
921) -> Result<Vec<VectorSearchResult>> {
922 let documents = ann_candidate_document_ranges(ann_results, flat)?;
923 let probe_scores: FxHashMap<(DocId, u16), f32> = ann_results
924 .iter()
925 .map(|&(doc_id, ordinal, score)| ((doc_id, ordinal), score))
926 .collect();
927 exact_score_binary_resolved_documents(
928 documents,
929 &probe_scores,
930 flat,
931 query,
932 dim_bits,
933 combiner,
934 limit,
935 )
936 .await
937}
938
939async fn exact_score_binary_candidate_document_ids(
940 candidate_doc_ids: Vec<DocId>,
941 probed_ordinal_scores: &[(u32, u16, f32)],
942 flat: &LazyFlatVectorData,
943 query: &[u8],
944 dim_bits: usize,
945 combiner: crate::query::MultiValueCombiner,
946 limit: usize,
947) -> Result<Vec<VectorSearchResult>> {
948 let documents = ann_candidate_document_ranges_from_ids(candidate_doc_ids, flat)?;
949 let probe_scores = binary_probe_score_map(probed_ordinal_scores);
952 exact_score_binary_resolved_documents(
953 documents,
954 &probe_scores,
955 flat,
956 query,
957 dim_bits,
958 combiner,
959 limit,
960 )
961 .await
962}
963
964fn binary_probe_score_map(
965 probed_ordinal_scores: &[(u32, u16, f32)],
966) -> FxHashMap<(DocId, u16), f32> {
967 probed_ordinal_scores
968 .iter()
969 .map(|&(doc_id, ordinal, score)| ((doc_id, ordinal), score))
970 .collect()
971}
972
973async fn exact_score_binary_resolved_documents(
974 documents: AnnCandidateDocuments,
975 probe_scores: &FxHashMap<(DocId, u16), f32>,
976 flat: &LazyFlatVectorData,
977 query: &[u8],
978 dim_bits: usize,
979 combiner: crate::query::MultiValueCombiner,
980 limit: usize,
981) -> Result<Vec<VectorSearchResult>> {
982 let vector_byte_size = flat.vector_byte_size();
983 let batch_len =
984 bounded_rerank_batch(vector_byte_size, BINARY_SCORE_BATCH, documents.vector_count);
985 let raw_capacity = batch_len
986 .checked_mul(vector_byte_size)
987 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
988 let mut raw = vec![0u8; raw_capacity];
989 let mut scores = vec![0.0f32; batch_len];
990 let mut batch_scores = vec![0.0f32; batch_len];
991 let mut batch = Vec::with_capacity(batch_len);
992 let mut unresolved = Vec::with_capacity(batch_len);
993 let mut unresolved_flat_indexes = Vec::with_capacity(batch_len);
994 let mut read_runs = Vec::new();
995 let mut cursor = CandidateVectorCursor::new(&documents.ranges);
996 let mut collector = FlatDocumentCollector::new(limit, combiner);
997 let mut scored = 0usize;
998
999 while cursor.fill_batch(flat, &mut batch, batch_len)? {
1000 unresolved.clear();
1001 for (batch_index, &(doc_id, ordinal, flat_index)) in batch.iter().enumerate() {
1002 if let Some(&score) = probe_scores.get(&(doc_id, ordinal)) {
1003 batch_scores[batch_index] = score;
1004 } else {
1005 unresolved.push((batch_index, flat_index));
1006 }
1007 }
1008 unresolved_flat_indexes.clear();
1009 unresolved_flat_indexes.extend(unresolved.iter().map(|&(_, flat_index)| flat_index));
1010 let raw_len = unresolved
1011 .len()
1012 .checked_mul(vector_byte_size)
1013 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
1014 let raw = &mut raw[..raw_len];
1015 read_vector_runs(flat, &unresolved_flat_indexes, &mut read_runs, raw).await?;
1016 crate::structures::simd::batch_hamming_scores(
1017 query,
1018 raw,
1019 vector_byte_size,
1020 dim_bits,
1021 &mut scores[..unresolved.len()],
1022 );
1023 for (buffer_index, &(batch_index, _)) in unresolved.iter().enumerate() {
1024 batch_scores[batch_index] = scores[buffer_index];
1025 }
1026 for (batch_index, &(doc_id, ordinal, _)) in batch.iter().enumerate() {
1027 collector.push(doc_id, ordinal, batch_scores[batch_index]);
1028 }
1029 scored += batch.len();
1030 }
1031 debug_assert_eq!(scored, documents.vector_count);
1032 Ok(collector.into_results())
1033}
1034
1035#[cfg(feature = "sync")]
1036fn exact_score_binary_candidate_documents_sync(
1037 ann_results: &[RawVectorCandidate],
1038 flat: &LazyFlatVectorData,
1039 query: &[u8],
1040 dim_bits: usize,
1041 combiner: crate::query::MultiValueCombiner,
1042 limit: usize,
1043) -> Result<Vec<VectorSearchResult>> {
1044 let documents = ann_candidate_document_ranges(ann_results, flat)?;
1045 let probe_scores: FxHashMap<(DocId, u16), f32> = ann_results
1046 .iter()
1047 .map(|&(doc_id, ordinal, score)| ((doc_id, ordinal), score))
1048 .collect();
1049 exact_score_binary_resolved_documents_sync(
1050 documents,
1051 &probe_scores,
1052 flat,
1053 query,
1054 dim_bits,
1055 combiner,
1056 limit,
1057 )
1058}
1059
1060#[cfg(feature = "sync")]
1061fn exact_score_binary_candidate_document_ids_sync(
1062 candidate_doc_ids: Vec<DocId>,
1063 probed_ordinal_scores: &[(u32, u16, f32)],
1064 flat: &LazyFlatVectorData,
1065 query: &[u8],
1066 dim_bits: usize,
1067 combiner: crate::query::MultiValueCombiner,
1068 limit: usize,
1069) -> Result<Vec<VectorSearchResult>> {
1070 let documents = ann_candidate_document_ranges_from_ids(candidate_doc_ids, flat)?;
1071 let probe_scores = binary_probe_score_map(probed_ordinal_scores);
1072 exact_score_binary_resolved_documents_sync(
1073 documents,
1074 &probe_scores,
1075 flat,
1076 query,
1077 dim_bits,
1078 combiner,
1079 limit,
1080 )
1081}
1082
1083#[cfg(feature = "sync")]
1084fn exact_score_binary_resolved_documents_sync(
1085 documents: AnnCandidateDocuments,
1086 probe_scores: &FxHashMap<(DocId, u16), f32>,
1087 flat: &LazyFlatVectorData,
1088 query: &[u8],
1089 dim_bits: usize,
1090 combiner: crate::query::MultiValueCombiner,
1091 limit: usize,
1092) -> Result<Vec<VectorSearchResult>> {
1093 let vector_byte_size = flat.vector_byte_size();
1094 let batch_len =
1095 bounded_rerank_batch(vector_byte_size, BINARY_SCORE_BATCH, documents.vector_count);
1096 let raw_capacity = batch_len
1097 .checked_mul(vector_byte_size)
1098 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
1099 let mut raw = vec![0u8; raw_capacity];
1100 let mut scores = vec![0.0f32; batch_len];
1101 let mut batch_scores = vec![0.0f32; batch_len];
1102 let mut batch = Vec::with_capacity(batch_len);
1103 let mut unresolved = Vec::with_capacity(batch_len);
1104 let mut unresolved_flat_indexes = Vec::with_capacity(batch_len);
1105 let mut read_runs = Vec::new();
1106 let mut cursor = CandidateVectorCursor::new(&documents.ranges);
1107 let mut collector = FlatDocumentCollector::new(limit, combiner);
1108 let mut scored = 0usize;
1109
1110 while cursor.fill_batch(flat, &mut batch, batch_len)? {
1111 unresolved.clear();
1112 for (batch_index, &(doc_id, ordinal, flat_index)) in batch.iter().enumerate() {
1113 if let Some(&score) = probe_scores.get(&(doc_id, ordinal)) {
1114 batch_scores[batch_index] = score;
1115 } else {
1116 unresolved.push((batch_index, flat_index));
1117 }
1118 }
1119 unresolved_flat_indexes.clear();
1120 unresolved_flat_indexes.extend(unresolved.iter().map(|&(_, flat_index)| flat_index));
1121 let raw_len = unresolved
1122 .len()
1123 .checked_mul(vector_byte_size)
1124 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
1125 let raw = &mut raw[..raw_len];
1126 read_vector_runs_sync(flat, &unresolved_flat_indexes, &mut read_runs, raw)?;
1127 crate::structures::simd::batch_hamming_scores(
1128 query,
1129 raw,
1130 vector_byte_size,
1131 dim_bits,
1132 &mut scores[..unresolved.len()],
1133 );
1134 for (buffer_index, &(batch_index, _)) in unresolved.iter().enumerate() {
1135 batch_scores[batch_index] = scores[buffer_index];
1136 }
1137 for (batch_index, &(doc_id, ordinal, _)) in batch.iter().enumerate() {
1138 collector.push(doc_id, ordinal, batch_scores[batch_index]);
1139 }
1140 scored += batch.len();
1141 }
1142 debug_assert_eq!(scored, documents.vector_count);
1143 Ok(collector.into_results())
1144}
1145
1146fn validate_coarse_centroids(centroids: &CoarseCentroids, dim: usize) -> Result<()> {
1147 let expected = (centroids.num_clusters as usize)
1148 .checked_mul(dim)
1149 .ok_or_else(|| Error::Corruption("coarse centroid size overflow".into()))?;
1150 if centroids.num_clusters == 0
1151 || centroids.dim != dim
1152 || centroids.centroids.len() != expected
1153 || centroids.centroids.iter().any(|value| !value.is_finite())
1154 {
1155 return Err(Error::Corruption(format!(
1156 "invalid coarse centroids: clusters={}, dim={}, values={} (expected dim={dim}, values={expected})",
1157 centroids.num_clusters,
1158 centroids.dim,
1159 centroids.centroids.len()
1160 )));
1161 }
1162 Ok(())
1163}
1164
1165#[derive(Debug, Default)]
1170pub struct DensePlanCache {
1171 pub(crate) tq: std::sync::Mutex<Option<std::sync::Arc<crate::structures::TqQueryPlan>>>,
1172 pub(crate) ivf_tq: std::sync::Mutex<Option<std::sync::Arc<crate::structures::TqIvfQueryPlan>>>,
1173}
1174
1175#[allow(clippy::too_many_arguments)]
1180fn search_tq_segment(
1181 index: &crate::segment::ann_disk::AnnDiskIndex,
1182 codec: &crate::structures::TqCodec,
1183 query: &[f32],
1184 fetch_k: usize,
1185 document_combiner: Option<crate::query::MultiValueCombiner>,
1186 field: Field,
1187 dim: usize,
1188 plan_cache: Option<&std::sync::Mutex<Option<std::sync::Arc<crate::structures::TqQueryPlan>>>>,
1189) -> Result<Vec<RawVectorCandidate>> {
1190 validate_tq_ann(index, codec, dim, field)?;
1191 let plan = cached_tq_query_plan(codec, query, plan_cache)?;
1192 match document_combiner {
1193 Some(combiner) => index
1194 .search_tq_combined_documents(fetch_k, &plan, combiner)
1195 .map(|candidates| {
1196 candidates
1197 .into_iter()
1198 .map(|candidate| (candidate.doc_id, 0, 0.0))
1202 .collect()
1203 }),
1204 None => index.search_tq_distinct(fetch_k, &plan),
1205 }
1206 .map_err(|error| {
1207 Error::Corruption(format!("invalid TQ payload for field {}: {error}", field.0))
1208 })
1209}
1210
1211fn cached_tq_query_plan(
1214 codec: &crate::structures::TqCodec,
1215 query: &[f32],
1216 plan_cache: Option<&std::sync::Mutex<Option<std::sync::Arc<crate::structures::TqQueryPlan>>>>,
1217) -> Result<std::sync::Arc<crate::structures::TqQueryPlan>> {
1218 Ok(match plan_cache {
1219 Some(cache) => {
1220 let mut cached = cache
1221 .lock()
1222 .map_err(|_| Error::Internal("TQ plan cache is poisoned".into()))?;
1223 match cached.as_ref() {
1224 Some(plan)
1225 if plan.fingerprint() == codec.fingerprint() && plan.matches_query(query) =>
1226 {
1227 std::sync::Arc::clone(plan)
1228 }
1229 _ => {
1230 let plan =
1231 std::sync::Arc::new(crate::structures::TqQueryPlan::build(codec, query));
1232 *cached = Some(std::sync::Arc::clone(&plan));
1233 plan
1234 }
1235 }
1236 }
1237 None => std::sync::Arc::new(crate::structures::TqQueryPlan::build(codec, query)),
1238 })
1239}
1240
1241fn validate_tq_ann(
1242 index: &crate::segment::ann_disk::AnnDiskIndex,
1243 codec: &crate::structures::TqCodec,
1244 dim: usize,
1245 field: Field,
1246) -> Result<()> {
1247 let header = index.header();
1248 if header.dim != dim
1249 || codec.dim() != dim
1250 || header.code_size != codec.code_size()
1251 || header.quantizer_version != codec.fingerprint()
1252 || header.codebook_version != 0
1253 || header.num_clusters != 1
1254 {
1255 return Err(Error::Corruption(format!(
1256 "TQ payload for field {} does not match the codec derived from schema dimension {dim}",
1257 field.0,
1258 )));
1259 }
1260 Ok(())
1261}
1262
1263#[allow(clippy::too_many_arguments)]
1267fn search_ivf_tq_segment(
1268 index: &crate::segment::ann_disk::AnnDiskIndex,
1269 centroids: &CoarseCentroids,
1270 codec: &crate::structures::TqCodec,
1271 query: &[f32],
1272 fetch_k: usize,
1273 document_combiner: Option<crate::query::MultiValueCombiner>,
1274 field: Field,
1275 nprobe: usize,
1276 routing: crate::dsl::IvfRoutingMode,
1277 plan_cache: Option<
1278 &std::sync::Mutex<Option<std::sync::Arc<crate::structures::TqIvfQueryPlan>>>,
1279 >,
1280) -> Result<Vec<RawVectorCandidate>> {
1281 let effective_nprobe = nprobe.clamp(1, centroids.num_clusters as usize);
1282 let request_fingerprint = crate::structures::TqIvfQueryPlan::request_fingerprint_for(
1283 centroids,
1284 query,
1285 effective_nprobe,
1286 routing,
1287 );
1288 let build = || {
1289 std::sync::Arc::new(crate::structures::TqIvfQueryPlan::build(
1290 centroids,
1291 codec,
1292 query,
1293 effective_nprobe,
1294 routing,
1295 ))
1296 };
1297 let plan = match plan_cache {
1298 Some(cache) => {
1299 let mut cached = cache
1300 .lock()
1301 .map_err(|_| Error::Internal("IVF-TQ plan cache is poisoned".into()))?;
1302 match cached.as_ref() {
1303 Some(plan)
1304 if plan.quantizer_version == centroids.version
1305 && plan.fingerprint == codec.fingerprint()
1306 && plan.request_fingerprint == request_fingerprint
1307 && plan.cluster_ids.len() == effective_nprobe =>
1308 {
1309 std::sync::Arc::clone(plan)
1310 }
1311 _ => {
1312 let plan = build();
1313 *cached = Some(std::sync::Arc::clone(&plan));
1314 plan
1315 }
1316 }
1317 }
1318 None => build(),
1319 };
1320 let candidates = match document_combiner {
1321 Some(combiner) => index
1322 .search_ivf_tq_combined_documents(fetch_k, &plan, combiner)
1323 .map(|documents| {
1324 documents
1325 .into_iter()
1326 .map(|candidate| (candidate.doc_id, 0, 0.0))
1330 .collect()
1331 }),
1332 None => index.search_ivf_tq_distinct(fetch_k, &plan),
1333 };
1334 candidates.map_err(|error| {
1335 Error::Corruption(format!(
1336 "invalid IVF-TQ payload for field {}: {error}",
1337 field.0
1338 ))
1339 })
1340}
1341
1342fn validate_ivf_tq_ann(
1343 index: &crate::segment::ann_disk::AnnDiskIndex,
1344 centroids: &CoarseCentroids,
1345 codec: &crate::structures::TqCodec,
1346 dim: usize,
1347 routing: crate::dsl::IvfRoutingMode,
1348 field: Field,
1349) -> Result<()> {
1350 let header = index.header();
1351 if !crate::structures::is_ivf_tq_cosine_generation(centroids.version)
1352 || !crate::structures::is_ivf_tq_cosine_generation(header.quantizer_version)
1353 {
1354 return Err(Error::Corruption(format!(
1355 "IVF-TQ field {} uses a legacy unmarked raw-vector generation that cannot \
1356 preserve cosine candidate semantics; rebuild the index with a current \
1357 Hermes version",
1358 field.0,
1359 )));
1360 }
1361 if header.dim != dim
1362 || codec.dim() != dim
1363 || header.code_size != codec.code_size()
1364 || header.num_clusters != centroids.num_clusters
1365 || header.quantizer_version != centroids.version
1366 || header.codebook_version != codec.fingerprint()
1367 || header.routing != routing
1368 {
1369 return Err(Error::Corruption(format!(
1370 "IVF-TQ payload for field {} does not match its quantizer/codec generation",
1371 field.0,
1372 )));
1373 }
1374 Ok(())
1375}
1376
1377fn validate_binary_ann(
1378 index: &crate::segment::ann_disk::AnnDiskIndex,
1379 quantizer: &crate::structures::BinaryCoarseQuantizer,
1380 config: &crate::dsl::BinaryDenseVectorConfig,
1381 dim: usize,
1382 field: Field,
1383) -> Result<()> {
1384 let header = index.header();
1385 if header.dim != dim
1386 || header.code_size != config.byte_len()
1387 || header.num_clusters != quantizer.num_clusters
1388 || header.quantizer_version != quantizer.version
1389 || header.codebook_version != 0
1390 || header.routing != config.ivf_routing
1391 || quantizer.dim_bits != dim
1392 {
1393 return Err(Error::Corruption(format!(
1394 "binary IVF field {} does not match its quantizer/schema generation",
1395 field.0,
1396 )));
1397 }
1398 Ok(())
1399}
1400
1401fn binary_probe_clusters(
1402 quantizer: &crate::structures::BinaryCoarseQuantizer,
1403 query: &[u8],
1404 nprobe: usize,
1405 routing: crate::dsl::IvfRoutingMode,
1406 cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
1407) -> Result<std::sync::Arc<[u32]>> {
1408 let effective_nprobe = nprobe.clamp(1, quantizer.num_clusters as usize);
1409 let request_fingerprint = crate::structures::vector::ivf::routing::binary_probe_fingerprint(
1410 query,
1411 effective_nprobe,
1412 routing,
1413 );
1414 if let Some(cache) = cache {
1415 let mut cached = cache
1416 .lock()
1417 .map_err(|_| Error::Internal("binary IVF probe cache is poisoned".into()))?;
1418 if let Some(plan) = cached.as_ref()
1419 && plan.quantizer_version == quantizer.version
1420 && plan.request_fingerprint == request_fingerprint
1421 && plan.cluster_ids.len() == effective_nprobe
1422 {
1423 return Ok(std::sync::Arc::clone(&plan.cluster_ids));
1424 }
1425 let plan = quantizer.probe(query, effective_nprobe, routing);
1426 let clusters = std::sync::Arc::clone(&plan.cluster_ids);
1427 *cached = Some(plan);
1428 return Ok(clusters);
1429 }
1430 Ok(quantizer
1431 .probe(query, effective_nprobe, routing)
1432 .cluster_ids)
1433}
1434
1435pub struct SegmentReader {
1441 meta: SegmentMeta,
1442 term_dict: Arc<AsyncSSTableReader<TermInfo>>,
1444 postings_handle: FileHandle,
1446 store: Arc<AsyncStoreReader>,
1448 schema: Arc<Schema>,
1449 vector_indexes: FxHashMap<u32, VectorIndex>,
1451 flat_vectors: FxHashMap<u32, LazyFlatVectorData>,
1453 dense_file_backed_bytes: u64,
1455 trained_vectors: Arc<crate::segment::TrainedVectorStructures>,
1457 sparse_indexes: FxHashMap<u32, SparseIndex>,
1459 bmp_indexes: FxHashMap<u32, BmpIndex>,
1461 sparse_file_backed_bytes: u64,
1463 positions_handle: Option<FileHandle>,
1465 fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldReader>,
1467 #[cfg(feature = "native")]
1469 dense_pin_report: crate::segment::pin::PinReport,
1470 #[cfg(feature = "native")]
1472 sparse_pin_report: crate::segment::pin::PinReport,
1473}
1474
1475impl SegmentReader {
1476 pub async fn open<D: Directory>(
1478 dir: &D,
1479 segment_id: SegmentId,
1480 schema: Arc<Schema>,
1481 term_cache_blocks: usize,
1482 ) -> Result<Self> {
1483 Self::open_with_store_cache(
1484 dir,
1485 segment_id,
1486 schema,
1487 term_cache_blocks,
1488 dir as *const D as usize,
1489 Arc::new(super::SharedStoreCache::new(0)),
1490 )
1491 .await
1492 }
1493
1494 pub(crate) async fn open_with_store_cache<D: Directory>(
1496 dir: &D,
1497 segment_id: SegmentId,
1498 schema: Arc<Schema>,
1499 term_cache_blocks: usize,
1500 store_cache_directory_namespace: usize,
1501 store_cache: Arc<super::SharedStoreCache>,
1502 ) -> Result<Self> {
1503 let files = SegmentFiles::new(segment_id.0);
1504
1505 let meta_slice = dir.open_read(&files.meta).await?;
1507 let meta_bytes = meta_slice.read_bytes().await?;
1508 let meta = SegmentMeta::deserialize(meta_bytes.as_slice())?;
1509 debug_assert_eq!(meta.id, segment_id.0);
1510
1511 let term_dict_handle = dir.open_lazy(&files.term_dict).await?;
1513 let term_dict = AsyncSSTableReader::open(term_dict_handle, term_cache_blocks).await?;
1514
1515 let postings_handle = dir.open_lazy(&files.postings).await?;
1517
1518 let store_handle = dir.open_lazy(&files.store).await?;
1520 let store = AsyncStoreReader::open(
1521 store_handle,
1522 store_cache_directory_namespace,
1523 segment_id.0,
1524 store_cache,
1525 )
1526 .await?;
1527
1528 let vectors_data = loader::load_vectors_file(dir, &files, &schema, meta.num_docs).await?;
1530 let dense_file_backed_bytes = vectors_data.file_backed_bytes;
1531 let vector_indexes = vectors_data.indexes;
1532 let flat_vectors = vectors_data.flat_vectors;
1533
1534 #[cfg(feature = "native")]
1539 for (field_id, lazy_flat) in &flat_vectors {
1540 if vector_indexes.contains_key(field_id) {
1541 lazy_flat.advise_random_access();
1542 }
1543 }
1544
1545 let sparse_data = loader::load_sparse_file(dir, &files, meta.num_docs, &schema).await?;
1547 let sparse_file_backed_bytes = sparse_data.file_backed_bytes;
1548 let sparse_indexes = sparse_data.maxscore_indexes;
1549 let bmp_indexes = sparse_data.bmp_indexes;
1550
1551 let positions_handle = loader::open_positions_file(dir, &files, &schema).await?;
1553
1554 let fast_fields = loader::load_fast_fields_file(dir, &files, &schema).await?;
1556
1557 {
1559 let mut parts = vec![format!(
1560 "[segment] loaded {:016x}: docs={}",
1561 segment_id.0, meta.num_docs
1562 )];
1563 if !vector_indexes.is_empty() || !flat_vectors.is_empty() {
1564 parts.push(format!(
1565 "dense vectors: {} ANN + {} flat fields",
1566 vector_indexes.len(),
1567 flat_vectors.len()
1568 ));
1569 }
1570 for (field_id, idx) in &sparse_indexes {
1571 parts.push(format!(
1572 "sparse vector field {}: {} dims, ~{}",
1573 field_id,
1574 idx.num_dimensions(),
1575 crate::format_bytes(idx.num_dimensions() as u64 * 24)
1576 ));
1577 }
1578 for (field_id, idx) in &bmp_indexes {
1579 parts.push(format!(
1580 "bmp field {}: {} dims, {} blocks",
1581 field_id,
1582 idx.dims(),
1583 idx.num_blocks
1584 ));
1585 }
1586 if !fast_fields.is_empty() {
1587 parts.push(format!("fast: {} fields", fast_fields.len()));
1588 }
1589 log::debug!("{}", parts.join(", "));
1590 }
1591
1592 #[allow(unused_mut)]
1593 let mut reader = Self {
1594 meta,
1595 term_dict: Arc::new(term_dict),
1596 postings_handle,
1597 store: Arc::new(store),
1598 schema,
1599 vector_indexes,
1600 flat_vectors,
1601 dense_file_backed_bytes,
1602 trained_vectors: Arc::new(crate::segment::TrainedVectorStructures::default()),
1603 sparse_indexes,
1604 bmp_indexes,
1605 sparse_file_backed_bytes,
1606 positions_handle,
1607 fast_fields,
1608 #[cfg(feature = "native")]
1609 dense_pin_report: Default::default(),
1610 #[cfg(feature = "native")]
1611 sparse_pin_report: Default::default(),
1612 };
1613
1614 #[cfg(feature = "native")]
1616 reader.apply_pin_policy(&crate::segment::pin::pin_policy().to_owned());
1617
1618 for (&field_id, vector_index) in &reader.vector_indexes {
1623 match vector_index {
1624 VectorIndex::BinaryIvf(index) | VectorIndex::IvfTq { index, .. } => {
1625 index.get().report_health(
1626 reader.schema.index_label(),
1627 field_id,
1628 reader.meta.id,
1629 );
1630 }
1631 VectorIndex::Tq { .. } => {}
1634 }
1635 }
1636
1637 Ok(reader)
1638 }
1639
1640 pub fn ann_health(&self, field: Field) -> Option<crate::segment::ann_disk::AnnHealth> {
1644 match self.vector_indexes.get(&field.0)? {
1645 VectorIndex::BinaryIvf(index) | VectorIndex::IvfTq { index, .. } => {
1646 Some(index.get().health())
1647 }
1648 VectorIndex::Tq { .. } => None,
1649 }
1650 }
1651
1652 #[cfg(feature = "native")]
1662 pub(crate) fn apply_pin_policy(&mut self, policy: &crate::segment::pin::PinPolicy) {
1663 use crate::segment::pin::PinReport;
1664
1665 if !policy.is_enabled() {
1666 return;
1667 }
1668 let mut remaining = policy.budget_bytes;
1669 let mut dense_report = PinReport::default();
1670 let mut sparse_report = PinReport::default();
1671
1672 for index in self.vector_indexes.values_mut() {
1674 index.pin_lookup_directory(policy.mode, &mut remaining, &mut dense_report);
1675 }
1676 for bmp in self.bmp_indexes.values_mut() {
1678 bmp.pin_block_starts(policy.mode, &mut remaining, &mut sparse_report);
1679 }
1680 for sparse in self.sparse_indexes.values_mut() {
1682 sparse.pin_skip_section(policy.mode, &mut remaining, &mut sparse_report);
1683 }
1684 for flat in self.flat_vectors.values_mut() {
1686 flat.pin_doc_ids(policy.mode, &mut remaining, &mut dense_report);
1687 }
1688 for bmp in self.bmp_indexes.values_mut() {
1689 bmp.pin_doc_maps(policy.mode, &mut remaining, &mut sparse_report);
1690 }
1691 for bmp in self.bmp_indexes.values_mut() {
1693 bmp.pin_query_hierarchy(policy.mode, &mut remaining, &mut sparse_report);
1694 }
1695
1696 let report = PinReport {
1697 intended_bytes: dense_report
1698 .intended_bytes
1699 .saturating_add(sparse_report.intended_bytes),
1700 pinned_bytes: dense_report
1701 .pinned_bytes
1702 .saturating_add(sparse_report.pinned_bytes),
1703 skipped_budget_bytes: dense_report
1704 .skipped_budget_bytes
1705 .saturating_add(sparse_report.skipped_budget_bytes),
1706 failed_bytes: dense_report
1707 .failed_bytes
1708 .saturating_add(sparse_report.failed_bytes),
1709 heap_copy_bytes: dense_report
1710 .heap_copy_bytes
1711 .saturating_add(sparse_report.heap_copy_bytes),
1712 };
1713 if report.skipped_budget_bytes > 0 || report.failed_bytes > 0 {
1714 log::warn!(
1715 "[pin] index={} segment {:016x}: pinned {}/{} (budget skipped {}, mlock failed {}) — \
1716 raise HERMES_PIN_METADATA_BUDGET_MB or RLIMIT_MEMLOCK for full coverage",
1717 self.schema.index_label(),
1718 self.meta.id,
1719 crate::format_bytes(report.pinned_bytes),
1720 crate::format_bytes(report.intended_bytes),
1721 crate::format_bytes(report.skipped_budget_bytes),
1722 crate::format_bytes(report.failed_bytes),
1723 );
1724 } else if report.pinned_bytes > 0 {
1725 log::info!(
1726 "[pin] index={} segment {:016x}: pinned {} of hot metadata ({:?})",
1727 self.schema.index_label(),
1728 self.meta.id,
1729 crate::format_bytes(report.pinned_bytes),
1730 policy.mode,
1731 );
1732 }
1733 self.dense_pin_report = dense_report;
1734 self.sparse_pin_report = sparse_report;
1735 }
1736
1737 pub fn meta(&self) -> &SegmentMeta {
1743 &self.meta
1744 }
1745
1746 pub fn num_docs(&self) -> u32 {
1747 self.meta.num_docs
1748 }
1749
1750 pub fn avg_field_len(&self, field: Field) -> f32 {
1752 self.meta.avg_field_len(field)
1753 }
1754
1755 pub fn schema(&self) -> &Schema {
1756 &self.schema
1757 }
1758
1759 pub fn sparse_indexes(&self) -> &FxHashMap<u32, SparseIndex> {
1761 &self.sparse_indexes
1762 }
1763
1764 pub fn sparse_index(&self, field: Field) -> Option<&SparseIndex> {
1766 self.sparse_indexes.get(&field.0)
1767 }
1768
1769 pub fn bmp_index(&self, field: Field) -> Option<&BmpIndex> {
1771 self.bmp_indexes.get(&field.0)
1772 }
1773
1774 pub fn bmp_indexes(&self) -> &FxHashMap<u32, BmpIndex> {
1776 &self.bmp_indexes
1777 }
1778
1779 pub fn vector_indexes(&self) -> &FxHashMap<u32, VectorIndex> {
1781 &self.vector_indexes
1782 }
1783
1784 pub fn flat_vectors(&self) -> &FxHashMap<u32, LazyFlatVectorData> {
1786 &self.flat_vectors
1787 }
1788
1789 pub fn fast_field(
1791 &self,
1792 field_id: u32,
1793 ) -> Option<&crate::structures::fast_field::FastFieldReader> {
1794 self.fast_fields.get(&field_id)
1795 }
1796
1797 pub fn fast_fields(&self) -> &FxHashMap<u32, crate::structures::fast_field::FastFieldReader> {
1799 &self.fast_fields
1800 }
1801
1802 pub fn term_dict_stats(&self) -> SSTableStats {
1804 self.term_dict.stats()
1805 }
1806
1807 pub fn memory_stats(&self) -> SegmentMemoryStats {
1809 let term_dict_stats = self.term_dict.stats();
1810
1811 let term_dict_cache_bytes = self.term_dict.cached_bytes();
1815 let store_cache_bytes = self.store.cached_bytes();
1816
1817 let sparse_heap_bytes: usize = self
1820 .sparse_indexes
1821 .values()
1822 .map(|s| s.estimated_heap_bytes())
1823 .sum::<usize>()
1824 + self
1825 .bmp_indexes
1826 .values()
1827 .map(|b| b.estimated_heap_bytes())
1828 .sum::<usize>();
1829
1830 let dense_heap_bytes: usize = self
1833 .vector_indexes
1834 .values()
1835 .map(|v| v.estimated_heap_bytes())
1836 .sum::<usize>()
1837 + self
1838 .flat_vectors
1839 .values()
1840 .map(LazyFlatVectorData::estimated_heap_bytes)
1841 .sum::<usize>();
1842
1843 #[cfg(feature = "native")]
1844 let (sparse_heap_bytes, dense_heap_bytes) = (
1845 sparse_heap_bytes.saturating_add(
1846 usize::try_from(self.sparse_pin_report.heap_copy_bytes).unwrap_or(usize::MAX),
1847 ),
1848 dense_heap_bytes.saturating_add(
1849 usize::try_from(self.dense_pin_report.heap_copy_bytes).unwrap_or(usize::MAX),
1850 ),
1851 );
1852
1853 #[cfg(feature = "native")]
1854 let (
1855 sparse_pinned_metadata_bytes,
1856 sparse_pin_intended_bytes,
1857 dense_pinned_metadata_bytes,
1858 dense_pin_intended_bytes,
1859 ) = (
1860 self.sparse_pin_report.pinned_bytes,
1861 self.sparse_pin_report.intended_bytes,
1862 self.dense_pin_report.pinned_bytes,
1863 self.dense_pin_report.intended_bytes,
1864 );
1865 #[cfg(not(feature = "native"))]
1866 let (
1867 sparse_pinned_metadata_bytes,
1868 sparse_pin_intended_bytes,
1869 dense_pinned_metadata_bytes,
1870 dense_pin_intended_bytes,
1871 ) = (0u64, 0u64, 0u64, 0u64);
1872
1873 let pinned_metadata_bytes =
1874 sparse_pinned_metadata_bytes.saturating_add(dense_pinned_metadata_bytes);
1875 let pin_intended_bytes = sparse_pin_intended_bytes.saturating_add(dense_pin_intended_bytes);
1876
1877 SegmentMemoryStats {
1878 segment_id: self.meta.id,
1879 num_docs: self.meta.num_docs,
1880 term_dict_cache_bytes,
1881 store_cache_bytes,
1882 sparse_heap_bytes,
1883 dense_heap_bytes,
1884 term_bloom_file_bytes: term_dict_stats.bloom_filter_size as u64,
1885 sparse_file_backed_bytes: self.sparse_file_backed_bytes,
1886 dense_file_backed_bytes: self.dense_file_backed_bytes,
1887 pinned_metadata_bytes,
1888 pin_intended_bytes,
1889 sparse_pinned_metadata_bytes,
1890 sparse_pin_intended_bytes,
1891 dense_pinned_metadata_bytes,
1892 dense_pin_intended_bytes,
1893 }
1894 }
1895
1896 pub async fn get_postings(
1901 &self,
1902 field: Field,
1903 term: &[u8],
1904 ) -> Result<Option<BlockPostingList>> {
1905 log::debug!(
1906 "SegmentReader::get_postings field={} term_len={}",
1907 field.0,
1908 term.len()
1909 );
1910
1911 let mut key = Vec::with_capacity(4 + term.len());
1913 key.extend_from_slice(&field.0.to_le_bytes());
1914 key.extend_from_slice(term);
1915
1916 let term_info = match self.term_dict.get(&key).await? {
1918 Some(info) => {
1919 log::debug!("SegmentReader::get_postings found term_info");
1920 info
1921 }
1922 None => {
1923 log::debug!("SegmentReader::get_postings term not found");
1924 return Ok(None);
1925 }
1926 };
1927
1928 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1930 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1932 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1933 posting_list.push(doc_id, tf);
1934 }
1935 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
1936 return Ok(Some(block_list));
1937 }
1938
1939 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
1941 Error::Corruption("TermInfo has neither inline nor external data".to_string())
1942 })?;
1943
1944 let range = checked_file_range(
1945 posting_offset,
1946 posting_len,
1947 self.postings_handle.len(),
1948 "posting",
1949 )?;
1950 let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
1951 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
1952
1953 Ok(Some(block_list))
1954 }
1955
1956 pub async fn get_prefix_postings(
1958 &self,
1959 field: Field,
1960 prefix: &[u8],
1961 ) -> Result<Vec<BlockPostingList>> {
1962 if prefix.is_empty() {
1963 return Err(Error::Query("prefix must not be empty".into()));
1964 }
1965 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
1967 key_prefix.extend_from_slice(&field.0.to_le_bytes());
1968 key_prefix.extend_from_slice(prefix);
1969
1970 let (entries, truncated) = self
1971 .term_dict
1972 .prefix_scan_limited(&key_prefix, MAX_PREFIX_TERMS)
1973 .await?;
1974 if truncated {
1975 return Err(Error::Query(format!(
1976 "prefix expands to more than {MAX_PREFIX_TERMS} terms"
1977 )));
1978 }
1979 let posting_count: u64 = entries
1980 .iter()
1981 .map(|(_, term_info)| term_info.doc_freq() as u64)
1982 .sum();
1983 if posting_count > MAX_PREFIX_POSTINGS {
1984 return Err(Error::Query(format!(
1985 "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
1986 )));
1987 }
1988 let mut results = Vec::with_capacity(entries.len());
1989
1990 for (_key, term_info) in entries {
1991 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1992 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1993 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1994 posting_list.push(doc_id, tf);
1995 }
1996 results.push(BlockPostingList::from_posting_list(&posting_list)?);
1997 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
1998 let range = checked_file_range(
1999 posting_offset,
2000 posting_len,
2001 self.postings_handle.len(),
2002 "prefix posting",
2003 )?;
2004 let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
2005 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
2006 }
2007 }
2008
2009 Ok(results)
2010 }
2011
2012 pub async fn doc(&self, local_doc_id: DocId) -> Result<Option<Document>> {
2017 self.doc_with_fields(local_doc_id, None).await
2018 }
2019
2020 pub async fn doc_with_fields(
2026 &self,
2027 local_doc_id: DocId,
2028 fields: Option<&rustc_hash::FxHashSet<u32>>,
2029 ) -> Result<Option<Document>> {
2030 let mut doc = match fields {
2031 Some(set) => {
2032 let field_ids: Vec<u32> = set.iter().copied().collect();
2033 match self
2034 .store
2035 .get_fields(local_doc_id, &self.schema, &field_ids)
2036 .await
2037 {
2038 Ok(Some(d)) => d,
2039 Ok(None) => return Ok(None),
2040 Err(e) => return Err(Error::from(e)),
2041 }
2042 }
2043 None => match self.store.get(local_doc_id, &self.schema).await {
2044 Ok(Some(d)) => d,
2045 Ok(None) => return Ok(None),
2046 Err(e) => return Err(Error::from(e)),
2047 },
2048 };
2049
2050 for (&field_id, lazy_flat) in &self.flat_vectors {
2052 if let Some(set) = fields
2054 && !set.contains(&field_id)
2055 {
2056 continue;
2057 }
2058
2059 let is_binary = lazy_flat.quantization == DenseVectorQuantization::Binary;
2060 let (start, entries) = lazy_flat.flat_indexes_for_doc(local_doc_id);
2061 for (j, &(_doc_id, _ordinal)) in entries.iter().enumerate() {
2062 let flat_idx = start + j;
2063 if is_binary {
2064 let vbs = lazy_flat.vector_byte_size();
2065 let mut raw = vec![0u8; vbs];
2066 match lazy_flat.read_vector_raw_into(flat_idx, &mut raw).await {
2067 Ok(()) => {
2068 doc.add_binary_dense_vector(Field(field_id), raw);
2069 }
2070 Err(e) => {
2071 log::warn!(
2072 "Failed to hydrate binary dense vector field {}: {}",
2073 field_id,
2074 e
2075 );
2076 }
2077 }
2078 } else {
2079 match lazy_flat.get_vector(flat_idx).await {
2080 Ok(vec) => {
2081 doc.add_dense_vector(Field(field_id), vec);
2082 }
2083 Err(e) => {
2084 log::warn!("Failed to hydrate dense vector field {}: {}", field_id, e);
2085 }
2086 }
2087 }
2088 }
2089 }
2090
2091 Ok(Some(doc))
2092 }
2093
2094 pub async fn prefetch_terms(
2096 &self,
2097 field: Field,
2098 start_term: &[u8],
2099 end_term: &[u8],
2100 ) -> Result<()> {
2101 let mut start_key = Vec::with_capacity(4 + start_term.len());
2102 start_key.extend_from_slice(&field.0.to_le_bytes());
2103 start_key.extend_from_slice(start_term);
2104
2105 let mut end_key = Vec::with_capacity(4 + end_term.len());
2106 end_key.extend_from_slice(&field.0.to_le_bytes());
2107 end_key.extend_from_slice(end_term);
2108
2109 self.term_dict.prefetch_range(&start_key, &end_key).await?;
2110 Ok(())
2111 }
2112
2113 pub fn store_has_dict(&self) -> bool {
2115 self.store.has_dict()
2116 }
2117
2118 pub fn store(&self) -> &super::store::AsyncStoreReader {
2120 &self.store
2121 }
2122
2123 pub fn store_raw_blocks(&self) -> Vec<RawStoreBlock> {
2125 self.store.raw_blocks()
2126 }
2127
2128 pub fn store_data_slice(&self) -> &FileHandle {
2130 self.store.data_slice()
2131 }
2132
2133 pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
2135 self.term_dict.all_entries().await.map_err(Error::from)
2136 }
2137
2138 pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
2143 let entries = self.term_dict.all_entries().await?;
2144 let mut result = Vec::with_capacity(entries.len());
2145
2146 for (key, term_info) in entries {
2147 if key.len() > 4 {
2149 let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
2150 let term_bytes = &key[4..];
2151 if let Ok(term_str) = std::str::from_utf8(term_bytes) {
2152 result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
2153 }
2154 }
2155 }
2156
2157 Ok(result)
2158 }
2159
2160 pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
2162 self.term_dict.iter()
2163 }
2164
2165 pub async fn prefetch_term_dict(&self) -> crate::Result<()> {
2169 self.term_dict
2170 .prefetch_all_data_bulk()
2171 .await
2172 .map_err(crate::Error::from)
2173 }
2174
2175 pub async fn read_postings(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
2177 let range = checked_file_range(offset, len, self.postings_handle.len(), "posting")?;
2178 let bytes = self.postings_handle.read_bytes_range(range).await?;
2179 Ok(bytes.to_vec())
2180 }
2181
2182 pub async fn read_position_bytes(&self, offset: u64, len: u64) -> Result<Option<Vec<u8>>> {
2184 let handle = match &self.positions_handle {
2185 Some(h) => h,
2186 None => return Ok(None),
2187 };
2188 let range = checked_file_range(offset, len, handle.len(), "position")?;
2189 let bytes = handle.read_bytes_range(range).await?;
2190 Ok(Some(bytes.to_vec()))
2191 }
2192
2193 pub fn has_positions_file(&self) -> bool {
2195 self.positions_handle.is_some()
2196 }
2197
2198 fn validate_dense_search_request(
2202 &self,
2203 field: Field,
2204 query: &[f32],
2205 nprobe: usize,
2206 rerank_factor: f32,
2207 combiner: crate::query::MultiValueCombiner,
2208 ) -> Result<DenseSearchParams> {
2209 let entry = self
2210 .schema
2211 .get_field_entry(field)
2212 .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
2213 if entry.field_type != crate::dsl::FieldType::DenseVector {
2214 return Err(Error::InvalidFieldType {
2215 expected: "dense_vector".to_string(),
2216 got: format!("{:?}", entry.field_type),
2217 });
2218 }
2219 let config = entry.dense_vector_config.as_ref().ok_or_else(|| {
2220 Error::Schema(format!(
2221 "dense vector field '{}' has no dense vector configuration",
2222 entry.name
2223 ))
2224 })?;
2225
2226 if query.is_empty() {
2227 return Err(Error::Query(format!(
2228 "dense query vector for field '{}' must not be empty",
2229 entry.name
2230 )));
2231 }
2232 if query.len() != config.dim {
2233 return Err(Error::Query(format!(
2234 "dense query vector dimension {} does not match field '{}' dimension {}",
2235 query.len(),
2236 entry.name,
2237 config.dim
2238 )));
2239 }
2240 if let Some((index, value)) = query
2241 .iter()
2242 .enumerate()
2243 .find(|(_, value)| !value.is_finite())
2244 {
2245 return Err(Error::Query(format!(
2246 "dense query vector for field '{}' contains non-finite value {value} at index {index}",
2247 entry.name
2248 )));
2249 }
2250
2251 let nprobe = match (nprobe, config.nprobe) {
2254 (0, 0) => 32,
2255 (0, schema_nprobe) => schema_nprobe,
2256 (query_nprobe, _) => query_nprobe,
2257 };
2258 if nprobe > MAX_DENSE_NPROBE {
2259 return Err(Error::Query(format!(
2260 "dense nprobe must be at most {MAX_DENSE_NPROBE}, got {nprobe}"
2261 )));
2262 }
2263
2264 checked_dense_fetch_k(0, rerank_factor)?;
2267 combiner.validate().map_err(Error::Query)?;
2268
2269 Ok(DenseSearchParams {
2270 dim: config.dim,
2271 nprobe,
2272 unit_norm: config.unit_norm,
2273 })
2274 }
2275
2276 fn validate_binary_search_request(&self, field: Field, query: &[u8]) -> Result<usize> {
2277 let entry = self
2278 .schema
2279 .get_field_entry(field)
2280 .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
2281 if entry.field_type != crate::dsl::FieldType::BinaryDenseVector {
2282 return Err(Error::InvalidFieldType {
2283 expected: "binary_dense_vector".to_string(),
2284 got: format!("{:?}", entry.field_type),
2285 });
2286 }
2287 let config = entry.binary_dense_vector_config.as_ref().ok_or_else(|| {
2288 Error::Schema(format!(
2289 "binary dense vector field '{}' has no configuration",
2290 entry.name
2291 ))
2292 })?;
2293 if config.dim == 0 || !config.dim.is_multiple_of(8) {
2294 return Err(Error::Schema(format!(
2295 "binary dense vector field '{}' has invalid dimension {}",
2296 entry.name, config.dim
2297 )));
2298 }
2299 if query.len() != config.byte_len() {
2300 return Err(Error::Query(format!(
2301 "binary query byte length {} does not match field '{}' byte length {}",
2302 query.len(),
2303 entry.name,
2304 config.byte_len()
2305 )));
2306 }
2307 if query.iter().all(|&byte| byte == 0) {
2308 return Err(Error::Query(format!(
2313 "binary query for field '{}' is all-zero: it carries no information and would \
2314 rank candidates by bit count rather than similarity",
2315 entry.name,
2316 )));
2317 }
2318 Ok(config.dim)
2319 }
2320
2321 #[cfg(test)]
2323 fn score_quantized_batch_legacy(
2324 query: &[f32],
2325 raw: &[u8],
2326 quant: crate::dsl::DenseVectorQuantization,
2327 dim: usize,
2328 scores: &mut [f32],
2329 unit_norm: bool,
2330 ) -> Result<()> {
2331 use crate::dsl::DenseVectorQuantization;
2332 use crate::structures::simd;
2333
2334 if query.len() != dim {
2335 return Err(Error::Query(format!(
2336 "dense SIMD query dimension {} does not match vector dimension {dim}",
2337 query.len()
2338 )));
2339 }
2340 let element_size = match quant {
2341 DenseVectorQuantization::F32 => std::mem::size_of::<f32>(),
2342 DenseVectorQuantization::F16 => std::mem::size_of::<u16>(),
2343 DenseVectorQuantization::UInt8 => 1,
2344 DenseVectorQuantization::Binary => {
2345 return Err(Error::InvalidFieldType {
2346 expected: "non-binary dense vector".to_string(),
2347 got: "binary dense vector".to_string(),
2348 });
2349 }
2350 };
2351 let required_bytes = scores
2352 .len()
2353 .checked_mul(dim)
2354 .and_then(|elements| elements.checked_mul(element_size))
2355 .ok_or_else(|| Error::Corruption("dense vector batch byte length overflow".into()))?;
2356 if raw.len() < required_bytes {
2357 return Err(Error::Corruption(format!(
2358 "dense vector batch is truncated: need {required_bytes} bytes, got {}",
2359 raw.len()
2360 )));
2361 }
2362 if quant == DenseVectorQuantization::F16
2363 && required_bytes > 0
2364 && !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<u16>())
2365 {
2366 return Err(Error::Corruption(
2367 "f16 vector data is not 2-byte aligned".to_string(),
2368 ));
2369 }
2370
2371 match (quant, unit_norm) {
2372 (DenseVectorQuantization::F32, false) => {
2373 let num_floats = scores.len() * dim;
2374 if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
2375 return Err(Error::Corruption(
2376 "f32 vector data is not 4-byte aligned".to_string(),
2377 ));
2378 }
2379 let vectors: &[f32] =
2380 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
2381 simd::batch_cosine_scores(query, vectors, dim, scores);
2382 }
2383 (DenseVectorQuantization::F32, true) => {
2384 let num_floats = scores.len() * dim;
2385 if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
2386 return Err(Error::Corruption(
2387 "f32 vector data is not 4-byte aligned".to_string(),
2388 ));
2389 }
2390 let vectors: &[f32] =
2391 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
2392 simd::batch_dot_scores(query, vectors, dim, scores);
2393 }
2394 (DenseVectorQuantization::F16, false) => {
2395 simd::batch_cosine_scores_f16(query, raw, dim, scores);
2396 }
2397 (DenseVectorQuantization::F16, true) => {
2398 simd::batch_dot_scores_f16(query, raw, dim, scores);
2399 }
2400 (DenseVectorQuantization::UInt8, false) => {
2401 simd::batch_cosine_scores_u8(query, raw, dim, scores);
2402 }
2403 (DenseVectorQuantization::UInt8, true) => {
2404 simd::batch_dot_scores_u8(query, raw, dim, scores);
2405 }
2406 (DenseVectorQuantization::Binary, _) => unreachable!("validated above"),
2407 }
2408 Ok(())
2409 }
2410
2411 pub async fn search_dense_vector(
2417 &self,
2418 field: Field,
2419 query: &[f32],
2420 k: usize,
2421 nprobe: usize,
2422 rerank_factor: f32,
2423 combiner: crate::query::MultiValueCombiner,
2424 ) -> Result<Vec<VectorSearchResult>> {
2425 self.search_dense_vector_impl(field, query, k, nprobe, rerank_factor, combiner, None)
2426 .await
2427 }
2428
2429 #[allow(clippy::too_many_arguments)]
2430 pub(crate) async fn search_dense_vector_with_probe_cache(
2431 &self,
2432 field: Field,
2433 query: &[f32],
2434 k: usize,
2435 nprobe: usize,
2436 rerank_factor: f32,
2437 combiner: crate::query::MultiValueCombiner,
2438 plan_cache: &DensePlanCache,
2439 ) -> Result<Vec<VectorSearchResult>> {
2440 self.search_dense_vector_impl(
2441 field,
2442 query,
2443 k,
2444 nprobe,
2445 rerank_factor,
2446 combiner,
2447 Some(plan_cache),
2448 )
2449 .await
2450 }
2451
2452 #[allow(clippy::too_many_arguments)]
2453 async fn search_dense_vector_impl(
2454 &self,
2455 field: Field,
2456 query: &[f32],
2457 k: usize,
2458 nprobe: usize,
2459 rerank_factor: f32,
2460 combiner: crate::query::MultiValueCombiner,
2461 plan_cache: Option<&DensePlanCache>,
2462 ) -> Result<Vec<VectorSearchResult>> {
2463 let params =
2464 self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
2465 let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
2466 if k == 0 {
2467 return Ok(Vec::new());
2468 }
2469
2470 let configured_ann_index = self.vector_indexes.get(&field.0);
2471 let lazy_flat = self.flat_vectors.get(&field.0);
2472 if configured_ann_index.is_none() && lazy_flat.is_none() {
2474 return Ok(Vec::new());
2475 }
2476
2477 if configured_ann_index.is_some() && lazy_flat.is_none() {
2478 return Err(Error::Corruption(format!(
2479 "dense ANN field {} is missing flat vector storage",
2480 field.0
2481 )));
2482 }
2483
2484 if let Some(flat) = lazy_flat
2485 && flat.dim != params.dim
2486 {
2487 return Err(Error::Corruption(format!(
2488 "dense vector field {} has schema dimension {} but flat storage dimension {}",
2489 field.0, params.dim, flat.dim
2490 )));
2491 }
2492
2493 let needs_document_aggregation = lazy_flat.is_some_and(|flat| {
2494 flat.num_vectors != flat.num_docs_with_vectors()
2495 && !matches!(combiner, crate::query::MultiValueCombiner::Max)
2496 });
2497 let ann_index = configured_ann_index;
2501
2502 let t0 = std::time::Instant::now();
2504 let mut flat_results = None;
2505 let results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
2506 match index {
2508 VectorIndex::Tq { index: lazy, codec } => {
2509 let flat = lazy_flat.expect("ANN/flat pairing validated above");
2510 search_tq_segment(
2512 lazy.get(),
2513 codec,
2514 query,
2515 fetch_k.min(flat.num_docs_with_vectors()),
2516 needs_document_aggregation.then_some(combiner),
2517 field,
2518 params.dim,
2519 plan_cache.map(|cache| &cache.tq),
2520 )?
2521 }
2522 VectorIndex::IvfTq { index: lazy, codec } => {
2523 let index = lazy.get();
2524 let centroids =
2525 self.trained_vectors
2526 .centroids
2527 .get(&field.0)
2528 .ok_or_else(|| {
2529 Error::Schema(format!(
2530 "IVF-TQ index requires coarse centroids for field {}",
2531 field.0
2532 ))
2533 })?;
2534 validate_coarse_centroids(centroids, params.dim)?;
2535 let routing = self
2536 .schema
2537 .get_field_entry(field)
2538 .and_then(|entry| entry.dense_vector_config.as_ref())
2539 .map_or(crate::dsl::IvfRoutingMode::Auto, |config| {
2540 config.ivf_routing
2541 });
2542 validate_ivf_tq_ann(index, centroids, codec, params.dim, routing, field)?;
2543 let flat = lazy_flat.expect("ANN/flat pairing validated above");
2544 search_ivf_tq_segment(
2545 index,
2546 centroids,
2547 codec,
2548 query,
2549 fetch_k.min(flat.num_docs_with_vectors()),
2550 needs_document_aggregation.then_some(combiner),
2551 field,
2552 params.nprobe,
2553 routing,
2554 plan_cache.map(|cache| &cache.ivf_tq),
2555 )?
2556 }
2557 VectorIndex::BinaryIvf(_) => {
2558 Vec::new()
2560 }
2561 }
2562 } else if let Some(lazy_flat) = lazy_flat {
2563 log::debug!(
2567 "[dense_vector_search] index={} field {}: brute-force on {} vectors (dim={}, quant={:?})",
2568 self.schema.index_label(),
2569 field.0,
2570 lazy_flat.num_vectors,
2571 lazy_flat.dim,
2572 lazy_flat.quantization
2573 );
2574 let dim = lazy_flat.dim;
2575 let n = lazy_flat.num_vectors;
2576 let quant = lazy_flat.quantization;
2577 let batch_len =
2578 bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
2579 let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
2580 let mut scores = vec![0f32; batch_len];
2581 let prepared_query = PreparedDenseScoreQuery::new(query, quant, dim, params.unit_norm)?;
2582
2583 for batch_start in (0..n).step_by(batch_len) {
2584 let batch_count = batch_len.min(n - batch_start);
2585 let batch_bytes = lazy_flat
2586 .read_vectors_batch(batch_start, batch_count)
2587 .await
2588 .map_err(crate::Error::Io)?;
2589 let raw = batch_bytes.as_slice();
2590
2591 prepared_query.score_batch(raw, &mut scores[..batch_count])?;
2592
2593 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2594 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2595 collector.push(doc_id, ordinal, score);
2596 }
2597 }
2598
2599 flat_results = Some(collector.into_results());
2600 Vec::new()
2601 } else {
2602 return Ok(Vec::new());
2603 };
2604 let l1_elapsed = t0.elapsed();
2605 {
2606 let kind = match ann_index {
2607 Some(VectorIndex::BinaryIvf(_)) => "binary_ivf",
2608 Some(VectorIndex::Tq { .. }) => "tq_flat",
2609 Some(VectorIndex::IvfTq { .. }) => "ivf_tq",
2610 None => "flat",
2611 };
2612 crate::observe::dense_l1(
2613 self.schema.index_label(),
2614 self.schema.get_field_name(field).unwrap_or("?"),
2615 kind,
2616 l1_elapsed.as_secs_f64(),
2617 flat_results.as_ref().map_or(results.len(), Vec::len),
2618 );
2619 }
2620 log::debug!(
2621 "[dense_vector_search] index={} field {}: L1 returned {} candidates in {:.1}ms",
2622 self.schema.index_label(),
2623 field.0,
2624 flat_results.as_ref().map_or(results.len(), Vec::len),
2625 l1_elapsed.as_secs_f64() * 1000.0
2626 );
2627
2628 if let Some(results) = flat_results {
2629 return Ok(results);
2630 }
2631
2632 if ann_index.is_some()
2635 && !results.is_empty()
2636 && let Some(lazy_flat) = lazy_flat
2637 {
2638 let t_rerank = std::time::Instant::now();
2639 let vbs = lazy_flat.vector_byte_size();
2640 let (reranked, stats) = exact_score_dense_candidate_documents(
2641 &results,
2642 lazy_flat,
2643 query,
2644 params.unit_norm,
2645 combiner,
2646 k,
2647 )
2648 .await?;
2649
2650 crate::observe::dense_rerank(
2651 self.schema.index_label(),
2652 self.schema.get_field_name(field).unwrap_or("?"),
2653 t_rerank.elapsed().as_secs_f64(),
2654 stats.resolve_elapsed.as_secs_f64(),
2655 stats.read_elapsed.as_secs_f64(),
2656 stats.vector_count,
2657 );
2658 log::debug!(
2659 "[dense_vector_search] index={} field {}: rerank {} vectors (dim={}, quant={:?}, bytes_per_vector={}): resolve={:.1}ms read={:.1}ms score={:.1}ms",
2660 self.schema.index_label(),
2661 field.0,
2662 stats.vector_count,
2663 lazy_flat.dim,
2664 lazy_flat.quantization,
2665 vbs,
2666 stats.resolve_elapsed.as_secs_f64() * 1000.0,
2667 stats.read_elapsed.as_secs_f64() * 1000.0,
2668 stats.score_elapsed.as_secs_f64() * 1000.0,
2669 );
2670
2671 log::debug!(
2672 "[dense_vector_search] index={} field {}: rerank total={:.1}ms",
2673 self.schema.index_label(),
2674 field.0,
2675 t_rerank.elapsed().as_secs_f64() * 1000.0
2676 );
2677 return Ok(reranked);
2678 }
2679
2680 Ok(combine_grouped_ordinal_results(results, combiner, k))
2681 }
2682
2683 async fn search_binary_dense_vector_impl(
2688 &self,
2689 field: Field,
2690 query: &[u8],
2691 k: usize,
2692 combiner: crate::query::MultiValueCombiner,
2693 probe_cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
2694 ) -> Result<Vec<VectorSearchResult>> {
2695 let schema_dim = self.validate_binary_search_request(field, query)?;
2696 combiner.validate().map_err(Error::Query)?;
2697 if k == 0 {
2698 return Ok(Vec::new());
2699 }
2700 let t0 = crate::observe::Timer::start();
2701 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0) {
2702 let ivf = lazy.get();
2703 let config = self
2704 .schema
2705 .get_field_entry(field)
2706 .and_then(|entry| entry.binary_dense_vector_config.as_ref())
2707 .ok_or_else(|| {
2708 Error::Schema(format!(
2709 "binary IVF field {} has no schema configuration",
2710 field.0
2711 ))
2712 })?;
2713 let quantizer = self
2714 .trained_vectors
2715 .binary_quantizers
2716 .get(&field.0)
2717 .ok_or_else(|| {
2718 Error::Schema(format!(
2719 "global binary IVF field {} has no loaded quantizer",
2720 field.0
2721 ))
2722 })?;
2723 validate_binary_ann(ivf, quantizer, config, schema_dim, field)?;
2724 let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
2725 Error::Corruption(format!(
2726 "global binary IVF field {} is missing flat vector storage",
2727 field.0
2728 ))
2729 })?;
2730 let single_valued = flat.num_vectors == flat.num_docs_with_vectors();
2731 let clusters = binary_probe_clusters(
2732 quantizer,
2733 query,
2734 config.nprobe,
2735 config.ivf_routing,
2736 probe_cache,
2737 )?;
2738 let results = if !single_valued
2739 && !matches!(combiner, crate::query::MultiValueCombiner::Max)
2740 {
2741 let candidate_limit =
2742 checked_binary_combined_fetch_k(k)?.min(flat.num_docs_with_vectors());
2743 let (candidate_documents, probed_ordinal_scores) = ivf
2744 .search_binary_combined_documents(candidate_limit, query, &clusters, combiner)
2745 .map_err(|error| {
2746 Error::Corruption(format!(
2747 "invalid binary IVF payload for field {}: {error}",
2748 field.0,
2749 ))
2750 })?;
2751 exact_score_binary_candidate_document_ids(
2752 candidate_documents
2753 .into_iter()
2754 .map(|candidate| candidate.doc_id)
2755 .collect(),
2756 &probed_ordinal_scores,
2757 flat,
2758 query,
2759 schema_dim,
2760 combiner,
2761 k,
2762 )
2763 .await?
2764 } else {
2765 let candidate_docs = if single_valued {
2766 k
2767 } else {
2768 checked_binary_combined_fetch_k(k)?
2773 }
2774 .min(flat.num_docs_with_vectors());
2775 let ann_results = if single_valued {
2776 ivf.search_binary_clusters::<false>(query, candidate_docs, &clusters)
2777 } else {
2778 ivf.search_binary_clusters::<true>(query, candidate_docs, &clusters)
2779 }
2780 .map_err(|error| {
2781 Error::Corruption(format!(
2782 "invalid binary IVF payload for field {}: {error}",
2783 field.0,
2784 ))
2785 })?;
2786 if single_valued {
2789 let ann_results = validate_binary_single_value_ann_results(ann_results, flat)?;
2790 combine_ordinal_results(ann_results, combiner, k)
2791 } else {
2792 exact_score_binary_candidate_documents(
2793 &ann_results,
2794 flat,
2795 query,
2796 schema_dim,
2797 combiner,
2798 k,
2799 )
2800 .await?
2801 }
2802 };
2803 crate::observe::dense_l1(
2804 self.schema.index_label(),
2805 self.schema.get_field_name(field).unwrap_or("?"),
2806 "global_binary_ivf",
2807 t0.secs(),
2808 results.len(),
2809 );
2810 return Ok(results);
2811 }
2812 let lazy_flat = match self.flat_vectors.get(&field.0) {
2813 Some(f) => f,
2814 None => return Ok(Vec::new()),
2815 };
2816
2817 let dim_bits = lazy_flat.dim;
2818 let byte_len = lazy_flat.vector_byte_size();
2819 let n = lazy_flat.num_vectors;
2820
2821 if dim_bits != schema_dim {
2822 return Err(Error::Corruption(format!(
2823 "binary vector field {} has schema dimension {} but flat storage dimension {}",
2824 field.0, schema_dim, dim_bits
2825 )));
2826 }
2827
2828 if byte_len != query.len() {
2829 return Err(Error::Schema(format!(
2830 "Binary query vector byte length {} != field byte length {}",
2831 query.len(),
2832 byte_len
2833 )));
2834 }
2835
2836 let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
2837 let mut collector = FlatDocumentCollector::new(k, combiner);
2838 let mut scores = vec![0f32; batch_len];
2839
2840 for batch_start in (0..n).step_by(batch_len) {
2841 let batch_count = batch_len.min(n - batch_start);
2842 let batch_bytes = lazy_flat
2843 .read_vectors_batch(batch_start, batch_count)
2844 .await
2845 .map_err(crate::Error::Io)?;
2846 let raw = batch_bytes.as_slice();
2847
2848 crate::structures::simd::batch_hamming_scores(
2849 query,
2850 raw,
2851 byte_len,
2852 dim_bits,
2853 &mut scores[..batch_count],
2854 );
2855
2856 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2857 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2858 collector.push(doc_id, ordinal, score);
2859 }
2860 }
2861
2862 let results = collector.into_results();
2863
2864 crate::observe::dense_l1(
2865 self.schema.index_label(),
2866 self.schema.get_field_name(field).unwrap_or("?"),
2867 "binary_flat",
2868 t0.secs(),
2869 results.len(),
2870 );
2871 Ok(results)
2872 }
2873
2874 pub async fn search_binary_dense_vector(
2875 &self,
2876 field: Field,
2877 query: &[u8],
2878 k: usize,
2879 combiner: crate::query::MultiValueCombiner,
2880 ) -> Result<Vec<VectorSearchResult>> {
2881 self.search_binary_dense_vector_impl(field, query, k, combiner, None)
2882 .await
2883 }
2884
2885 pub(crate) async fn search_binary_dense_vector_with_probe_cache(
2886 &self,
2887 field: Field,
2888 query: &[u8],
2889 k: usize,
2890 combiner: crate::query::MultiValueCombiner,
2891 probe_cache: &std::sync::Mutex<Option<crate::structures::IvfProbePlan>>,
2892 ) -> Result<Vec<VectorSearchResult>> {
2893 self.search_binary_dense_vector_impl(field, query, k, combiner, Some(probe_cache))
2894 .await
2895 }
2896
2897 pub fn coarse_centroids(&self, field_id: u32) -> Option<&Arc<CoarseCentroids>> {
2899 self.trained_vectors.centroids.get(&field_id)
2900 }
2901
2902 pub fn set_trained_vectors(
2903 &mut self,
2904 trained_vectors: Arc<crate::segment::TrainedVectorStructures>,
2905 ) {
2906 self.trained_vectors = trained_vectors;
2907 }
2908
2909 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
2911 self.vector_indexes.get(&field.0)
2912 }
2913
2914 pub async fn get_positions(
2919 &self,
2920 field: Field,
2921 term: &[u8],
2922 ) -> Result<Option<crate::structures::PositionPostingList>> {
2923 let handle = match &self.positions_handle {
2925 Some(h) => h,
2926 None => return Ok(None),
2927 };
2928
2929 let mut key = Vec::with_capacity(4 + term.len());
2931 key.extend_from_slice(&field.0.to_le_bytes());
2932 key.extend_from_slice(term);
2933
2934 let term_info = match self.term_dict.get(&key).await? {
2936 Some(info) => info,
2937 None => return Ok(None),
2938 };
2939
2940 let (offset, length) = match term_info.position_info() {
2942 Some((o, l)) => (o, l),
2943 None => return Ok(None),
2944 };
2945
2946 let range = checked_file_range(offset, length, handle.len(), "position list")?;
2950 let slice = handle.slice(range);
2951 let data = slice.read_bytes().await?;
2952
2953 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2955
2956 Ok(Some(pos_list))
2957 }
2958
2959 pub fn has_positions(&self, field: Field) -> bool {
2961 if let Some(entry) = self.schema.get_field_entry(field) {
2963 entry.positions.is_some()
2964 } else {
2965 false
2966 }
2967 }
2968}
2969
2970#[cfg(feature = "sync")]
2972impl SegmentReader {
2973 pub fn get_postings_sync(&self, field: Field, term: &[u8]) -> Result<Option<BlockPostingList>> {
2975 let mut key = Vec::with_capacity(4 + term.len());
2977 key.extend_from_slice(&field.0.to_le_bytes());
2978 key.extend_from_slice(term);
2979
2980 let term_info = match self.term_dict.get_sync(&key)? {
2982 Some(info) => info,
2983 None => return Ok(None),
2984 };
2985
2986 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2988 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2989 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2990 posting_list.push(doc_id, tf);
2991 }
2992 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
2993 return Ok(Some(block_list));
2994 }
2995
2996 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
2998 Error::Corruption("TermInfo has neither inline nor external data".to_string())
2999 })?;
3000
3001 let range = checked_file_range(
3002 posting_offset,
3003 posting_len,
3004 self.postings_handle.len(),
3005 "posting",
3006 )?;
3007 let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
3008 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
3009
3010 Ok(Some(block_list))
3011 }
3012
3013 pub fn get_prefix_postings_sync(
3015 &self,
3016 field: Field,
3017 prefix: &[u8],
3018 ) -> Result<Vec<BlockPostingList>> {
3019 if prefix.is_empty() {
3020 return Err(Error::Query("prefix must not be empty".into()));
3021 }
3022 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
3023 key_prefix.extend_from_slice(&field.0.to_le_bytes());
3024 key_prefix.extend_from_slice(prefix);
3025
3026 let (entries, truncated) = self
3027 .term_dict
3028 .prefix_scan_limited_sync(&key_prefix, MAX_PREFIX_TERMS)?;
3029 if truncated {
3030 return Err(Error::Query(format!(
3031 "prefix expands to more than {MAX_PREFIX_TERMS} terms"
3032 )));
3033 }
3034 let posting_count: u64 = entries
3035 .iter()
3036 .map(|(_, term_info)| term_info.doc_freq() as u64)
3037 .sum();
3038 if posting_count > MAX_PREFIX_POSTINGS {
3039 return Err(Error::Query(format!(
3040 "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
3041 )));
3042 }
3043 let mut results = Vec::with_capacity(entries.len());
3044
3045 for (_key, term_info) in entries {
3046 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
3047 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
3048 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
3049 posting_list.push(doc_id, tf);
3050 }
3051 results.push(BlockPostingList::from_posting_list(&posting_list)?);
3052 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
3053 let range = checked_file_range(
3054 posting_offset,
3055 posting_len,
3056 self.postings_handle.len(),
3057 "prefix posting",
3058 )?;
3059 let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
3060 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
3061 }
3062 }
3063
3064 Ok(results)
3065 }
3066
3067 pub fn get_positions_sync(
3069 &self,
3070 field: Field,
3071 term: &[u8],
3072 ) -> Result<Option<crate::structures::PositionPostingList>> {
3073 let handle = match &self.positions_handle {
3074 Some(h) => h,
3075 None => return Ok(None),
3076 };
3077
3078 let mut key = Vec::with_capacity(4 + term.len());
3080 key.extend_from_slice(&field.0.to_le_bytes());
3081 key.extend_from_slice(term);
3082
3083 let term_info = match self.term_dict.get_sync(&key)? {
3085 Some(info) => info,
3086 None => return Ok(None),
3087 };
3088
3089 let (offset, length) = match term_info.position_info() {
3090 Some((o, l)) => (o, l),
3091 None => return Ok(None),
3092 };
3093
3094 let range = checked_file_range(offset, length, handle.len(), "position list")?;
3095 let slice = handle.slice(range);
3096 let data = slice.read_bytes_sync()?;
3097
3098 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
3099 Ok(Some(pos_list))
3100 }
3101
3102 pub fn search_dense_vector_sync(
3105 &self,
3106 field: Field,
3107 query: &[f32],
3108 k: usize,
3109 nprobe: usize,
3110 rerank_factor: f32,
3111 combiner: crate::query::MultiValueCombiner,
3112 ) -> Result<Vec<VectorSearchResult>> {
3113 self.search_dense_vector_sync_impl(field, query, k, nprobe, rerank_factor, combiner, None)
3114 }
3115
3116 #[cfg(feature = "sync")]
3117 #[allow(clippy::too_many_arguments)]
3118 pub(crate) fn search_dense_vector_sync_with_probe_cache(
3119 &self,
3120 field: Field,
3121 query: &[f32],
3122 k: usize,
3123 nprobe: usize,
3124 rerank_factor: f32,
3125 combiner: crate::query::MultiValueCombiner,
3126 plan_cache: &DensePlanCache,
3127 ) -> Result<Vec<VectorSearchResult>> {
3128 self.search_dense_vector_sync_impl(
3129 field,
3130 query,
3131 k,
3132 nprobe,
3133 rerank_factor,
3134 combiner,
3135 Some(plan_cache),
3136 )
3137 }
3138
3139 #[cfg(feature = "sync")]
3140 #[allow(clippy::too_many_arguments)]
3141 fn search_dense_vector_sync_impl(
3142 &self,
3143 field: Field,
3144 query: &[f32],
3145 k: usize,
3146 nprobe: usize,
3147 rerank_factor: f32,
3148 combiner: crate::query::MultiValueCombiner,
3149 plan_cache: Option<&DensePlanCache>,
3150 ) -> Result<Vec<VectorSearchResult>> {
3151 let params =
3152 self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
3153 let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
3154 if k == 0 {
3155 return Ok(Vec::new());
3156 }
3157
3158 let configured_ann_index = self.vector_indexes.get(&field.0);
3159 let lazy_flat = self.flat_vectors.get(&field.0);
3160 if configured_ann_index.is_none() && lazy_flat.is_none() {
3161 return Ok(Vec::new());
3162 }
3163
3164 if configured_ann_index.is_some() && lazy_flat.is_none() {
3165 return Err(Error::Corruption(format!(
3166 "dense ANN field {} is missing flat vector storage",
3167 field.0
3168 )));
3169 }
3170
3171 if let Some(flat) = lazy_flat
3172 && flat.dim != params.dim
3173 {
3174 return Err(Error::Corruption(format!(
3175 "dense vector field {} has schema dimension {} but flat storage dimension {}",
3176 field.0, params.dim, flat.dim
3177 )));
3178 }
3179
3180 let needs_document_aggregation = lazy_flat.is_some_and(|flat| {
3181 flat.num_vectors != flat.num_docs_with_vectors()
3182 && !matches!(combiner, crate::query::MultiValueCombiner::Max)
3183 });
3184 let ann_index = configured_ann_index;
3187
3188 let results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
3189 match index {
3191 VectorIndex::Tq { index: lazy, codec } => {
3192 let flat = lazy_flat.expect("ANN/flat pairing validated above");
3193 search_tq_segment(
3194 lazy.get(),
3195 codec,
3196 query,
3197 fetch_k.min(flat.num_docs_with_vectors()),
3198 needs_document_aggregation.then_some(combiner),
3199 field,
3200 params.dim,
3201 plan_cache.map(|cache| &cache.tq),
3202 )?
3203 }
3204 VectorIndex::IvfTq { index: lazy, codec } => {
3205 let index = lazy.get();
3206 let centroids =
3207 self.trained_vectors
3208 .centroids
3209 .get(&field.0)
3210 .ok_or_else(|| {
3211 Error::Schema(format!(
3212 "IVF-TQ index requires coarse centroids for field {}",
3213 field.0
3214 ))
3215 })?;
3216 validate_coarse_centroids(centroids, params.dim)?;
3217 let routing = self
3218 .schema
3219 .get_field_entry(field)
3220 .and_then(|entry| entry.dense_vector_config.as_ref())
3221 .map_or(crate::dsl::IvfRoutingMode::Auto, |config| {
3222 config.ivf_routing
3223 });
3224 validate_ivf_tq_ann(index, centroids, codec, params.dim, routing, field)?;
3225 let flat = lazy_flat.expect("ANN/flat pairing validated above");
3226 search_ivf_tq_segment(
3227 index,
3228 centroids,
3229 codec,
3230 query,
3231 fetch_k.min(flat.num_docs_with_vectors()),
3232 needs_document_aggregation.then_some(combiner),
3233 field,
3234 params.nprobe,
3235 routing,
3236 plan_cache.map(|cache| &cache.ivf_tq),
3237 )?
3238 }
3239 VectorIndex::BinaryIvf(_) => {
3240 Vec::new()
3242 }
3243 }
3244 } else if let Some(lazy_flat) = lazy_flat {
3245 let dim = lazy_flat.dim;
3247 let n = lazy_flat.num_vectors;
3248 let quant = lazy_flat.quantization;
3249 let batch_len =
3250 bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
3251 let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
3252 let mut scores = vec![0f32; batch_len];
3253 let prepared_query = PreparedDenseScoreQuery::new(query, quant, dim, params.unit_norm)?;
3254
3255 for batch_start in (0..n).step_by(batch_len) {
3256 let batch_count = batch_len.min(n - batch_start);
3257 let batch_bytes = lazy_flat
3258 .read_vectors_batch_sync(batch_start, batch_count)
3259 .map_err(crate::Error::Io)?;
3260 let raw = batch_bytes.as_slice();
3261
3262 prepared_query.score_batch(raw, &mut scores[..batch_count])?;
3263
3264 for (i, &score) in scores.iter().enumerate().take(batch_count) {
3265 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
3266 collector.push(doc_id, ordinal, score);
3267 }
3268 }
3269
3270 return Ok(collector.into_results());
3271 } else {
3272 return Ok(Vec::new());
3273 };
3274
3275 if ann_index.is_some()
3277 && !results.is_empty()
3278 && let Some(lazy_flat) = lazy_flat
3279 {
3280 return exact_score_dense_candidate_documents_sync(
3281 &results,
3282 lazy_flat,
3283 query,
3284 params.unit_norm,
3285 combiner,
3286 k,
3287 );
3288 }
3289
3290 Ok(combine_grouped_ordinal_results(results, combiner, k))
3291 }
3292
3293 #[cfg(feature = "sync")]
3298 fn search_binary_dense_vector_sync_impl(
3299 &self,
3300 field: Field,
3301 query: &[u8],
3302 k: usize,
3303 combiner: crate::query::MultiValueCombiner,
3304 probe_cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
3305 ) -> Result<Vec<VectorSearchResult>> {
3306 let schema_dim = self.validate_binary_search_request(field, query)?;
3307 combiner.validate().map_err(Error::Query)?;
3308 if k == 0 {
3309 return Ok(Vec::new());
3310 }
3311 let t0 = crate::observe::Timer::start();
3312 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0) {
3313 let ivf = lazy.get();
3314 let config = self
3315 .schema
3316 .get_field_entry(field)
3317 .and_then(|entry| entry.binary_dense_vector_config.as_ref())
3318 .ok_or_else(|| {
3319 Error::Schema(format!(
3320 "binary IVF field {} has no schema configuration",
3321 field.0
3322 ))
3323 })?;
3324 let quantizer = self
3325 .trained_vectors
3326 .binary_quantizers
3327 .get(&field.0)
3328 .ok_or_else(|| {
3329 Error::Schema(format!(
3330 "global binary IVF field {} has no loaded quantizer",
3331 field.0
3332 ))
3333 })?;
3334 validate_binary_ann(ivf, quantizer, config, schema_dim, field)?;
3335 let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
3336 Error::Corruption(format!(
3337 "global binary IVF field {} is missing flat vector storage",
3338 field.0
3339 ))
3340 })?;
3341 let single_valued = flat.num_vectors == flat.num_docs_with_vectors();
3342 let clusters = binary_probe_clusters(
3343 quantizer,
3344 query,
3345 config.nprobe,
3346 config.ivf_routing,
3347 probe_cache,
3348 )?;
3349 let results = if !single_valued
3350 && !matches!(combiner, crate::query::MultiValueCombiner::Max)
3351 {
3352 let candidate_limit =
3353 checked_binary_combined_fetch_k(k)?.min(flat.num_docs_with_vectors());
3354 let (candidate_documents, probed_ordinal_scores) = ivf
3355 .search_binary_combined_documents(candidate_limit, query, &clusters, combiner)
3356 .map_err(|error| {
3357 Error::Corruption(format!(
3358 "invalid binary IVF payload for field {}: {error}",
3359 field.0,
3360 ))
3361 })?;
3362 exact_score_binary_candidate_document_ids_sync(
3363 candidate_documents
3364 .into_iter()
3365 .map(|candidate| candidate.doc_id)
3366 .collect(),
3367 &probed_ordinal_scores,
3368 flat,
3369 query,
3370 schema_dim,
3371 combiner,
3372 k,
3373 )?
3374 } else {
3375 let candidate_docs = if single_valued {
3376 k
3377 } else {
3378 checked_binary_combined_fetch_k(k)?
3379 }
3380 .min(flat.num_docs_with_vectors());
3381 let ann_results = if single_valued {
3382 ivf.search_binary_clusters::<false>(query, candidate_docs, &clusters)
3383 } else {
3384 ivf.search_binary_clusters::<true>(query, candidate_docs, &clusters)
3385 }
3386 .map_err(|error| {
3387 Error::Corruption(format!(
3388 "invalid binary IVF payload for field {}: {error}",
3389 field.0,
3390 ))
3391 })?;
3392 if single_valued {
3393 let ann_results = validate_binary_single_value_ann_results(ann_results, flat)?;
3394 combine_ordinal_results(ann_results, combiner, k)
3395 } else {
3396 exact_score_binary_candidate_documents_sync(
3397 &ann_results,
3398 flat,
3399 query,
3400 schema_dim,
3401 combiner,
3402 k,
3403 )?
3404 }
3405 };
3406 crate::observe::dense_l1(
3407 self.schema.index_label(),
3408 self.schema.get_field_name(field).unwrap_or("?"),
3409 "global_binary_ivf",
3410 t0.secs(),
3411 results.len(),
3412 );
3413 return Ok(results);
3414 }
3415 let lazy_flat = match self.flat_vectors.get(&field.0) {
3416 Some(f) => f,
3417 None => return Ok(Vec::new()),
3418 };
3419
3420 let dim_bits = lazy_flat.dim;
3421 let byte_len = lazy_flat.vector_byte_size();
3422 let n = lazy_flat.num_vectors;
3423
3424 if dim_bits != schema_dim {
3425 return Err(Error::Corruption(format!(
3426 "binary vector field {} has schema dimension {} but flat storage dimension {}",
3427 field.0, schema_dim, dim_bits
3428 )));
3429 }
3430
3431 if byte_len != query.len() {
3432 return Err(Error::Schema(format!(
3433 "Binary query vector byte length {} != field byte length {}",
3434 query.len(),
3435 byte_len
3436 )));
3437 }
3438
3439 let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
3440 let mut collector = FlatDocumentCollector::new(k, combiner);
3441 let mut scores = vec![0f32; batch_len];
3442
3443 for batch_start in (0..n).step_by(batch_len) {
3444 let batch_count = batch_len.min(n - batch_start);
3445 let batch_bytes = lazy_flat
3446 .read_vectors_batch_sync(batch_start, batch_count)
3447 .map_err(crate::Error::Io)?;
3448 let raw = batch_bytes.as_slice();
3449
3450 crate::structures::simd::batch_hamming_scores(
3451 query,
3452 raw,
3453 byte_len,
3454 dim_bits,
3455 &mut scores[..batch_count],
3456 );
3457
3458 for (i, &score) in scores.iter().enumerate().take(batch_count) {
3459 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
3460 collector.push(doc_id, ordinal, score);
3461 }
3462 }
3463
3464 let results = collector.into_results();
3465
3466 crate::observe::dense_l1(
3467 self.schema.index_label(),
3468 self.schema.get_field_name(field).unwrap_or("?"),
3469 "binary_flat",
3470 t0.secs(),
3471 results.len(),
3472 );
3473 Ok(results)
3474 }
3475
3476 #[cfg(feature = "sync")]
3477 pub fn search_binary_dense_vector_sync(
3478 &self,
3479 field: Field,
3480 query: &[u8],
3481 k: usize,
3482 combiner: crate::query::MultiValueCombiner,
3483 ) -> Result<Vec<VectorSearchResult>> {
3484 self.search_binary_dense_vector_sync_impl(field, query, k, combiner, None)
3485 }
3486
3487 #[cfg(feature = "sync")]
3488 pub(crate) fn search_binary_dense_vector_sync_with_probe_cache(
3489 &self,
3490 field: Field,
3491 query: &[u8],
3492 k: usize,
3493 combiner: crate::query::MultiValueCombiner,
3494 probe_cache: &std::sync::Mutex<Option<crate::structures::IvfProbePlan>>,
3495 ) -> Result<Vec<VectorSearchResult>> {
3496 self.search_binary_dense_vector_sync_impl(field, query, k, combiner, Some(probe_cache))
3497 }
3498}
3499
3500#[cfg(test)]
3501mod dense_search_safety_tests {
3502 use super::*;
3503
3504 #[test]
3505 fn dense_fetch_count_rejects_non_finite_and_unbounded_factors() {
3506 for factor in [
3507 f32::NAN,
3508 f32::INFINITY,
3509 f32::NEG_INFINITY,
3510 0.0,
3511 0.5,
3512 2.01,
3513 MAX_DENSE_RERANK_FACTOR + 1.0,
3514 ] {
3515 assert!(
3516 checked_dense_fetch_k(10, factor).is_err(),
3517 "factor={factor}"
3518 );
3519 }
3520 }
3521
3522 fn values_as_bytes<T>(values: &[T]) -> &[u8] {
3523 unsafe {
3524 std::slice::from_raw_parts(values.as_ptr() as *const u8, std::mem::size_of_val(values))
3525 }
3526 }
3527
3528 fn assert_prepared_dense_scores_match_legacy(
3529 quantization: DenseVectorQuantization,
3530 raw: &[u8],
3531 unit_norm: bool,
3532 ) {
3533 const DIM: usize = 4;
3534 const VECTOR_COUNT: usize = 4;
3535 let query = [0.25, -0.5, 0.75, 1.0];
3536 let mut expected = [0.0; VECTOR_COUNT];
3537 SegmentReader::score_quantized_batch_legacy(
3538 &query,
3539 raw,
3540 quantization,
3541 DIM,
3542 &mut expected,
3543 unit_norm,
3544 )
3545 .unwrap();
3546
3547 let prepared = PreparedDenseScoreQuery::new(&query, quantization, DIM, unit_norm).unwrap();
3548 let vector_bytes = DIM
3549 * match quantization {
3550 DenseVectorQuantization::F32 => std::mem::size_of::<f32>(),
3551 DenseVectorQuantization::F16 => std::mem::size_of::<u16>(),
3552 DenseVectorQuantization::UInt8 => 1,
3553 DenseVectorQuantization::Binary => unreachable!(),
3554 };
3555 let split = 2 * vector_bytes;
3556 let mut actual = [0.0; VECTOR_COUNT];
3557 prepared
3558 .score_batch(&raw[..split], &mut actual[..2])
3559 .unwrap();
3560 prepared
3561 .score_batch(&raw[split..], &mut actual[2..])
3562 .unwrap();
3563
3564 assert_eq!(
3565 actual.map(f32::to_bits),
3566 expected.map(f32::to_bits),
3567 "quantization={quantization:?}, unit_norm={unit_norm}"
3568 );
3569 }
3570
3571 #[test]
3572 fn prepared_dense_query_matches_legacy_scoring_across_batches() {
3573 let vectors_f32 = [
3574 0.5, -0.25, 0.75, 1.0, -1.0, 0.5, 0.25, 0.125, 0.0, 0.0, 0.0, 0.0, 0.75, 0.5, -0.5,
3575 -0.25,
3576 ];
3577 let vectors_f16: Vec<u16> = vectors_f32
3578 .iter()
3579 .map(|&value| crate::structures::simd::f32_to_f16(value))
3580 .collect();
3581 let vectors_u8 = [
3582 255, 96, 224, 160, 0, 192, 144, 128, 128, 128, 128, 128, 224, 192, 64, 96,
3583 ];
3584
3585 for unit_norm in [false, true] {
3586 assert_prepared_dense_scores_match_legacy(
3587 DenseVectorQuantization::F32,
3588 values_as_bytes(&vectors_f32),
3589 unit_norm,
3590 );
3591 assert_prepared_dense_scores_match_legacy(
3592 DenseVectorQuantization::F16,
3593 values_as_bytes(&vectors_f16),
3594 unit_norm,
3595 );
3596 assert_prepared_dense_scores_match_legacy(
3597 DenseVectorQuantization::UInt8,
3598 &vectors_u8,
3599 unit_norm,
3600 );
3601 }
3602 }
3603
3604 #[test]
3605 fn prepared_dense_query_preserves_scoring_validation_errors() {
3606 assert!(matches!(
3607 PreparedDenseScoreQuery::new(&[1.0], DenseVectorQuantization::F32, 2, false).err(),
3608 Some(Error::Query(_))
3609 ));
3610 assert!(matches!(
3611 PreparedDenseScoreQuery::new(&[1.0], DenseVectorQuantization::Binary, 1, false).err(),
3612 Some(Error::InvalidFieldType { .. })
3613 ));
3614
3615 let query = [1.0, 2.0];
3616 let prepared =
3617 PreparedDenseScoreQuery::new(&query, DenseVectorQuantization::F32, 2, false).unwrap();
3618 let mut scores = [0.0];
3619 assert!(matches!(
3620 prepared.score_batch(&[0; 7], &mut scores),
3621 Err(Error::Corruption(_))
3622 ));
3623 }
3624
3625 #[test]
3626 fn flat_document_collector_does_not_let_one_multivalue_doc_crowd_out_others() {
3627 let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
3628 collector.push(1, 0, 1.0);
3629 collector.push(1, 1, 0.9);
3630 collector.push(2, 0, 0.8);
3631
3632 let results = collector.into_results();
3633 assert_eq!(
3634 results
3635 .iter()
3636 .map(|result| result.doc_id)
3637 .collect::<Vec<_>>(),
3638 vec![1, 2]
3639 );
3640 assert_eq!(results[0].ordinals.len(), 2);
3641 }
3642
3643 #[test]
3644 fn flat_document_collector_evicts_by_score_then_doc_id() {
3645 let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
3646 collector.push(1, 0, 0.5);
3647 collector.push(3, 0, 0.8);
3648 collector.push(2, 0, 0.9);
3649 let results = collector.into_results();
3650 assert_eq!(
3651 results
3652 .iter()
3653 .map(|result| result.doc_id)
3654 .collect::<Vec<_>>(),
3655 vec![2, 3]
3656 );
3657
3658 let mut tied = FlatDocumentCollector::new(1, crate::query::MultiValueCombiner::Max);
3659 tied.push(2, 0, 1.0);
3660 tied.push(1, 0, 1.0);
3661 let results = tied.into_results();
3662 assert_eq!(results[0].doc_id, 1);
3663 }
3664
3665 #[test]
3666 fn dense_fetch_count_rounds_up_and_detects_overflow() {
3667 assert_eq!(checked_dense_fetch_k(3, 1.5).unwrap(), 5);
3668 assert_eq!(checked_dense_fetch_k(10_000, 2.0).unwrap(), 20_000);
3669 assert!(checked_dense_fetch_k(10_001, 2.0).is_err());
3670 assert!(checked_dense_fetch_k(usize::MAX, 2.0).is_err());
3671 }
3672
3673 #[test]
3674 fn binary_combined_fetch_count_uses_shared_bounded_oversampling() {
3675 assert_eq!(checked_binary_combined_fetch_k(3).unwrap(), 6);
3676 assert_eq!(checked_binary_combined_fetch_k(10_000).unwrap(), 20_000);
3677 assert_eq!(checked_binary_combined_fetch_k(10_001).unwrap(), 20_000);
3678 assert_eq!(checked_binary_combined_fetch_k(20_000).unwrap(), 20_000);
3679 assert!(checked_binary_combined_fetch_k(20_001).is_err());
3680 assert!(checked_binary_combined_fetch_k(usize::MAX).is_err());
3681 }
3682
3683 #[cfg(feature = "native")]
3684 #[test]
3685 fn legacy_ivf_tq_generation_is_rejected_while_opening() {
3686 use crate::directories::OwnedBytes;
3687 use crate::dsl::IvfRoutingMode;
3688 use crate::segment::ann_disk::{AnnDiskIndex, AnnKind};
3689
3690 let centroids = CoarseCentroids {
3691 num_clusters: 1,
3692 dim: 2,
3693 centroids: vec![1.0, 0.0],
3694 version: 7,
3695 soar_config: None,
3696 routing_index: None,
3697 };
3698 let mut build_centroids = centroids.clone();
3699 build_centroids.version =
3700 crate::structures::mark_ivf_tq_cosine_generation(build_centroids.version);
3701 let mut bytes = crate::segment::ann_build::build_ivf_tq(
3702 2,
3703 IvfRoutingMode::Flat,
3704 &build_centroids,
3705 &[(0, 0)],
3706 &[1.0, 0.0],
3707 )
3708 .unwrap();
3709 bytes[24..32].copy_from_slice(¢roids.version.to_le_bytes());
3712 let error = AnnDiskIndex::open(OwnedBytes::new(bytes), AnnKind::IvfTq, 1)
3713 .err()
3714 .expect("legacy IVF-TQ payload must fail while opening")
3715 .to_string();
3716 assert!(error.contains("unsupported legacy generation"), "{error}");
3717 }
3718
3719 #[test]
3720 fn rerank_batch_is_capped_by_actual_candidate_vectors() {
3721 assert_eq!(bounded_rerank_batch(3_072, DENSE_SCORE_BATCH, 20), 20);
3722 assert_eq!(
3723 bounded_rerank_batch(3_072, DENSE_SCORE_BATCH, 10_000),
3724 MAX_VECTOR_SCORE_BATCH_BYTES / 3_072
3725 );
3726 assert_eq!(bounded_rerank_batch(3_072, DENSE_SCORE_BATCH, 0), 1);
3727 }
3728
3729 #[test]
3730 fn file_ranges_reject_overflow_and_truncation() {
3731 assert_eq!(checked_file_range(4, 3, 7, "test").unwrap(), 4..7);
3732 assert!(checked_file_range(u64::MAX, 1, u64::MAX, "test").is_err());
3733 assert!(checked_file_range(5, 3, 7, "test").is_err());
3734 }
3735
3736 #[test]
3737 fn shared_tq_plan_cache_rebuilds_for_divergent_query_clones() {
3738 let codec = crate::structures::TqCodec::new(4);
3739 let cache = std::sync::Mutex::new(None);
3740 let original_query = vec![1.0, 2.0, 3.0, 4.0];
3741
3742 let original =
3743 cached_tq_query_plan(&codec, &original_query, Some(&cache)).expect("build plan");
3744 let reused =
3745 cached_tq_query_plan(&codec, &original_query, Some(&cache)).expect("reuse plan");
3746 assert!(
3747 std::sync::Arc::ptr_eq(&original, &reused),
3748 "unchanged queries must share their plan across segments"
3749 );
3750
3751 let mut divergent_clone = original_query.clone();
3752 divergent_clone[0] = -1.0;
3753 let rebuilt =
3754 cached_tq_query_plan(&codec, &divergent_clone, Some(&cache)).expect("rebuild plan");
3755 assert!(
3756 !std::sync::Arc::ptr_eq(&original, &rebuilt),
3757 "a clone with a mutated vector must not reuse stale LUTs"
3758 );
3759 assert!(rebuilt.matches_query(&divergent_clone));
3760 assert!(!rebuilt.matches_query(&original_query));
3761 }
3762
3763 #[test]
3764 fn candidate_vector_reads_coalesce_contiguous_values() {
3765 let mut runs = Vec::new();
3766 plan_vector_read_runs(&[3, 4, 5, 9, 12, 13], &mut runs).unwrap();
3767 assert_eq!(runs.len(), 3);
3768 assert!(matches!(
3769 runs.as_slice(),
3770 [
3771 VectorReadRun {
3772 buffer_start: 0,
3773 flat_start: 3,
3774 count: 3,
3775 },
3776 VectorReadRun {
3777 buffer_start: 3,
3778 flat_start: 9,
3779 count: 1,
3780 },
3781 VectorReadRun {
3782 buffer_start: 4,
3783 flat_start: 12,
3784 count: 2,
3785 },
3786 ]
3787 ));
3788 assert!(plan_vector_read_runs(&[3, 3], &mut runs).is_err());
3789 }
3790
3791 #[tokio::test]
3792 async fn binary_single_value_ann_fast_path_validates_and_deduplicates() {
3793 use crate::directories::{FileHandle, OwnedBytes};
3794 use crate::segment::FlatVectorData;
3795
3796 let mut encoded = Vec::new();
3797 FlatVectorData::serialize_binary_from_bits_streaming(
3798 8,
3799 &[0x0f, 0xf0],
3800 &[(1, 0), (3, 2)],
3801 &mut encoded,
3802 )
3803 .unwrap();
3804 let flat = LazyFlatVectorData::open_with_doc_limit(
3805 FileHandle::from_bytes(OwnedBytes::new(encoded)),
3806 Some(4),
3807 )
3808 .await
3809 .unwrap();
3810 assert_eq!(flat.num_vectors, flat.num_docs_with_vectors());
3811
3812 let validated = validate_binary_single_value_ann_results(
3813 vec![(3, 2, 0.9), (1, 0, 0.8), (3, 2, 0.7)],
3814 &flat,
3815 )
3816 .unwrap();
3817 assert_eq!(validated, vec![(3, 2, 0.9), (1, 0, 0.8)]);
3818
3819 assert!(matches!(
3820 validate_binary_single_value_ann_results(vec![(2, 0, 1.0)], &flat),
3821 Err(Error::Corruption(_))
3822 ));
3823 assert!(matches!(
3824 validate_binary_single_value_ann_results(vec![(3, 0, 1.0)], &flat),
3825 Err(Error::Corruption(_))
3826 ));
3827 }
3828
3829 #[tokio::test]
3830 async fn multivalue_ann_rerank_streams_past_document_candidate_cap() {
3831 use crate::directories::{FileHandle, OwnedBytes};
3832 use crate::segment::FlatVectorData;
3833
3834 const VALUES: usize = MAX_DENSE_CANDIDATES_PER_SEGMENT + 1;
3835 let mut encoded = Vec::new();
3836 let vectors = vec![1.0f32; VALUES];
3837 let doc_ids: Vec<_> = (0..VALUES).map(|ordinal| (0, ordinal as u16)).collect();
3838 FlatVectorData::serialize_binary_from_flat_streaming(
3839 1,
3840 &vectors,
3841 &doc_ids,
3842 DenseVectorQuantization::F32,
3843 &mut encoded,
3844 )
3845 .unwrap();
3846 let flat = LazyFlatVectorData::open_with_doc_limit(
3847 FileHandle::from_bytes(OwnedBytes::new(encoded)),
3848 Some(1),
3849 )
3850 .await
3851 .unwrap();
3852
3853 let (results, stats) = exact_score_dense_candidate_documents(
3854 &[(0, 0, 0.0)],
3855 &flat,
3856 &[1.0],
3857 false,
3858 crate::query::MultiValueCombiner::Max,
3859 1,
3860 )
3861 .await
3862 .unwrap();
3863 assert_eq!(stats.vector_count, VALUES);
3864 assert_eq!(results.len(), 1);
3865 assert_eq!(results[0].ordinals.len(), VALUES);
3866 assert!((results[0].score - 1.0).abs() < 1e-5);
3867 }
3868}