1pub(crate) mod bmp;
4pub(crate) mod loader;
5mod types;
6
7pub use bmp::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)]
28pub struct SegmentMemoryStats {
29 pub segment_id: u128,
31 pub num_docs: u32,
33 pub term_dict_cache_bytes: usize,
35 pub store_cache_bytes: usize,
37 pub sparse_index_bytes: usize,
39 pub dense_index_bytes: usize,
41 pub bloom_filter_bytes: usize,
43 pub pinned_metadata_bytes: u64,
45 pub pin_intended_bytes: u64,
48}
49
50impl SegmentMemoryStats {
51 pub fn total_bytes(&self) -> usize {
53 self.term_dict_cache_bytes
54 + self.store_cache_bytes
55 + self.sparse_index_bytes
56 + self.dense_index_bytes
57 + self.bloom_filter_bytes
58 }
59}
60
61use std::cmp::Ordering;
62use std::collections::BinaryHeap;
63use std::sync::Arc;
64
65use rustc_hash::FxHashMap;
66
67use super::vector_data::LazyFlatVectorData;
68use crate::directories::{Directory, FileHandle};
69use crate::dsl::{DenseVectorQuantization, Document, Field, Schema};
70use crate::query::{MAX_DENSE_NPROBE, MAX_DENSE_RERANK_FACTOR};
71use crate::structures::{
72 AsyncSSTableReader, BlockPostingList, CoarseCentroids, SSTableStats, TermInfo,
73};
74use crate::{DocId, Error, Result};
75
76use super::store::{AsyncStoreReader, RawStoreBlock};
77use super::types::{SegmentFiles, SegmentId, SegmentMeta};
78
79pub(crate) fn combine_ordinal_results(
85 raw: impl IntoIterator<Item = (u32, u16, f32)>,
86 combiner: crate::query::MultiValueCombiner,
87 limit: usize,
88) -> Vec<VectorSearchResult> {
89 let collected: Vec<(u32, u16, f32)> = raw.into_iter().collect();
90
91 let num_raw = collected.len();
92 if log::log_enabled!(log::Level::Debug) {
93 let mut ids: Vec<u32> = collected.iter().map(|(d, _, _)| *d).collect();
94 ids.sort_unstable();
95 ids.dedup();
96 log::debug!(
97 "combine_ordinal_results: {} raw entries, {} unique docs, combiner={:?}, limit={}",
98 num_raw,
99 ids.len(),
100 combiner,
101 limit
102 );
103 }
104
105 let all_single = collected.iter().all(|&(_, ord, _)| ord == 0);
107 if all_single {
108 let mut results: Vec<VectorSearchResult> = collected
109 .into_iter()
110 .map(|(doc_id, _, score)| VectorSearchResult::new(doc_id, score, vec![(0, score)]))
111 .collect();
112 results.sort_unstable_by(|a, b| {
113 b.score
114 .total_cmp(&a.score)
115 .then_with(|| a.doc_id.cmp(&b.doc_id))
116 });
117 results.truncate(limit);
118 return results;
119 }
120
121 let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
123 rustc_hash::FxHashMap::default();
124 for (doc_id, ordinal, score) in collected {
125 doc_ordinals
126 .entry(doc_id as DocId)
127 .or_default()
128 .push((ordinal as u32, score));
129 }
130 let mut results: Vec<VectorSearchResult> = doc_ordinals
131 .into_iter()
132 .map(|(doc_id, ordinals)| {
133 let combined_score = combiner.combine(&ordinals);
134 VectorSearchResult::new(doc_id, combined_score, ordinals)
135 })
136 .collect();
137 results.sort_unstable_by(|a, b| {
138 b.score
139 .total_cmp(&a.score)
140 .then_with(|| a.doc_id.cmp(&b.doc_id))
141 });
142 results.truncate(limit);
143 results
144}
145
146struct HeapVectorResult(VectorSearchResult);
151
152impl PartialEq for HeapVectorResult {
153 fn eq(&self, other: &Self) -> bool {
154 self.0.score.to_bits() == other.0.score.to_bits() && self.0.doc_id == other.0.doc_id
155 }
156}
157
158impl Eq for HeapVectorResult {}
159
160impl Ord for HeapVectorResult {
161 fn cmp(&self, other: &Self) -> Ordering {
162 other
165 .0
166 .score
167 .total_cmp(&self.0.score)
168 .then_with(|| self.0.doc_id.cmp(&other.0.doc_id))
169 }
170}
171
172impl PartialOrd for HeapVectorResult {
173 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
174 Some(self.cmp(other))
175 }
176}
177
178struct FlatDocumentCollector {
182 heap: BinaryHeap<HeapVectorResult>,
183 limit: usize,
184 combiner: crate::query::MultiValueCombiner,
185 current_doc: Option<DocId>,
186 current_ordinals: Vec<(u32, f32)>,
187}
188
189impl FlatDocumentCollector {
190 fn new(limit: usize, combiner: crate::query::MultiValueCombiner) -> Self {
191 Self {
192 heap: BinaryHeap::with_capacity(limit.min(8 * 1024)),
193 limit,
194 combiner,
195 current_doc: None,
196 current_ordinals: Vec::new(),
197 }
198 }
199
200 fn push(&mut self, doc_id: DocId, ordinal: u16, score: f32) {
201 if self.current_doc.is_some_and(|current| current != doc_id) {
202 self.finish_current();
203 }
204 self.current_doc = Some(doc_id);
205 self.current_ordinals.push((ordinal as u32, score));
206 }
207
208 fn finish_current(&mut self) {
209 let Some(doc_id) = self.current_doc.take() else {
210 return;
211 };
212 let score = self.combiner.combine(&self.current_ordinals);
213 let should_retain = self.heap.len() < self.limit
214 || self.heap.peek().is_some_and(|worst| {
215 HeapVectorResult(VectorSearchResult::new(doc_id, score, Vec::new()))
216 .cmp(worst)
217 .is_lt()
218 });
219
220 if !should_retain {
221 self.current_ordinals.clear();
225 return;
226 }
227
228 let ordinals = std::mem::take(&mut self.current_ordinals);
229 let entry = HeapVectorResult(VectorSearchResult::new(doc_id, score, ordinals));
230 if self.heap.len() < self.limit {
231 self.heap.push(entry);
232 } else if let Some(mut worst) = self.heap.peek_mut() {
233 let mut evicted = std::mem::replace(&mut worst.0, entry.0);
236 evicted.ordinals.clear();
237 self.current_ordinals = evicted.ordinals;
238 }
239 }
240
241 fn into_results(mut self) -> Vec<VectorSearchResult> {
242 self.finish_current();
243 let mut results: Vec<_> = self.heap.into_iter().map(|entry| entry.0).collect();
244 results.sort_unstable_by(|a, b| {
245 b.score
246 .total_cmp(&a.score)
247 .then_with(|| a.doc_id.cmp(&b.doc_id))
248 });
249 results
250 }
251}
252
253fn combine_grouped_ordinal_results(
256 raw: impl IntoIterator<Item = RawVectorCandidate>,
257 combiner: crate::query::MultiValueCombiner,
258 limit: usize,
259) -> Vec<VectorSearchResult> {
260 let mut collector = FlatDocumentCollector::new(limit, combiner);
261 for (doc_id, ordinal, score) in raw {
262 collector.push(doc_id, ordinal, score);
263 }
264 collector.into_results()
265}
266
267#[derive(Clone, Copy)]
268struct DenseSearchParams {
269 dim: usize,
270 nprobe: usize,
271 unit_norm: bool,
272}
273
274fn checked_dense_fetch_k(k: usize, rerank_factor: f32) -> Result<usize> {
276 if !rerank_factor.is_finite() || !(1.0..=MAX_DENSE_RERANK_FACTOR).contains(&rerank_factor) {
277 return Err(Error::Query(format!(
278 "dense rerank_factor must be finite and in [1, {MAX_DENSE_RERANK_FACTOR}], got {rerank_factor}"
279 )));
280 }
281
282 let fetch = (k as f64) * (rerank_factor as f64);
283 if !fetch.is_finite()
284 || fetch > usize::MAX as f64
285 || fetch > MAX_DENSE_CANDIDATES_PER_SEGMENT as f64
286 {
287 return Err(Error::Query(format!(
288 "dense candidate count exceeds the per-segment maximum of \
289 {MAX_DENSE_CANDIDATES_PER_SEGMENT}: k={k}, rerank_factor={rerank_factor}"
290 )));
291 }
292 Ok(fetch.ceil() as usize)
293}
294
295#[inline]
296fn bounded_vector_score_batch(vector_byte_size: usize, preferred: usize) -> usize {
297 preferred.min((MAX_VECTOR_SCORE_BATCH_BYTES / vector_byte_size.max(1)).max(1))
298}
299
300fn checked_file_range(
301 offset: u64,
302 length: u64,
303 file_length: u64,
304 description: &str,
305) -> Result<std::ops::Range<u64>> {
306 let end = offset
307 .checked_add(length)
308 .ok_or_else(|| Error::Corruption(format!("{description} byte range overflows u64")))?;
309 if end > file_length {
310 return Err(Error::Corruption(format!(
311 "{description} byte range {offset}..{end} exceeds file length {file_length}"
312 )));
313 }
314 Ok(offset..end)
315}
316
317type RawVectorCandidate = (u32, u16, f32);
318type CandidateVectorRef = (DocId, u16, usize); #[derive(Clone, Copy)]
321struct CandidateDocumentRange {
322 doc_id: DocId,
323 start: usize,
324 end: usize,
325}
326
327struct AnnCandidateDocuments {
328 ranges: Vec<CandidateDocumentRange>,
329 vector_count: usize,
330}
331
332fn ann_candidate_document_ranges(
341 ann_results: &[RawVectorCandidate],
342 flat: &LazyFlatVectorData,
343) -> Result<AnnCandidateDocuments> {
344 let mut candidate_docs: Vec<DocId> = ann_results.iter().map(|candidate| candidate.0).collect();
345 candidate_docs.sort_unstable();
346 candidate_docs.dedup();
347
348 let mut ranges = Vec::with_capacity(candidate_docs.len());
349 let mut vector_count = 0usize;
350 for doc_id in candidate_docs {
351 let (start, count) = flat.flat_indexes_for_doc_range(doc_id);
352 if count == 0 {
353 return Err(Error::Corruption(format!(
354 "ANN candidate document {doc_id} is missing from flat vector storage"
355 )));
356 }
357 vector_count = vector_count
358 .checked_add(count)
359 .ok_or_else(|| Error::Query("ANN candidate vector expansion overflow".to_string()))?;
360 let end = start
361 .checked_add(count)
362 .ok_or_else(|| Error::Corruption("flat vector range overflow".to_string()))?;
363 if end > flat.num_vectors {
364 return Err(Error::Corruption(format!(
365 "flat vector range {start}..{end} for document {doc_id} exceeds {} vectors",
366 flat.num_vectors
367 )));
368 }
369 ranges.push(CandidateDocumentRange { doc_id, start, end });
370 }
371 Ok(AnnCandidateDocuments {
372 ranges,
373 vector_count,
374 })
375}
376
377struct CandidateVectorCursor<'a> {
378 ranges: &'a [CandidateDocumentRange],
379 range_index: usize,
380 flat_index: usize,
381}
382
383impl<'a> CandidateVectorCursor<'a> {
384 fn new(ranges: &'a [CandidateDocumentRange]) -> Self {
385 Self {
386 ranges,
387 range_index: 0,
388 flat_index: ranges.first().map_or(0, |range| range.start),
389 }
390 }
391
392 fn fill_batch(
396 &mut self,
397 flat: &LazyFlatVectorData,
398 batch: &mut Vec<CandidateVectorRef>,
399 limit: usize,
400 ) -> Result<bool> {
401 batch.clear();
402 while batch.len() < limit && self.range_index < self.ranges.len() {
403 let range = self.ranges[self.range_index];
404 if self.flat_index == range.end {
405 self.range_index += 1;
406 if let Some(next) = self.ranges.get(self.range_index) {
407 self.flat_index = next.start;
408 }
409 continue;
410 }
411 let (stored_doc_id, ordinal) = flat.get_doc_id(self.flat_index);
412 if stored_doc_id != range.doc_id {
413 return Err(Error::Corruption(format!(
414 "flat vector doc map is not contiguous for document {}",
415 range.doc_id
416 )));
417 }
418 batch.push((range.doc_id, ordinal, self.flat_index));
419 self.flat_index += 1;
420 }
421 Ok(!batch.is_empty())
422 }
423}
424
425#[derive(Clone, Copy)]
426struct VectorReadRun {
427 buffer_start: usize,
428 flat_start: usize,
429 count: usize,
430}
431
432fn plan_vector_read_runs(indexes: &[usize], runs: &mut Vec<VectorReadRun>) -> Result<()> {
437 runs.clear();
438 for (buffer_index, &flat_index) in indexes.iter().enumerate() {
439 if let Some(run) = runs.last_mut()
440 && run
441 .flat_start
442 .checked_add(run.count)
443 .is_some_and(|next| next == flat_index)
444 {
445 run.count += 1;
446 continue;
447 }
448 if buffer_index > 0 && flat_index <= indexes[buffer_index - 1] {
449 return Err(Error::Corruption(
450 "candidate flat-vector indexes are not strictly ordered".into(),
451 ));
452 }
453 runs.push(VectorReadRun {
454 buffer_start: buffer_index,
455 flat_start: flat_index,
456 count: 1,
457 });
458 }
459 Ok(())
460}
461
462async fn read_vector_runs(
463 flat: &LazyFlatVectorData,
464 indexes: &[usize],
465 runs: &mut Vec<VectorReadRun>,
466 output: &mut [u8],
467) -> Result<()> {
468 plan_vector_read_runs(indexes, runs)?;
469 let vector_byte_size = flat.vector_byte_size();
470 for run in runs {
471 let bytes = flat
472 .read_vectors_batch(run.flat_start, run.count)
473 .await
474 .map_err(Error::Io)?;
475 let start = run
476 .buffer_start
477 .checked_mul(vector_byte_size)
478 .ok_or_else(|| Error::Query("dense rerank buffer offset overflow".into()))?;
479 let end = start
480 .checked_add(bytes.len())
481 .ok_or_else(|| Error::Query("dense rerank buffer range overflow".into()))?;
482 let destination = output
483 .get_mut(start..end)
484 .ok_or_else(|| Error::Corruption("dense rerank buffer is too short".into()))?;
485 destination.copy_from_slice(bytes.as_slice());
486 }
487 Ok(())
488}
489
490#[cfg(feature = "sync")]
491fn read_vector_runs_sync(
492 flat: &LazyFlatVectorData,
493 indexes: &[usize],
494 runs: &mut Vec<VectorReadRun>,
495 output: &mut [u8],
496) -> Result<()> {
497 plan_vector_read_runs(indexes, runs)?;
498 let vector_byte_size = flat.vector_byte_size();
499 for run in runs {
500 let bytes = flat
501 .read_vectors_batch_sync(run.flat_start, run.count)
502 .map_err(Error::Io)?;
503 let start = run
504 .buffer_start
505 .checked_mul(vector_byte_size)
506 .ok_or_else(|| Error::Query("dense rerank buffer offset overflow".into()))?;
507 let end = start
508 .checked_add(bytes.len())
509 .ok_or_else(|| Error::Query("dense rerank buffer range overflow".into()))?;
510 let destination = output
511 .get_mut(start..end)
512 .ok_or_else(|| Error::Corruption("dense rerank buffer is too short".into()))?;
513 destination.copy_from_slice(bytes.as_slice());
514 }
515 Ok(())
516}
517
518#[derive(Default)]
519struct DenseRerankStats {
520 vector_count: usize,
521 resolve_elapsed: std::time::Duration,
522 read_elapsed: std::time::Duration,
523 score_elapsed: std::time::Duration,
524}
525
526async fn exact_score_dense_candidate_documents(
527 ann_results: &[RawVectorCandidate],
528 flat: &LazyFlatVectorData,
529 query: &[f32],
530 unit_norm: bool,
531 combiner: crate::query::MultiValueCombiner,
532 limit: usize,
533) -> Result<(Vec<VectorSearchResult>, DenseRerankStats)> {
534 let resolve_started = std::time::Instant::now();
535 let documents = ann_candidate_document_ranges(ann_results, flat)?;
536 let mut stats = DenseRerankStats {
537 vector_count: documents.vector_count,
538 resolve_elapsed: resolve_started.elapsed(),
539 ..Default::default()
540 };
541 let vector_byte_size = flat.vector_byte_size();
542 let batch_len = bounded_vector_score_batch(vector_byte_size, DENSE_SCORE_BATCH);
543 let raw_capacity = batch_len
544 .checked_mul(vector_byte_size)
545 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".to_string()))?;
546 let mut raw = vec![0u8; raw_capacity];
547 let mut scores = vec![0.0f32; batch_len];
548 let mut batch = Vec::with_capacity(batch_len);
549 let mut flat_indexes = Vec::with_capacity(batch_len);
550 let mut read_runs = Vec::new();
551 let mut cursor = CandidateVectorCursor::new(&documents.ranges);
552 let mut collector = FlatDocumentCollector::new(limit, combiner);
553 let mut scored = 0usize;
554
555 while cursor.fill_batch(flat, &mut batch, batch_len)? {
556 flat_indexes.clear();
557 flat_indexes.extend(batch.iter().map(|&(_, _, flat_index)| flat_index));
558 #[cfg(feature = "native")]
559 flat.prefetch_vectors(flat_indexes.iter().copied());
560 let raw_len = batch
561 .len()
562 .checked_mul(vector_byte_size)
563 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".to_string()))?;
564 let raw = &mut raw[..raw_len];
565
566 let read_started = std::time::Instant::now();
567 read_vector_runs(flat, &flat_indexes, &mut read_runs, raw).await?;
568 stats.read_elapsed += read_started.elapsed();
569
570 let score_started = std::time::Instant::now();
571 SegmentReader::score_quantized_batch(
572 query,
573 raw,
574 flat.quantization,
575 flat.dim,
576 &mut scores[..batch.len()],
577 unit_norm,
578 )?;
579 stats.score_elapsed += score_started.elapsed();
580 for (buffer_index, &(doc_id, ordinal, _)) in batch.iter().enumerate() {
581 collector.push(doc_id, ordinal, scores[buffer_index]);
582 }
583 scored += batch.len();
584 }
585 debug_assert_eq!(scored, documents.vector_count);
586 Ok((collector.into_results(), stats))
587}
588
589#[cfg(feature = "sync")]
590fn exact_score_dense_candidate_documents_sync(
591 ann_results: &[RawVectorCandidate],
592 flat: &LazyFlatVectorData,
593 query: &[f32],
594 unit_norm: bool,
595 combiner: crate::query::MultiValueCombiner,
596 limit: usize,
597) -> Result<Vec<VectorSearchResult>> {
598 let documents = ann_candidate_document_ranges(ann_results, flat)?;
599 let vector_byte_size = flat.vector_byte_size();
600 let batch_len = bounded_vector_score_batch(vector_byte_size, DENSE_SCORE_BATCH);
601 let raw_capacity = batch_len
602 .checked_mul(vector_byte_size)
603 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".to_string()))?;
604 let mut raw = vec![0u8; raw_capacity];
605 let mut scores = vec![0.0f32; batch_len];
606 let mut batch = Vec::with_capacity(batch_len);
607 let mut flat_indexes = Vec::with_capacity(batch_len);
608 let mut read_runs = Vec::new();
609 let mut cursor = CandidateVectorCursor::new(&documents.ranges);
610 let mut collector = FlatDocumentCollector::new(limit, combiner);
611 let mut scored = 0usize;
612
613 while cursor.fill_batch(flat, &mut batch, batch_len)? {
614 flat_indexes.clear();
615 flat_indexes.extend(batch.iter().map(|&(_, _, flat_index)| flat_index));
616 let raw_len = batch
617 .len()
618 .checked_mul(vector_byte_size)
619 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".to_string()))?;
620 let raw = &mut raw[..raw_len];
621 read_vector_runs_sync(flat, &flat_indexes, &mut read_runs, raw)?;
622 SegmentReader::score_quantized_batch(
623 query,
624 raw,
625 flat.quantization,
626 flat.dim,
627 &mut scores[..batch.len()],
628 unit_norm,
629 )?;
630 for (buffer_index, &(doc_id, ordinal, _)) in batch.iter().enumerate() {
631 collector.push(doc_id, ordinal, scores[buffer_index]);
632 }
633 scored += batch.len();
634 }
635 debug_assert_eq!(scored, documents.vector_count);
636 Ok(collector.into_results())
637}
638
639async fn exact_score_binary_candidate_documents(
640 ann_results: &[RawVectorCandidate],
641 flat: &LazyFlatVectorData,
642 query: &[u8],
643 dim_bits: usize,
644 combiner: crate::query::MultiValueCombiner,
645 limit: usize,
646) -> Result<Vec<VectorSearchResult>> {
647 let documents = ann_candidate_document_ranges(ann_results, flat)?;
648 let probe_scores: FxHashMap<(DocId, u16), f32> = ann_results
649 .iter()
650 .map(|&(doc_id, ordinal, score)| ((doc_id, ordinal), score))
651 .collect();
652 let vector_byte_size = flat.vector_byte_size();
653 let batch_len = bounded_vector_score_batch(vector_byte_size, BINARY_SCORE_BATCH);
654 let raw_capacity = batch_len
655 .checked_mul(vector_byte_size)
656 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
657 let mut raw = vec![0u8; raw_capacity];
658 let mut scores = vec![0.0f32; batch_len];
659 let mut batch_scores = vec![0.0f32; batch_len];
660 let mut batch = Vec::with_capacity(batch_len);
661 let mut unresolved = Vec::with_capacity(batch_len);
662 let mut unresolved_flat_indexes = Vec::with_capacity(batch_len);
663 let mut read_runs = Vec::new();
664 let mut cursor = CandidateVectorCursor::new(&documents.ranges);
665 let mut collector = FlatDocumentCollector::new(limit, combiner);
666 let mut scored = 0usize;
667
668 while cursor.fill_batch(flat, &mut batch, batch_len)? {
669 unresolved.clear();
670 for (batch_index, &(doc_id, ordinal, flat_index)) in batch.iter().enumerate() {
671 if let Some(&score) = probe_scores.get(&(doc_id, ordinal)) {
672 batch_scores[batch_index] = score;
673 } else {
674 unresolved.push((batch_index, flat_index));
675 }
676 }
677 unresolved_flat_indexes.clear();
678 unresolved_flat_indexes.extend(unresolved.iter().map(|&(_, flat_index)| flat_index));
679 #[cfg(feature = "native")]
680 flat.prefetch_vectors(unresolved_flat_indexes.iter().copied());
681 let raw_len = unresolved
682 .len()
683 .checked_mul(vector_byte_size)
684 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
685 let raw = &mut raw[..raw_len];
686 read_vector_runs(flat, &unresolved_flat_indexes, &mut read_runs, raw).await?;
687 crate::structures::simd::batch_hamming_scores(
688 query,
689 raw,
690 vector_byte_size,
691 dim_bits,
692 &mut scores[..unresolved.len()],
693 );
694 for (buffer_index, &(batch_index, _)) in unresolved.iter().enumerate() {
695 batch_scores[batch_index] = scores[buffer_index];
696 }
697 for (batch_index, &(doc_id, ordinal, _)) in batch.iter().enumerate() {
698 collector.push(doc_id, ordinal, batch_scores[batch_index]);
699 }
700 scored += batch.len();
701 }
702 debug_assert_eq!(scored, documents.vector_count);
703 Ok(collector.into_results())
704}
705
706#[cfg(feature = "sync")]
707fn exact_score_binary_candidate_documents_sync(
708 ann_results: &[RawVectorCandidate],
709 flat: &LazyFlatVectorData,
710 query: &[u8],
711 dim_bits: usize,
712 combiner: crate::query::MultiValueCombiner,
713 limit: usize,
714) -> Result<Vec<VectorSearchResult>> {
715 let documents = ann_candidate_document_ranges(ann_results, flat)?;
716 let probe_scores: FxHashMap<(DocId, u16), f32> = ann_results
717 .iter()
718 .map(|&(doc_id, ordinal, score)| ((doc_id, ordinal), score))
719 .collect();
720 let vector_byte_size = flat.vector_byte_size();
721 let batch_len = bounded_vector_score_batch(vector_byte_size, BINARY_SCORE_BATCH);
722 let raw_capacity = batch_len
723 .checked_mul(vector_byte_size)
724 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
725 let mut raw = vec![0u8; raw_capacity];
726 let mut scores = vec![0.0f32; batch_len];
727 let mut batch_scores = vec![0.0f32; batch_len];
728 let mut batch = Vec::with_capacity(batch_len);
729 let mut unresolved = Vec::with_capacity(batch_len);
730 let mut unresolved_flat_indexes = Vec::with_capacity(batch_len);
731 let mut read_runs = Vec::new();
732 let mut cursor = CandidateVectorCursor::new(&documents.ranges);
733 let mut collector = FlatDocumentCollector::new(limit, combiner);
734 let mut scored = 0usize;
735
736 while cursor.fill_batch(flat, &mut batch, batch_len)? {
737 unresolved.clear();
738 for (batch_index, &(doc_id, ordinal, flat_index)) in batch.iter().enumerate() {
739 if let Some(&score) = probe_scores.get(&(doc_id, ordinal)) {
740 batch_scores[batch_index] = score;
741 } else {
742 unresolved.push((batch_index, flat_index));
743 }
744 }
745 unresolved_flat_indexes.clear();
746 unresolved_flat_indexes.extend(unresolved.iter().map(|&(_, flat_index)| flat_index));
747 let raw_len = unresolved
748 .len()
749 .checked_mul(vector_byte_size)
750 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
751 let raw = &mut raw[..raw_len];
752 read_vector_runs_sync(flat, &unresolved_flat_indexes, &mut read_runs, raw)?;
753 crate::structures::simd::batch_hamming_scores(
754 query,
755 raw,
756 vector_byte_size,
757 dim_bits,
758 &mut scores[..unresolved.len()],
759 );
760 for (buffer_index, &(batch_index, _)) in unresolved.iter().enumerate() {
761 batch_scores[batch_index] = scores[buffer_index];
762 }
763 for (batch_index, &(doc_id, ordinal, _)) in batch.iter().enumerate() {
764 collector.push(doc_id, ordinal, batch_scores[batch_index]);
765 }
766 scored += batch.len();
767 }
768 debug_assert_eq!(scored, documents.vector_count);
769 Ok(collector.into_results())
770}
771
772fn validate_coarse_centroids(centroids: &CoarseCentroids, dim: usize) -> Result<()> {
773 let expected = (centroids.num_clusters as usize)
774 .checked_mul(dim)
775 .ok_or_else(|| Error::Corruption("coarse centroid size overflow".into()))?;
776 if centroids.num_clusters == 0
777 || centroids.dim != dim
778 || centroids.centroids.len() != expected
779 || centroids.centroids.iter().any(|value| !value.is_finite())
780 {
781 return Err(Error::Corruption(format!(
782 "invalid coarse centroids: clusters={}, dim={}, values={} (expected dim={dim}, values={expected})",
783 centroids.num_clusters,
784 centroids.dim,
785 centroids.centroids.len()
786 )));
787 }
788 Ok(())
789}
790
791fn validate_ivf_pq_ann(
792 index: &crate::segment::ann_disk::AnnDiskIndex,
793 centroids: &CoarseCentroids,
794 codebook: &crate::structures::PQCodebook,
795 dim: usize,
796 routing: crate::dsl::IvfRoutingMode,
797) -> Result<()> {
798 let header = index.header();
799 if header.dim != dim
800 || codebook.config.dim != dim
801 || header.code_size != codebook.config.num_subspaces
802 || header.num_clusters != centroids.num_clusters
803 || header.quantizer_version != centroids.version
804 || header.codebook_version != codebook.version
805 || header.routing != routing
806 {
807 return Err(Error::Corruption(format!(
808 "IVF-PQ payload/codebook/centroid metadata does not match schema dimension {dim}"
809 )));
810 }
811 Ok(())
812}
813
814fn validate_binary_ann(
815 index: &crate::segment::ann_disk::AnnDiskIndex,
816 quantizer: &crate::structures::BinaryCoarseQuantizer,
817 config: &crate::dsl::BinaryDenseVectorConfig,
818 dim: usize,
819 field: Field,
820) -> Result<()> {
821 let header = index.header();
822 if header.dim != dim
823 || header.code_size != config.byte_len()
824 || header.num_clusters != quantizer.num_clusters
825 || header.quantizer_version != quantizer.version
826 || header.codebook_version != 0
827 || header.routing != config.ivf_routing
828 || quantizer.dim_bits != dim
829 {
830 return Err(Error::Corruption(format!(
831 "binary IVF field {} does not match its quantizer/schema generation",
832 field.0,
833 )));
834 }
835 Ok(())
836}
837
838fn float_query_plan(
839 centroids: &CoarseCentroids,
840 codebook: &crate::structures::PQCodebook,
841 query: &[f32],
842 nprobe: usize,
843 routing: crate::dsl::IvfRoutingMode,
844 cache: Option<&std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>>,
845) -> Result<std::sync::Arc<crate::structures::IvfPqQueryPlan>> {
846 let effective_nprobe = nprobe.clamp(1, centroids.num_clusters as usize);
847 let request_fingerprint = crate::structures::vector::ivf::routing::float_probe_fingerprint(
848 query,
849 effective_nprobe,
850 routing,
851 );
852 if let Some(cache) = cache {
853 let mut cached = cache
854 .lock()
855 .map_err(|_| Error::Internal("dense IVF probe cache is poisoned".into()))?;
856 if let Some(plan) = cached.as_ref()
857 && plan.quantizer_version == centroids.version
858 && plan.codebook_version == codebook.version
859 && plan.request_fingerprint == request_fingerprint
860 && plan.cluster_ids.len() == effective_nprobe
861 {
862 return Ok(std::sync::Arc::clone(plan));
863 }
864 let plan = std::sync::Arc::new(crate::structures::IvfPqQueryPlan::build(
865 centroids,
866 codebook,
867 query,
868 effective_nprobe,
869 routing,
870 ));
871 *cached = Some(std::sync::Arc::clone(&plan));
872 return Ok(plan);
873 }
874 Ok(std::sync::Arc::new(
875 crate::structures::IvfPqQueryPlan::build(
876 centroids,
877 codebook,
878 query,
879 effective_nprobe,
880 routing,
881 ),
882 ))
883}
884
885fn binary_probe_clusters(
886 quantizer: &crate::structures::BinaryCoarseQuantizer,
887 query: &[u8],
888 nprobe: usize,
889 routing: crate::dsl::IvfRoutingMode,
890 cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
891) -> Result<std::sync::Arc<[u32]>> {
892 let effective_nprobe = nprobe.clamp(1, quantizer.num_clusters as usize);
893 let request_fingerprint = crate::structures::vector::ivf::routing::binary_probe_fingerprint(
894 query,
895 effective_nprobe,
896 routing,
897 );
898 if let Some(cache) = cache {
899 let mut cached = cache
900 .lock()
901 .map_err(|_| Error::Internal("binary IVF probe cache is poisoned".into()))?;
902 if let Some(plan) = cached.as_ref()
903 && plan.quantizer_version == quantizer.version
904 && plan.request_fingerprint == request_fingerprint
905 && plan.cluster_ids.len() == effective_nprobe
906 {
907 return Ok(std::sync::Arc::clone(&plan.cluster_ids));
908 }
909 let plan = quantizer.probe(query, effective_nprobe, routing);
910 let clusters = std::sync::Arc::clone(&plan.cluster_ids);
911 *cached = Some(plan);
912 return Ok(clusters);
913 }
914 Ok(quantizer
915 .probe(query, effective_nprobe, routing)
916 .cluster_ids)
917}
918
919pub struct SegmentReader {
925 meta: SegmentMeta,
926 term_dict: Arc<AsyncSSTableReader<TermInfo>>,
928 postings_handle: FileHandle,
930 store: Arc<AsyncStoreReader>,
932 schema: Arc<Schema>,
933 vector_indexes: FxHashMap<u32, VectorIndex>,
935 flat_vectors: FxHashMap<u32, LazyFlatVectorData>,
937 trained_vectors: Arc<crate::segment::TrainedVectorStructures>,
939 sparse_indexes: FxHashMap<u32, SparseIndex>,
941 bmp_indexes: FxHashMap<u32, BmpIndex>,
943 positions_handle: Option<FileHandle>,
945 fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldReader>,
947 #[cfg(feature = "native")]
952 pin_report: crate::segment::pin::PinReport,
953}
954
955impl SegmentReader {
956 pub async fn open<D: Directory>(
958 dir: &D,
959 segment_id: SegmentId,
960 schema: Arc<Schema>,
961 cache_blocks: usize,
962 ) -> Result<Self> {
963 Self::open_with_cache_blocks(dir, segment_id, schema, cache_blocks, cache_blocks).await
964 }
965
966 pub async fn open_with_cache_blocks<D: Directory>(
972 dir: &D,
973 segment_id: SegmentId,
974 schema: Arc<Schema>,
975 term_cache_blocks: usize,
976 store_cache_blocks: usize,
977 ) -> Result<Self> {
978 let files = SegmentFiles::new(segment_id.0);
979
980 let meta_slice = dir.open_read(&files.meta).await?;
982 let meta_bytes = meta_slice.read_bytes().await?;
983 let meta = SegmentMeta::deserialize(meta_bytes.as_slice())?;
984 debug_assert_eq!(meta.id, segment_id.0);
985
986 let term_dict_handle = dir.open_lazy(&files.term_dict).await?;
988 let term_dict = AsyncSSTableReader::open(term_dict_handle, term_cache_blocks).await?;
989
990 let postings_handle = dir.open_lazy(&files.postings).await?;
992
993 let store_handle = dir.open_lazy(&files.store).await?;
995 let store = AsyncStoreReader::open(store_handle, store_cache_blocks).await?;
996
997 let vectors_data = loader::load_vectors_file(dir, &files, &schema, meta.num_docs).await?;
999 let vector_indexes = vectors_data.indexes;
1000 let flat_vectors = vectors_data.flat_vectors;
1001
1002 #[cfg(feature = "native")]
1007 for (field_id, lazy_flat) in &flat_vectors {
1008 if vector_indexes.contains_key(field_id) {
1009 lazy_flat.advise_random_access();
1010 }
1011 }
1012
1013 let sparse_data = loader::load_sparse_file(dir, &files, meta.num_docs, &schema).await?;
1015 let sparse_indexes = sparse_data.maxscore_indexes;
1016 let bmp_indexes = sparse_data.bmp_indexes;
1017
1018 let positions_handle = loader::open_positions_file(dir, &files, &schema).await?;
1020
1021 let fast_fields = loader::load_fast_fields_file(dir, &files, &schema).await?;
1023
1024 {
1026 let mut parts = vec![format!(
1027 "[segment] loaded {:016x}: docs={}",
1028 segment_id.0, meta.num_docs
1029 )];
1030 if !vector_indexes.is_empty() || !flat_vectors.is_empty() {
1031 parts.push(format!(
1032 "dense: {} ann + {} flat fields",
1033 vector_indexes.len(),
1034 flat_vectors.len()
1035 ));
1036 }
1037 for (field_id, idx) in &sparse_indexes {
1038 parts.push(format!(
1039 "sparse field {}: {} dims, ~{:.1} KB",
1040 field_id,
1041 idx.num_dimensions(),
1042 idx.num_dimensions() as f64 * 24.0 / 1024.0
1043 ));
1044 }
1045 for (field_id, idx) in &bmp_indexes {
1046 parts.push(format!(
1047 "bmp field {}: {} dims, {} blocks",
1048 field_id,
1049 idx.dims(),
1050 idx.num_blocks
1051 ));
1052 }
1053 if !fast_fields.is_empty() {
1054 parts.push(format!("fast: {} fields", fast_fields.len()));
1055 }
1056 log::debug!("{}", parts.join(", "));
1057 }
1058
1059 #[allow(unused_mut)]
1060 let mut reader = Self {
1061 meta,
1062 term_dict: Arc::new(term_dict),
1063 postings_handle,
1064 store: Arc::new(store),
1065 schema,
1066 vector_indexes,
1067 flat_vectors,
1068 trained_vectors: Arc::new(crate::segment::TrainedVectorStructures::default()),
1069 sparse_indexes,
1070 bmp_indexes,
1071 positions_handle,
1072 fast_fields,
1073 #[cfg(feature = "native")]
1074 pin_report: Default::default(),
1075 };
1076
1077 #[cfg(feature = "native")]
1079 reader.apply_pin_policy(&crate::segment::pin::pin_policy().to_owned());
1080
1081 Ok(reader)
1082 }
1083
1084 #[cfg(feature = "native")]
1094 pub(crate) fn apply_pin_policy(&mut self, policy: &crate::segment::pin::PinPolicy) {
1095 use crate::segment::pin::PinReport;
1096
1097 if !policy.is_enabled() {
1098 return;
1099 }
1100 let mut remaining = policy.budget_bytes;
1101 let mut report = PinReport::default();
1102
1103 for index in self.vector_indexes.values_mut() {
1105 index.pin_lookup_directory(policy.mode, &mut remaining, &mut report);
1106 }
1107 for bmp in self.bmp_indexes.values_mut() {
1109 bmp.pin_block_starts(policy.mode, &mut remaining, &mut report);
1110 }
1111 for sparse in self.sparse_indexes.values_mut() {
1113 sparse.pin_skip_section(policy.mode, &mut remaining, &mut report);
1114 }
1115 for flat in self.flat_vectors.values_mut() {
1117 flat.pin_doc_ids(policy.mode, &mut remaining, &mut report);
1118 }
1119 for bmp in self.bmp_indexes.values_mut() {
1120 bmp.pin_doc_maps(policy.mode, &mut remaining, &mut report);
1121 }
1122 for bmp in self.bmp_indexes.values_mut() {
1124 bmp.pin_sb_grid(policy.mode, &mut remaining, &mut report);
1125 }
1126
1127 if report.skipped_budget_bytes > 0 || report.failed_bytes > 0 {
1128 log::warn!(
1129 "[pin] segment {:016x}: pinned {}/{} bytes (budget skipped {}, mlock failed {}) — raise HERMES_PIN_METADATA_BUDGET_MB or RLIMIT_MEMLOCK for full coverage",
1130 self.meta.id,
1131 report.pinned_bytes,
1132 report.intended_bytes,
1133 report.skipped_budget_bytes,
1134 report.failed_bytes,
1135 );
1136 } else if report.pinned_bytes > 0 {
1137 log::info!(
1138 "[pin] segment {:016x}: pinned {} bytes of hot metadata ({:?})",
1139 self.meta.id,
1140 report.pinned_bytes,
1141 policy.mode,
1142 );
1143 }
1144 self.pin_report = report;
1145 }
1146
1147 pub fn meta(&self) -> &SegmentMeta {
1153 &self.meta
1154 }
1155
1156 pub fn num_docs(&self) -> u32 {
1157 self.meta.num_docs
1158 }
1159
1160 pub fn avg_field_len(&self, field: Field) -> f32 {
1162 self.meta.avg_field_len(field)
1163 }
1164
1165 pub fn schema(&self) -> &Schema {
1166 &self.schema
1167 }
1168
1169 pub fn sparse_indexes(&self) -> &FxHashMap<u32, SparseIndex> {
1171 &self.sparse_indexes
1172 }
1173
1174 pub fn sparse_index(&self, field: Field) -> Option<&SparseIndex> {
1176 self.sparse_indexes.get(&field.0)
1177 }
1178
1179 pub fn bmp_index(&self, field: Field) -> Option<&BmpIndex> {
1181 self.bmp_indexes.get(&field.0)
1182 }
1183
1184 pub fn bmp_indexes(&self) -> &FxHashMap<u32, BmpIndex> {
1186 &self.bmp_indexes
1187 }
1188
1189 pub fn vector_indexes(&self) -> &FxHashMap<u32, VectorIndex> {
1191 &self.vector_indexes
1192 }
1193
1194 pub fn flat_vectors(&self) -> &FxHashMap<u32, LazyFlatVectorData> {
1196 &self.flat_vectors
1197 }
1198
1199 pub fn fast_field(
1201 &self,
1202 field_id: u32,
1203 ) -> Option<&crate::structures::fast_field::FastFieldReader> {
1204 self.fast_fields.get(&field_id)
1205 }
1206
1207 pub fn fast_fields(&self) -> &FxHashMap<u32, crate::structures::fast_field::FastFieldReader> {
1209 &self.fast_fields
1210 }
1211
1212 pub fn term_dict_stats(&self) -> SSTableStats {
1214 self.term_dict.stats()
1215 }
1216
1217 pub fn memory_stats(&self) -> SegmentMemoryStats {
1219 let term_dict_stats = self.term_dict.stats();
1220
1221 let term_dict_cache_bytes = self.term_dict.cached_bytes();
1225 let store_cache_bytes = self.store.cached_bytes();
1226
1227 let sparse_index_bytes: usize = self
1229 .sparse_indexes
1230 .values()
1231 .map(|s| s.estimated_memory_bytes())
1232 .sum::<usize>()
1233 + self
1234 .bmp_indexes
1235 .values()
1236 .map(|b| b.estimated_memory_bytes())
1237 .sum::<usize>();
1238
1239 let dense_index_bytes: usize = self
1242 .vector_indexes
1243 .values()
1244 .map(|v| v.estimated_memory_bytes())
1245 .sum();
1246
1247 #[cfg(feature = "native")]
1248 let (pinned_metadata_bytes, pin_intended_bytes) =
1249 (self.pin_report.pinned_bytes, self.pin_report.intended_bytes);
1250 #[cfg(not(feature = "native"))]
1251 let (pinned_metadata_bytes, pin_intended_bytes) = (0u64, 0u64);
1252
1253 SegmentMemoryStats {
1254 segment_id: self.meta.id,
1255 num_docs: self.meta.num_docs,
1256 term_dict_cache_bytes,
1257 store_cache_bytes,
1258 sparse_index_bytes,
1259 dense_index_bytes,
1260 bloom_filter_bytes: term_dict_stats.bloom_filter_size,
1261 pinned_metadata_bytes,
1262 pin_intended_bytes,
1263 }
1264 }
1265
1266 pub async fn get_postings(
1271 &self,
1272 field: Field,
1273 term: &[u8],
1274 ) -> Result<Option<BlockPostingList>> {
1275 log::debug!(
1276 "SegmentReader::get_postings field={} term_len={}",
1277 field.0,
1278 term.len()
1279 );
1280
1281 let mut key = Vec::with_capacity(4 + term.len());
1283 key.extend_from_slice(&field.0.to_le_bytes());
1284 key.extend_from_slice(term);
1285
1286 let term_info = match self.term_dict.get(&key).await? {
1288 Some(info) => {
1289 log::debug!("SegmentReader::get_postings found term_info");
1290 info
1291 }
1292 None => {
1293 log::debug!("SegmentReader::get_postings term not found");
1294 return Ok(None);
1295 }
1296 };
1297
1298 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1300 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1302 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1303 posting_list.push(doc_id, tf);
1304 }
1305 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
1306 return Ok(Some(block_list));
1307 }
1308
1309 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
1311 Error::Corruption("TermInfo has neither inline nor external data".to_string())
1312 })?;
1313
1314 let range = checked_file_range(
1315 posting_offset,
1316 posting_len,
1317 self.postings_handle.len(),
1318 "posting",
1319 )?;
1320 let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
1321 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
1322
1323 Ok(Some(block_list))
1324 }
1325
1326 pub async fn get_prefix_postings(
1328 &self,
1329 field: Field,
1330 prefix: &[u8],
1331 ) -> Result<Vec<BlockPostingList>> {
1332 if prefix.is_empty() {
1333 return Err(Error::Query("prefix must not be empty".into()));
1334 }
1335 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
1337 key_prefix.extend_from_slice(&field.0.to_le_bytes());
1338 key_prefix.extend_from_slice(prefix);
1339
1340 let (entries, truncated) = self
1341 .term_dict
1342 .prefix_scan_limited(&key_prefix, MAX_PREFIX_TERMS)
1343 .await?;
1344 if truncated {
1345 return Err(Error::Query(format!(
1346 "prefix expands to more than {MAX_PREFIX_TERMS} terms"
1347 )));
1348 }
1349 let posting_count: u64 = entries
1350 .iter()
1351 .map(|(_, term_info)| term_info.doc_freq() as u64)
1352 .sum();
1353 if posting_count > MAX_PREFIX_POSTINGS {
1354 return Err(Error::Query(format!(
1355 "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
1356 )));
1357 }
1358 let mut results = Vec::with_capacity(entries.len());
1359
1360 for (_key, term_info) in entries {
1361 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1362 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1363 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1364 posting_list.push(doc_id, tf);
1365 }
1366 results.push(BlockPostingList::from_posting_list(&posting_list)?);
1367 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
1368 let range = checked_file_range(
1369 posting_offset,
1370 posting_len,
1371 self.postings_handle.len(),
1372 "prefix posting",
1373 )?;
1374 let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
1375 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
1376 }
1377 }
1378
1379 Ok(results)
1380 }
1381
1382 pub async fn doc(&self, local_doc_id: DocId) -> Result<Option<Document>> {
1387 self.doc_with_fields(local_doc_id, None).await
1388 }
1389
1390 pub async fn doc_with_fields(
1396 &self,
1397 local_doc_id: DocId,
1398 fields: Option<&rustc_hash::FxHashSet<u32>>,
1399 ) -> Result<Option<Document>> {
1400 let mut doc = match fields {
1401 Some(set) => {
1402 let field_ids: Vec<u32> = set.iter().copied().collect();
1403 match self
1404 .store
1405 .get_fields(local_doc_id, &self.schema, &field_ids)
1406 .await
1407 {
1408 Ok(Some(d)) => d,
1409 Ok(None) => return Ok(None),
1410 Err(e) => return Err(Error::from(e)),
1411 }
1412 }
1413 None => match self.store.get(local_doc_id, &self.schema).await {
1414 Ok(Some(d)) => d,
1415 Ok(None) => return Ok(None),
1416 Err(e) => return Err(Error::from(e)),
1417 },
1418 };
1419
1420 for (&field_id, lazy_flat) in &self.flat_vectors {
1422 if let Some(set) = fields
1424 && !set.contains(&field_id)
1425 {
1426 continue;
1427 }
1428
1429 let is_binary = lazy_flat.quantization == DenseVectorQuantization::Binary;
1430 let (start, entries) = lazy_flat.flat_indexes_for_doc(local_doc_id);
1431 for (j, &(_doc_id, _ordinal)) in entries.iter().enumerate() {
1432 let flat_idx = start + j;
1433 if is_binary {
1434 let vbs = lazy_flat.vector_byte_size();
1435 let mut raw = vec![0u8; vbs];
1436 match lazy_flat.read_vector_raw_into(flat_idx, &mut raw).await {
1437 Ok(()) => {
1438 doc.add_binary_dense_vector(Field(field_id), raw);
1439 }
1440 Err(e) => {
1441 log::warn!("Failed to hydrate binary vector field {}: {}", field_id, e);
1442 }
1443 }
1444 } else {
1445 match lazy_flat.get_vector(flat_idx).await {
1446 Ok(vec) => {
1447 doc.add_dense_vector(Field(field_id), vec);
1448 }
1449 Err(e) => {
1450 log::warn!("Failed to hydrate vector field {}: {}", field_id, e);
1451 }
1452 }
1453 }
1454 }
1455 }
1456
1457 Ok(Some(doc))
1458 }
1459
1460 pub async fn prefetch_terms(
1462 &self,
1463 field: Field,
1464 start_term: &[u8],
1465 end_term: &[u8],
1466 ) -> Result<()> {
1467 let mut start_key = Vec::with_capacity(4 + start_term.len());
1468 start_key.extend_from_slice(&field.0.to_le_bytes());
1469 start_key.extend_from_slice(start_term);
1470
1471 let mut end_key = Vec::with_capacity(4 + end_term.len());
1472 end_key.extend_from_slice(&field.0.to_le_bytes());
1473 end_key.extend_from_slice(end_term);
1474
1475 self.term_dict.prefetch_range(&start_key, &end_key).await?;
1476 Ok(())
1477 }
1478
1479 pub fn store_has_dict(&self) -> bool {
1481 self.store.has_dict()
1482 }
1483
1484 pub fn store(&self) -> &super::store::AsyncStoreReader {
1486 &self.store
1487 }
1488
1489 pub fn store_raw_blocks(&self) -> Vec<RawStoreBlock> {
1491 self.store.raw_blocks()
1492 }
1493
1494 pub fn store_data_slice(&self) -> &FileHandle {
1496 self.store.data_slice()
1497 }
1498
1499 pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
1501 self.term_dict.all_entries().await.map_err(Error::from)
1502 }
1503
1504 pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
1509 let entries = self.term_dict.all_entries().await?;
1510 let mut result = Vec::with_capacity(entries.len());
1511
1512 for (key, term_info) in entries {
1513 if key.len() > 4 {
1515 let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
1516 let term_bytes = &key[4..];
1517 if let Ok(term_str) = std::str::from_utf8(term_bytes) {
1518 result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
1519 }
1520 }
1521 }
1522
1523 Ok(result)
1524 }
1525
1526 pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
1528 self.term_dict.iter()
1529 }
1530
1531 pub async fn prefetch_term_dict(&self) -> crate::Result<()> {
1535 self.term_dict
1536 .prefetch_all_data_bulk()
1537 .await
1538 .map_err(crate::Error::from)
1539 }
1540
1541 pub async fn read_postings(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
1543 let range = checked_file_range(offset, len, self.postings_handle.len(), "posting")?;
1544 let bytes = self.postings_handle.read_bytes_range(range).await?;
1545 Ok(bytes.to_vec())
1546 }
1547
1548 pub async fn read_position_bytes(&self, offset: u64, len: u64) -> Result<Option<Vec<u8>>> {
1550 let handle = match &self.positions_handle {
1551 Some(h) => h,
1552 None => return Ok(None),
1553 };
1554 let range = checked_file_range(offset, len, handle.len(), "position")?;
1555 let bytes = handle.read_bytes_range(range).await?;
1556 Ok(Some(bytes.to_vec()))
1557 }
1558
1559 pub fn has_positions_file(&self) -> bool {
1561 self.positions_handle.is_some()
1562 }
1563
1564 fn validate_dense_search_request(
1568 &self,
1569 field: Field,
1570 query: &[f32],
1571 nprobe: usize,
1572 rerank_factor: f32,
1573 combiner: crate::query::MultiValueCombiner,
1574 ) -> Result<DenseSearchParams> {
1575 let entry = self
1576 .schema
1577 .get_field_entry(field)
1578 .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
1579 if entry.field_type != crate::dsl::FieldType::DenseVector {
1580 return Err(Error::InvalidFieldType {
1581 expected: "dense_vector".to_string(),
1582 got: format!("{:?}", entry.field_type),
1583 });
1584 }
1585 let config = entry.dense_vector_config.as_ref().ok_or_else(|| {
1586 Error::Schema(format!(
1587 "dense vector field '{}' has no dense vector configuration",
1588 entry.name
1589 ))
1590 })?;
1591
1592 if query.is_empty() {
1593 return Err(Error::Query(format!(
1594 "dense query vector for field '{}' must not be empty",
1595 entry.name
1596 )));
1597 }
1598 if query.len() != config.dim {
1599 return Err(Error::Query(format!(
1600 "dense query vector dimension {} does not match field '{}' dimension {}",
1601 query.len(),
1602 entry.name,
1603 config.dim
1604 )));
1605 }
1606 if let Some((index, value)) = query
1607 .iter()
1608 .enumerate()
1609 .find(|(_, value)| !value.is_finite())
1610 {
1611 return Err(Error::Query(format!(
1612 "dense query vector for field '{}' contains non-finite value {value} at index {index}",
1613 entry.name
1614 )));
1615 }
1616
1617 let nprobe = match (nprobe, config.nprobe) {
1620 (0, 0) => 32,
1621 (0, schema_nprobe) => schema_nprobe,
1622 (query_nprobe, _) => query_nprobe,
1623 };
1624 if nprobe > MAX_DENSE_NPROBE {
1625 return Err(Error::Query(format!(
1626 "dense nprobe must be at most {MAX_DENSE_NPROBE}, got {nprobe}"
1627 )));
1628 }
1629
1630 checked_dense_fetch_k(0, rerank_factor)?;
1633 combiner.validate().map_err(Error::Query)?;
1634
1635 Ok(DenseSearchParams {
1636 dim: config.dim,
1637 nprobe,
1638 unit_norm: config.unit_norm,
1639 })
1640 }
1641
1642 fn validate_binary_search_request(&self, field: Field, query: &[u8]) -> Result<usize> {
1643 let entry = self
1644 .schema
1645 .get_field_entry(field)
1646 .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
1647 if entry.field_type != crate::dsl::FieldType::BinaryDenseVector {
1648 return Err(Error::InvalidFieldType {
1649 expected: "binary_dense_vector".to_string(),
1650 got: format!("{:?}", entry.field_type),
1651 });
1652 }
1653 let config = entry.binary_dense_vector_config.as_ref().ok_or_else(|| {
1654 Error::Schema(format!(
1655 "binary dense vector field '{}' has no configuration",
1656 entry.name
1657 ))
1658 })?;
1659 if config.dim == 0 || !config.dim.is_multiple_of(8) {
1660 return Err(Error::Schema(format!(
1661 "binary dense vector field '{}' has invalid dimension {}",
1662 entry.name, config.dim
1663 )));
1664 }
1665 if query.len() != config.byte_len() {
1666 return Err(Error::Query(format!(
1667 "binary query byte length {} does not match field '{}' byte length {}",
1668 query.len(),
1669 entry.name,
1670 config.byte_len()
1671 )));
1672 }
1673 Ok(config.dim)
1674 }
1675
1676 fn score_quantized_batch(
1682 query: &[f32],
1683 raw: &[u8],
1684 quant: crate::dsl::DenseVectorQuantization,
1685 dim: usize,
1686 scores: &mut [f32],
1687 unit_norm: bool,
1688 ) -> Result<()> {
1689 use crate::dsl::DenseVectorQuantization;
1690 use crate::structures::simd;
1691
1692 if query.len() != dim {
1693 return Err(Error::Query(format!(
1694 "dense SIMD query dimension {} does not match vector dimension {dim}",
1695 query.len()
1696 )));
1697 }
1698 let element_size = match quant {
1699 DenseVectorQuantization::F32 => std::mem::size_of::<f32>(),
1700 DenseVectorQuantization::F16 => std::mem::size_of::<u16>(),
1701 DenseVectorQuantization::UInt8 => 1,
1702 DenseVectorQuantization::Binary => {
1703 return Err(Error::InvalidFieldType {
1704 expected: "non-binary dense vector".to_string(),
1705 got: "binary dense vector".to_string(),
1706 });
1707 }
1708 };
1709 let required_bytes = scores
1710 .len()
1711 .checked_mul(dim)
1712 .and_then(|elements| elements.checked_mul(element_size))
1713 .ok_or_else(|| Error::Corruption("dense vector batch byte length overflow".into()))?;
1714 if raw.len() < required_bytes {
1715 return Err(Error::Corruption(format!(
1716 "dense vector batch is truncated: need {required_bytes} bytes, got {}",
1717 raw.len()
1718 )));
1719 }
1720 if quant == DenseVectorQuantization::F16
1721 && required_bytes > 0
1722 && !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<u16>())
1723 {
1724 return Err(Error::Corruption(
1725 "f16 vector data is not 2-byte aligned".to_string(),
1726 ));
1727 }
1728
1729 match (quant, unit_norm) {
1730 (DenseVectorQuantization::F32, false) => {
1731 let num_floats = scores.len() * dim;
1732 if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
1733 return Err(Error::Corruption(
1734 "f32 vector data is not 4-byte aligned".to_string(),
1735 ));
1736 }
1737 let vectors: &[f32] =
1738 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
1739 simd::batch_cosine_scores(query, vectors, dim, scores);
1740 }
1741 (DenseVectorQuantization::F32, true) => {
1742 let num_floats = scores.len() * dim;
1743 if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
1744 return Err(Error::Corruption(
1745 "f32 vector data is not 4-byte aligned".to_string(),
1746 ));
1747 }
1748 let vectors: &[f32] =
1749 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
1750 simd::batch_dot_scores(query, vectors, dim, scores);
1751 }
1752 (DenseVectorQuantization::F16, false) => {
1753 simd::batch_cosine_scores_f16(query, raw, dim, scores);
1754 }
1755 (DenseVectorQuantization::F16, true) => {
1756 simd::batch_dot_scores_f16(query, raw, dim, scores);
1757 }
1758 (DenseVectorQuantization::UInt8, false) => {
1759 simd::batch_cosine_scores_u8(query, raw, dim, scores);
1760 }
1761 (DenseVectorQuantization::UInt8, true) => {
1762 simd::batch_dot_scores_u8(query, raw, dim, scores);
1763 }
1764 (DenseVectorQuantization::Binary, _) => unreachable!("validated above"),
1765 }
1766 Ok(())
1767 }
1768
1769 pub async fn search_dense_vector(
1775 &self,
1776 field: Field,
1777 query: &[f32],
1778 k: usize,
1779 nprobe: usize,
1780 rerank_factor: f32,
1781 combiner: crate::query::MultiValueCombiner,
1782 ) -> Result<Vec<VectorSearchResult>> {
1783 self.search_dense_vector_impl(field, query, k, nprobe, rerank_factor, combiner, None)
1784 .await
1785 }
1786
1787 #[allow(clippy::too_many_arguments)]
1788 pub(crate) async fn search_dense_vector_with_probe_cache(
1789 &self,
1790 field: Field,
1791 query: &[f32],
1792 k: usize,
1793 nprobe: usize,
1794 rerank_factor: f32,
1795 combiner: crate::query::MultiValueCombiner,
1796 probe_cache: &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
1797 ) -> Result<Vec<VectorSearchResult>> {
1798 self.search_dense_vector_impl(
1799 field,
1800 query,
1801 k,
1802 nprobe,
1803 rerank_factor,
1804 combiner,
1805 Some(probe_cache),
1806 )
1807 .await
1808 }
1809
1810 #[allow(clippy::too_many_arguments)]
1811 async fn search_dense_vector_impl(
1812 &self,
1813 field: Field,
1814 query: &[f32],
1815 k: usize,
1816 nprobe: usize,
1817 rerank_factor: f32,
1818 combiner: crate::query::MultiValueCombiner,
1819 probe_cache: Option<
1820 &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
1821 >,
1822 ) -> Result<Vec<VectorSearchResult>> {
1823 let params =
1824 self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
1825 let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
1826 if k == 0 {
1827 return Ok(Vec::new());
1828 }
1829
1830 let ann_index = self.vector_indexes.get(&field.0);
1831 let lazy_flat = self.flat_vectors.get(&field.0);
1832 if ann_index.is_none() && lazy_flat.is_none() {
1834 return Ok(Vec::new());
1835 }
1836
1837 if ann_index.is_some() && lazy_flat.is_none() {
1838 return Err(Error::Corruption(format!(
1839 "dense ANN field {} is missing flat vector storage",
1840 field.0
1841 )));
1842 }
1843
1844 if let Some(flat) = lazy_flat
1845 && flat.dim != params.dim
1846 {
1847 return Err(Error::Corruption(format!(
1848 "dense vector field {} has schema dimension {} but flat storage dimension {}",
1849 field.0, params.dim, flat.dim
1850 )));
1851 }
1852
1853 let t0 = std::time::Instant::now();
1855 let mut flat_results = None;
1856 let results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
1857 match index {
1859 VectorIndex::IvfPq(lazy) => {
1860 let index = lazy.get();
1861 let codebook =
1862 self.trained_vectors
1863 .codebooks
1864 .get(&field.0)
1865 .ok_or_else(|| {
1866 Error::Schema(format!(
1867 "IVF-PQ index requires a global codebook for field {}",
1868 field.0
1869 ))
1870 })?;
1871 let centroids =
1872 self.trained_vectors
1873 .centroids
1874 .get(&field.0)
1875 .ok_or_else(|| {
1876 Error::Schema(format!(
1877 "IVF-PQ index requires coarse centroids for field {}",
1878 field.0
1879 ))
1880 })?;
1881 validate_coarse_centroids(centroids, params.dim)?;
1882 let routing = self
1883 .schema
1884 .get_field_entry(field)
1885 .and_then(|entry| entry.dense_vector_config.as_ref())
1886 .map_or(crate::dsl::IvfRoutingMode::Auto, |config| {
1887 config.ivf_routing
1888 });
1889 validate_ivf_pq_ann(index, centroids, codebook, params.dim, routing)?;
1890 let query_plan = float_query_plan(
1891 centroids,
1892 codebook,
1893 query,
1894 params.nprobe,
1895 routing,
1896 probe_cache,
1897 )?;
1898 let flat = lazy_flat.expect("ANN/flat pairing validated above");
1899 index
1900 .search_ivf_pq_distinct(
1901 fetch_k.min(flat.num_docs_with_vectors()),
1902 &query_plan,
1903 )
1904 .map_err(|error| {
1905 Error::Corruption(format!(
1906 "invalid IVF-PQ payload for field {}: {error}",
1907 field.0,
1908 ))
1909 })?
1910 .into_iter()
1911 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
1912 .collect()
1913 }
1914 VectorIndex::BinaryIvf(_) => {
1915 Vec::new()
1917 }
1918 }
1919 } else if let Some(lazy_flat) = lazy_flat {
1920 log::debug!(
1924 "[search_dense] field {}: brute-force on {} vectors (dim={}, quant={:?})",
1925 field.0,
1926 lazy_flat.num_vectors,
1927 lazy_flat.dim,
1928 lazy_flat.quantization
1929 );
1930 let dim = lazy_flat.dim;
1931 let n = lazy_flat.num_vectors;
1932 let quant = lazy_flat.quantization;
1933 let batch_len =
1934 bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
1935 let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
1936 let mut scores = vec![0f32; batch_len];
1937
1938 for batch_start in (0..n).step_by(batch_len) {
1939 let batch_count = batch_len.min(n - batch_start);
1940 let batch_bytes = lazy_flat
1941 .read_vectors_batch(batch_start, batch_count)
1942 .await
1943 .map_err(crate::Error::Io)?;
1944 let raw = batch_bytes.as_slice();
1945
1946 Self::score_quantized_batch(
1947 query,
1948 raw,
1949 quant,
1950 dim,
1951 &mut scores[..batch_count],
1952 params.unit_norm,
1953 )?;
1954
1955 for (i, &score) in scores.iter().enumerate().take(batch_count) {
1956 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1957 collector.push(doc_id, ordinal, score);
1958 }
1959 }
1960
1961 flat_results = Some(collector.into_results());
1962 Vec::new()
1963 } else {
1964 return Ok(Vec::new());
1965 };
1966 let l1_elapsed = t0.elapsed();
1967 {
1968 let kind = match ann_index {
1969 Some(VectorIndex::IvfPq(_)) => "ivf_pq",
1970 Some(VectorIndex::BinaryIvf(_)) => "binary_ivf",
1971 None => "flat",
1972 };
1973 crate::observe::dense_l1(
1974 self.schema.index_label(),
1975 self.schema.get_field_name(field).unwrap_or("?"),
1976 kind,
1977 l1_elapsed.as_secs_f64(),
1978 flat_results.as_ref().map_or(results.len(), Vec::len),
1979 );
1980 }
1981 log::debug!(
1982 "[search_dense] field {}: L1 returned {} candidates in {:.1}ms",
1983 field.0,
1984 flat_results.as_ref().map_or(results.len(), Vec::len),
1985 l1_elapsed.as_secs_f64() * 1000.0
1986 );
1987
1988 if let Some(results) = flat_results {
1989 return Ok(results);
1990 }
1991
1992 if ann_index.is_some()
1995 && !results.is_empty()
1996 && let Some(lazy_flat) = lazy_flat
1997 {
1998 let t_rerank = std::time::Instant::now();
1999 let vbs = lazy_flat.vector_byte_size();
2000 let (reranked, stats) = exact_score_dense_candidate_documents(
2001 &results,
2002 lazy_flat,
2003 query,
2004 params.unit_norm,
2005 combiner,
2006 k,
2007 )
2008 .await?;
2009
2010 crate::observe::dense_rerank(
2011 self.schema.index_label(),
2012 self.schema.get_field_name(field).unwrap_or("?"),
2013 t_rerank.elapsed().as_secs_f64(),
2014 stats.resolve_elapsed.as_secs_f64(),
2015 stats.read_elapsed.as_secs_f64(),
2016 stats.vector_count,
2017 );
2018 log::debug!(
2019 "[search_dense] field {}: rerank {} vectors (dim={}, quant={:?}, {}B/vec): resolve={:.1}ms read={:.1}ms score={:.1}ms",
2020 field.0,
2021 stats.vector_count,
2022 lazy_flat.dim,
2023 lazy_flat.quantization,
2024 vbs,
2025 stats.resolve_elapsed.as_secs_f64() * 1000.0,
2026 stats.read_elapsed.as_secs_f64() * 1000.0,
2027 stats.score_elapsed.as_secs_f64() * 1000.0,
2028 );
2029
2030 log::debug!(
2031 "[search_dense] field {}: rerank total={:.1}ms",
2032 field.0,
2033 t_rerank.elapsed().as_secs_f64() * 1000.0
2034 );
2035 return Ok(reranked);
2036 }
2037
2038 Ok(combine_grouped_ordinal_results(results, combiner, k))
2039 }
2040
2041 async fn search_binary_dense_vector_impl(
2046 &self,
2047 field: Field,
2048 query: &[u8],
2049 k: usize,
2050 combiner: crate::query::MultiValueCombiner,
2051 probe_cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
2052 ) -> Result<Vec<VectorSearchResult>> {
2053 let schema_dim = self.validate_binary_search_request(field, query)?;
2054 combiner.validate().map_err(Error::Query)?;
2055 if k == 0 {
2056 return Ok(Vec::new());
2057 }
2058 let t0 = crate::observe::Timer::start();
2059 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0) {
2060 let ivf = lazy.get();
2061 let config = self
2062 .schema
2063 .get_field_entry(field)
2064 .and_then(|entry| entry.binary_dense_vector_config.as_ref())
2065 .ok_or_else(|| {
2066 Error::Schema(format!(
2067 "binary IVF field {} has no schema configuration",
2068 field.0
2069 ))
2070 })?;
2071 let quantizer = self
2072 .trained_vectors
2073 .binary_quantizers
2074 .get(&field.0)
2075 .ok_or_else(|| {
2076 Error::Schema(format!(
2077 "global binary IVF field {} has no loaded quantizer",
2078 field.0
2079 ))
2080 })?;
2081 validate_binary_ann(ivf, quantizer, config, schema_dim, field)?;
2082 let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
2083 Error::Corruption(format!(
2084 "global binary IVF field {} is missing flat vector storage",
2085 field.0
2086 ))
2087 })?;
2088 let clusters = binary_probe_clusters(
2089 quantizer,
2090 query,
2091 config.nprobe,
2092 config.ivf_routing,
2093 probe_cache,
2094 )?;
2095 let single_valued = flat.num_vectors == flat.num_docs_with_vectors();
2096 let candidate_docs = k.min(flat.num_docs_with_vectors());
2097 let ann_results = if single_valued {
2098 ivf.search_binary_clusters::<false>(query, candidate_docs, &clusters)
2099 } else {
2100 ivf.search_binary_clusters::<true>(query, candidate_docs, &clusters)
2101 }
2102 .map_err(|error| {
2103 Error::Corruption(format!(
2104 "invalid binary IVF payload for field {}: {error}",
2105 field.0,
2106 ))
2107 })?;
2108 let results = exact_score_binary_candidate_documents(
2109 &ann_results,
2110 flat,
2111 query,
2112 schema_dim,
2113 combiner,
2114 k,
2115 )
2116 .await?;
2117 crate::observe::dense_l1(
2118 self.schema.index_label(),
2119 self.schema.get_field_name(field).unwrap_or("?"),
2120 "global_binary_ivf",
2121 t0.secs(),
2122 results.len(),
2123 );
2124 return Ok(results);
2125 }
2126 let lazy_flat = match self.flat_vectors.get(&field.0) {
2127 Some(f) => f,
2128 None => return Ok(Vec::new()),
2129 };
2130
2131 let dim_bits = lazy_flat.dim;
2132 let byte_len = lazy_flat.vector_byte_size();
2133 let n = lazy_flat.num_vectors;
2134
2135 if dim_bits != schema_dim {
2136 return Err(Error::Corruption(format!(
2137 "binary vector field {} has schema dimension {} but flat storage dimension {}",
2138 field.0, schema_dim, dim_bits
2139 )));
2140 }
2141
2142 if byte_len != query.len() {
2143 return Err(Error::Schema(format!(
2144 "Binary query vector byte length {} != field byte length {}",
2145 query.len(),
2146 byte_len
2147 )));
2148 }
2149
2150 let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
2151 let mut collector = FlatDocumentCollector::new(k, combiner);
2152 let mut scores = vec![0f32; batch_len];
2153
2154 for batch_start in (0..n).step_by(batch_len) {
2155 let batch_count = batch_len.min(n - batch_start);
2156 let batch_bytes = lazy_flat
2157 .read_vectors_batch(batch_start, batch_count)
2158 .await
2159 .map_err(crate::Error::Io)?;
2160 let raw = batch_bytes.as_slice();
2161
2162 crate::structures::simd::batch_hamming_scores(
2163 query,
2164 raw,
2165 byte_len,
2166 dim_bits,
2167 &mut scores[..batch_count],
2168 );
2169
2170 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2171 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2172 collector.push(doc_id, ordinal, score);
2173 }
2174 }
2175
2176 let results = collector.into_results();
2177
2178 crate::observe::dense_l1(
2179 self.schema.index_label(),
2180 self.schema.get_field_name(field).unwrap_or("?"),
2181 "binary_flat",
2182 t0.secs(),
2183 results.len(),
2184 );
2185 Ok(results)
2186 }
2187
2188 pub async fn search_binary_dense_vector(
2189 &self,
2190 field: Field,
2191 query: &[u8],
2192 k: usize,
2193 combiner: crate::query::MultiValueCombiner,
2194 ) -> Result<Vec<VectorSearchResult>> {
2195 self.search_binary_dense_vector_impl(field, query, k, combiner, None)
2196 .await
2197 }
2198
2199 pub(crate) async fn search_binary_dense_vector_with_probe_cache(
2200 &self,
2201 field: Field,
2202 query: &[u8],
2203 k: usize,
2204 combiner: crate::query::MultiValueCombiner,
2205 probe_cache: &std::sync::Mutex<Option<crate::structures::IvfProbePlan>>,
2206 ) -> Result<Vec<VectorSearchResult>> {
2207 self.search_binary_dense_vector_impl(field, query, k, combiner, Some(probe_cache))
2208 .await
2209 }
2210
2211 pub fn coarse_centroids(&self, field_id: u32) -> Option<&Arc<CoarseCentroids>> {
2213 self.trained_vectors.centroids.get(&field_id)
2214 }
2215
2216 pub fn set_trained_vectors(
2217 &mut self,
2218 trained_vectors: Arc<crate::segment::TrainedVectorStructures>,
2219 ) {
2220 self.trained_vectors = trained_vectors;
2221 }
2222
2223 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
2225 self.vector_indexes.get(&field.0)
2226 }
2227
2228 pub async fn get_positions(
2233 &self,
2234 field: Field,
2235 term: &[u8],
2236 ) -> Result<Option<crate::structures::PositionPostingList>> {
2237 let handle = match &self.positions_handle {
2239 Some(h) => h,
2240 None => return Ok(None),
2241 };
2242
2243 let mut key = Vec::with_capacity(4 + term.len());
2245 key.extend_from_slice(&field.0.to_le_bytes());
2246 key.extend_from_slice(term);
2247
2248 let term_info = match self.term_dict.get(&key).await? {
2250 Some(info) => info,
2251 None => return Ok(None),
2252 };
2253
2254 let (offset, length) = match term_info.position_info() {
2256 Some((o, l)) => (o, l),
2257 None => return Ok(None),
2258 };
2259
2260 let range = checked_file_range(offset, length, handle.len(), "position list")?;
2264 let slice = handle.slice(range);
2265 let data = slice.read_bytes().await?;
2266
2267 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2269
2270 Ok(Some(pos_list))
2271 }
2272
2273 pub fn has_positions(&self, field: Field) -> bool {
2275 if let Some(entry) = self.schema.get_field_entry(field) {
2277 entry.positions.is_some()
2278 } else {
2279 false
2280 }
2281 }
2282}
2283
2284#[cfg(feature = "sync")]
2286impl SegmentReader {
2287 pub fn get_postings_sync(&self, field: Field, term: &[u8]) -> Result<Option<BlockPostingList>> {
2289 let mut key = Vec::with_capacity(4 + term.len());
2291 key.extend_from_slice(&field.0.to_le_bytes());
2292 key.extend_from_slice(term);
2293
2294 let term_info = match self.term_dict.get_sync(&key)? {
2296 Some(info) => info,
2297 None => return Ok(None),
2298 };
2299
2300 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2302 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2303 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2304 posting_list.push(doc_id, tf);
2305 }
2306 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
2307 return Ok(Some(block_list));
2308 }
2309
2310 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
2312 Error::Corruption("TermInfo has neither inline nor external data".to_string())
2313 })?;
2314
2315 let range = checked_file_range(
2316 posting_offset,
2317 posting_len,
2318 self.postings_handle.len(),
2319 "posting",
2320 )?;
2321 let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
2322 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
2323
2324 Ok(Some(block_list))
2325 }
2326
2327 pub fn get_prefix_postings_sync(
2329 &self,
2330 field: Field,
2331 prefix: &[u8],
2332 ) -> Result<Vec<BlockPostingList>> {
2333 if prefix.is_empty() {
2334 return Err(Error::Query("prefix must not be empty".into()));
2335 }
2336 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
2337 key_prefix.extend_from_slice(&field.0.to_le_bytes());
2338 key_prefix.extend_from_slice(prefix);
2339
2340 let (entries, truncated) = self
2341 .term_dict
2342 .prefix_scan_limited_sync(&key_prefix, MAX_PREFIX_TERMS)?;
2343 if truncated {
2344 return Err(Error::Query(format!(
2345 "prefix expands to more than {MAX_PREFIX_TERMS} terms"
2346 )));
2347 }
2348 let posting_count: u64 = entries
2349 .iter()
2350 .map(|(_, term_info)| term_info.doc_freq() as u64)
2351 .sum();
2352 if posting_count > MAX_PREFIX_POSTINGS {
2353 return Err(Error::Query(format!(
2354 "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
2355 )));
2356 }
2357 let mut results = Vec::with_capacity(entries.len());
2358
2359 for (_key, term_info) in entries {
2360 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2361 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2362 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2363 posting_list.push(doc_id, tf);
2364 }
2365 results.push(BlockPostingList::from_posting_list(&posting_list)?);
2366 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
2367 let range = checked_file_range(
2368 posting_offset,
2369 posting_len,
2370 self.postings_handle.len(),
2371 "prefix posting",
2372 )?;
2373 let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
2374 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
2375 }
2376 }
2377
2378 Ok(results)
2379 }
2380
2381 pub fn get_positions_sync(
2383 &self,
2384 field: Field,
2385 term: &[u8],
2386 ) -> Result<Option<crate::structures::PositionPostingList>> {
2387 let handle = match &self.positions_handle {
2388 Some(h) => h,
2389 None => return Ok(None),
2390 };
2391
2392 let mut key = Vec::with_capacity(4 + term.len());
2394 key.extend_from_slice(&field.0.to_le_bytes());
2395 key.extend_from_slice(term);
2396
2397 let term_info = match self.term_dict.get_sync(&key)? {
2399 Some(info) => info,
2400 None => return Ok(None),
2401 };
2402
2403 let (offset, length) = match term_info.position_info() {
2404 Some((o, l)) => (o, l),
2405 None => return Ok(None),
2406 };
2407
2408 let range = checked_file_range(offset, length, handle.len(), "position list")?;
2409 let slice = handle.slice(range);
2410 let data = slice.read_bytes_sync()?;
2411
2412 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2413 Ok(Some(pos_list))
2414 }
2415
2416 pub fn search_dense_vector_sync(
2419 &self,
2420 field: Field,
2421 query: &[f32],
2422 k: usize,
2423 nprobe: usize,
2424 rerank_factor: f32,
2425 combiner: crate::query::MultiValueCombiner,
2426 ) -> Result<Vec<VectorSearchResult>> {
2427 self.search_dense_vector_sync_impl(field, query, k, nprobe, rerank_factor, combiner, None)
2428 }
2429
2430 #[cfg(feature = "sync")]
2431 #[allow(clippy::too_many_arguments)]
2432 pub(crate) fn search_dense_vector_sync_with_probe_cache(
2433 &self,
2434 field: Field,
2435 query: &[f32],
2436 k: usize,
2437 nprobe: usize,
2438 rerank_factor: f32,
2439 combiner: crate::query::MultiValueCombiner,
2440 probe_cache: &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
2441 ) -> Result<Vec<VectorSearchResult>> {
2442 self.search_dense_vector_sync_impl(
2443 field,
2444 query,
2445 k,
2446 nprobe,
2447 rerank_factor,
2448 combiner,
2449 Some(probe_cache),
2450 )
2451 }
2452
2453 #[cfg(feature = "sync")]
2454 #[allow(clippy::too_many_arguments)]
2455 fn search_dense_vector_sync_impl(
2456 &self,
2457 field: Field,
2458 query: &[f32],
2459 k: usize,
2460 nprobe: usize,
2461 rerank_factor: f32,
2462 combiner: crate::query::MultiValueCombiner,
2463 probe_cache: Option<
2464 &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
2465 >,
2466 ) -> Result<Vec<VectorSearchResult>> {
2467 let params =
2468 self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
2469 let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
2470 if k == 0 {
2471 return Ok(Vec::new());
2472 }
2473
2474 let ann_index = self.vector_indexes.get(&field.0);
2475 let lazy_flat = self.flat_vectors.get(&field.0);
2476 if ann_index.is_none() && lazy_flat.is_none() {
2477 return Ok(Vec::new());
2478 }
2479
2480 if ann_index.is_some() && lazy_flat.is_none() {
2481 return Err(Error::Corruption(format!(
2482 "dense ANN field {} is missing flat vector storage",
2483 field.0
2484 )));
2485 }
2486
2487 if let Some(flat) = lazy_flat
2488 && flat.dim != params.dim
2489 {
2490 return Err(Error::Corruption(format!(
2491 "dense vector field {} has schema dimension {} but flat storage dimension {}",
2492 field.0, params.dim, flat.dim
2493 )));
2494 }
2495
2496 let results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
2497 match index {
2499 VectorIndex::IvfPq(lazy) => {
2500 let index = lazy.get();
2501 let codebook =
2502 self.trained_vectors
2503 .codebooks
2504 .get(&field.0)
2505 .ok_or_else(|| {
2506 Error::Schema(format!(
2507 "IVF-PQ index requires a global codebook for field {}",
2508 field.0
2509 ))
2510 })?;
2511 let centroids =
2512 self.trained_vectors
2513 .centroids
2514 .get(&field.0)
2515 .ok_or_else(|| {
2516 Error::Schema(format!(
2517 "IVF-PQ index requires coarse centroids for field {}",
2518 field.0
2519 ))
2520 })?;
2521 validate_coarse_centroids(centroids, params.dim)?;
2522 let routing = self
2523 .schema
2524 .get_field_entry(field)
2525 .and_then(|entry| entry.dense_vector_config.as_ref())
2526 .map_or(crate::dsl::IvfRoutingMode::Auto, |config| {
2527 config.ivf_routing
2528 });
2529 validate_ivf_pq_ann(index, centroids, codebook, params.dim, routing)?;
2530 let query_plan = float_query_plan(
2531 centroids,
2532 codebook,
2533 query,
2534 params.nprobe,
2535 routing,
2536 probe_cache,
2537 )?;
2538 let flat = lazy_flat.expect("ANN/flat pairing validated above");
2539 index
2540 .search_ivf_pq_distinct(
2541 fetch_k.min(flat.num_docs_with_vectors()),
2542 &query_plan,
2543 )
2544 .map_err(|error| {
2545 Error::Corruption(format!(
2546 "invalid IVF-PQ payload for field {}: {error}",
2547 field.0,
2548 ))
2549 })?
2550 .into_iter()
2551 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
2552 .collect()
2553 }
2554 VectorIndex::BinaryIvf(_) => {
2555 Vec::new()
2557 }
2558 }
2559 } else if let Some(lazy_flat) = lazy_flat {
2560 let dim = lazy_flat.dim;
2562 let n = lazy_flat.num_vectors;
2563 let quant = lazy_flat.quantization;
2564 let batch_len =
2565 bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
2566 let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
2567 let mut scores = vec![0f32; batch_len];
2568
2569 for batch_start in (0..n).step_by(batch_len) {
2570 let batch_count = batch_len.min(n - batch_start);
2571 let batch_bytes = lazy_flat
2572 .read_vectors_batch_sync(batch_start, batch_count)
2573 .map_err(crate::Error::Io)?;
2574 let raw = batch_bytes.as_slice();
2575
2576 Self::score_quantized_batch(
2577 query,
2578 raw,
2579 quant,
2580 dim,
2581 &mut scores[..batch_count],
2582 params.unit_norm,
2583 )?;
2584
2585 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2586 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2587 collector.push(doc_id, ordinal, score);
2588 }
2589 }
2590
2591 return Ok(collector.into_results());
2592 } else {
2593 return Ok(Vec::new());
2594 };
2595
2596 if ann_index.is_some()
2598 && !results.is_empty()
2599 && let Some(lazy_flat) = lazy_flat
2600 {
2601 return exact_score_dense_candidate_documents_sync(
2602 &results,
2603 lazy_flat,
2604 query,
2605 params.unit_norm,
2606 combiner,
2607 k,
2608 );
2609 }
2610
2611 Ok(combine_grouped_ordinal_results(results, combiner, k))
2612 }
2613
2614 #[cfg(feature = "sync")]
2619 fn search_binary_dense_vector_sync_impl(
2620 &self,
2621 field: Field,
2622 query: &[u8],
2623 k: usize,
2624 combiner: crate::query::MultiValueCombiner,
2625 probe_cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
2626 ) -> Result<Vec<VectorSearchResult>> {
2627 let schema_dim = self.validate_binary_search_request(field, query)?;
2628 combiner.validate().map_err(Error::Query)?;
2629 if k == 0 {
2630 return Ok(Vec::new());
2631 }
2632 let t0 = crate::observe::Timer::start();
2633 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0) {
2634 let ivf = lazy.get();
2635 let config = self
2636 .schema
2637 .get_field_entry(field)
2638 .and_then(|entry| entry.binary_dense_vector_config.as_ref())
2639 .ok_or_else(|| {
2640 Error::Schema(format!(
2641 "binary IVF field {} has no schema configuration",
2642 field.0
2643 ))
2644 })?;
2645 let quantizer = self
2646 .trained_vectors
2647 .binary_quantizers
2648 .get(&field.0)
2649 .ok_or_else(|| {
2650 Error::Schema(format!(
2651 "global binary IVF field {} has no loaded quantizer",
2652 field.0
2653 ))
2654 })?;
2655 validate_binary_ann(ivf, quantizer, config, schema_dim, field)?;
2656 let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
2657 Error::Corruption(format!(
2658 "global binary IVF field {} is missing flat vector storage",
2659 field.0
2660 ))
2661 })?;
2662 let clusters = binary_probe_clusters(
2663 quantizer,
2664 query,
2665 config.nprobe,
2666 config.ivf_routing,
2667 probe_cache,
2668 )?;
2669 let candidate_docs = k.min(flat.num_docs_with_vectors());
2670 let ann_results = if flat.num_vectors == flat.num_docs_with_vectors() {
2671 ivf.search_binary_clusters::<false>(query, candidate_docs, &clusters)
2672 } else {
2673 ivf.search_binary_clusters::<true>(query, candidate_docs, &clusters)
2674 }
2675 .map_err(|error| {
2676 Error::Corruption(format!(
2677 "invalid binary IVF payload for field {}: {error}",
2678 field.0,
2679 ))
2680 })?;
2681 let results = exact_score_binary_candidate_documents_sync(
2682 &ann_results,
2683 flat,
2684 query,
2685 schema_dim,
2686 combiner,
2687 k,
2688 )?;
2689 crate::observe::dense_l1(
2690 self.schema.index_label(),
2691 self.schema.get_field_name(field).unwrap_or("?"),
2692 "global_binary_ivf",
2693 t0.secs(),
2694 results.len(),
2695 );
2696 return Ok(results);
2697 }
2698 let lazy_flat = match self.flat_vectors.get(&field.0) {
2699 Some(f) => f,
2700 None => return Ok(Vec::new()),
2701 };
2702
2703 let dim_bits = lazy_flat.dim;
2704 let byte_len = lazy_flat.vector_byte_size();
2705 let n = lazy_flat.num_vectors;
2706
2707 if dim_bits != schema_dim {
2708 return Err(Error::Corruption(format!(
2709 "binary vector field {} has schema dimension {} but flat storage dimension {}",
2710 field.0, schema_dim, dim_bits
2711 )));
2712 }
2713
2714 if byte_len != query.len() {
2715 return Err(Error::Schema(format!(
2716 "Binary query vector byte length {} != field byte length {}",
2717 query.len(),
2718 byte_len
2719 )));
2720 }
2721
2722 let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
2723 let mut collector = FlatDocumentCollector::new(k, combiner);
2724 let mut scores = vec![0f32; batch_len];
2725
2726 for batch_start in (0..n).step_by(batch_len) {
2727 let batch_count = batch_len.min(n - batch_start);
2728 let batch_bytes = lazy_flat
2729 .read_vectors_batch_sync(batch_start, batch_count)
2730 .map_err(crate::Error::Io)?;
2731 let raw = batch_bytes.as_slice();
2732
2733 crate::structures::simd::batch_hamming_scores(
2734 query,
2735 raw,
2736 byte_len,
2737 dim_bits,
2738 &mut scores[..batch_count],
2739 );
2740
2741 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2742 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2743 collector.push(doc_id, ordinal, score);
2744 }
2745 }
2746
2747 let results = collector.into_results();
2748
2749 crate::observe::dense_l1(
2750 self.schema.index_label(),
2751 self.schema.get_field_name(field).unwrap_or("?"),
2752 "binary_flat",
2753 t0.secs(),
2754 results.len(),
2755 );
2756 Ok(results)
2757 }
2758
2759 #[cfg(feature = "sync")]
2760 pub fn search_binary_dense_vector_sync(
2761 &self,
2762 field: Field,
2763 query: &[u8],
2764 k: usize,
2765 combiner: crate::query::MultiValueCombiner,
2766 ) -> Result<Vec<VectorSearchResult>> {
2767 self.search_binary_dense_vector_sync_impl(field, query, k, combiner, None)
2768 }
2769
2770 #[cfg(feature = "sync")]
2771 pub(crate) fn search_binary_dense_vector_sync_with_probe_cache(
2772 &self,
2773 field: Field,
2774 query: &[u8],
2775 k: usize,
2776 combiner: crate::query::MultiValueCombiner,
2777 probe_cache: &std::sync::Mutex<Option<crate::structures::IvfProbePlan>>,
2778 ) -> Result<Vec<VectorSearchResult>> {
2779 self.search_binary_dense_vector_sync_impl(field, query, k, combiner, Some(probe_cache))
2780 }
2781}
2782
2783#[cfg(test)]
2784mod dense_search_safety_tests {
2785 use super::*;
2786
2787 #[test]
2788 fn dense_fetch_count_rejects_non_finite_and_unbounded_factors() {
2789 for factor in [
2790 f32::NAN,
2791 f32::INFINITY,
2792 f32::NEG_INFINITY,
2793 0.0,
2794 0.5,
2795 MAX_DENSE_RERANK_FACTOR + 1.0,
2796 ] {
2797 assert!(
2798 checked_dense_fetch_k(10, factor).is_err(),
2799 "factor={factor}"
2800 );
2801 }
2802 }
2803
2804 #[test]
2805 fn flat_document_collector_does_not_let_one_multivalue_doc_crowd_out_others() {
2806 let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
2807 collector.push(1, 0, 1.0);
2808 collector.push(1, 1, 0.9);
2809 collector.push(2, 0, 0.8);
2810
2811 let results = collector.into_results();
2812 assert_eq!(
2813 results
2814 .iter()
2815 .map(|result| result.doc_id)
2816 .collect::<Vec<_>>(),
2817 vec![1, 2]
2818 );
2819 assert_eq!(results[0].ordinals.len(), 2);
2820 }
2821
2822 #[test]
2823 fn flat_document_collector_evicts_by_score_then_doc_id() {
2824 let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
2825 collector.push(1, 0, 0.5);
2826 collector.push(3, 0, 0.8);
2827 collector.push(2, 0, 0.9);
2828 let results = collector.into_results();
2829 assert_eq!(
2830 results
2831 .iter()
2832 .map(|result| result.doc_id)
2833 .collect::<Vec<_>>(),
2834 vec![2, 3]
2835 );
2836
2837 let mut tied = FlatDocumentCollector::new(1, crate::query::MultiValueCombiner::Max);
2838 tied.push(2, 0, 1.0);
2839 tied.push(1, 0, 1.0);
2840 let results = tied.into_results();
2841 assert_eq!(results[0].doc_id, 1);
2842 }
2843
2844 #[test]
2845 fn dense_fetch_count_rounds_up_and_detects_overflow() {
2846 assert_eq!(checked_dense_fetch_k(3, 1.5).unwrap(), 5);
2847 assert_eq!(checked_dense_fetch_k(5_000, 3.0).unwrap(), 15_000);
2848 assert!(checked_dense_fetch_k(50_000, 3.0).is_err());
2849 assert!(checked_dense_fetch_k(usize::MAX, 2.0).is_err());
2850 }
2851
2852 #[test]
2853 fn file_ranges_reject_overflow_and_truncation() {
2854 assert_eq!(checked_file_range(4, 3, 7, "test").unwrap(), 4..7);
2855 assert!(checked_file_range(u64::MAX, 1, u64::MAX, "test").is_err());
2856 assert!(checked_file_range(5, 3, 7, "test").is_err());
2857 }
2858
2859 #[test]
2860 fn candidate_vector_reads_coalesce_contiguous_values() {
2861 let mut runs = Vec::new();
2862 plan_vector_read_runs(&[3, 4, 5, 9, 12, 13], &mut runs).unwrap();
2863 assert_eq!(runs.len(), 3);
2864 assert!(matches!(
2865 runs.as_slice(),
2866 [
2867 VectorReadRun {
2868 buffer_start: 0,
2869 flat_start: 3,
2870 count: 3,
2871 },
2872 VectorReadRun {
2873 buffer_start: 3,
2874 flat_start: 9,
2875 count: 1,
2876 },
2877 VectorReadRun {
2878 buffer_start: 4,
2879 flat_start: 12,
2880 count: 2,
2881 },
2882 ]
2883 ));
2884 assert!(plan_vector_read_runs(&[3, 3], &mut runs).is_err());
2885 }
2886
2887 #[tokio::test]
2888 async fn multivalue_ann_rerank_streams_past_document_candidate_cap() {
2889 use crate::directories::{FileHandle, OwnedBytes};
2890 use crate::segment::FlatVectorData;
2891
2892 const VALUES: usize = MAX_DENSE_CANDIDATES_PER_SEGMENT + 1;
2893 let mut encoded = Vec::new();
2894 let vectors = vec![1.0f32; VALUES];
2895 let doc_ids: Vec<_> = (0..VALUES).map(|ordinal| (0, ordinal as u16)).collect();
2896 FlatVectorData::serialize_binary_from_flat_streaming(
2897 1,
2898 &vectors,
2899 &doc_ids,
2900 DenseVectorQuantization::F32,
2901 &mut encoded,
2902 )
2903 .unwrap();
2904 let flat = LazyFlatVectorData::open_with_doc_limit(
2905 FileHandle::from_bytes(OwnedBytes::new(encoded)),
2906 Some(1),
2907 )
2908 .await
2909 .unwrap();
2910
2911 let (results, stats) = exact_score_dense_candidate_documents(
2912 &[(0, 0, 0.0)],
2913 &flat,
2914 &[1.0],
2915 false,
2916 crate::query::MultiValueCombiner::Max,
2917 1,
2918 )
2919 .await
2920 .unwrap();
2921 assert_eq!(stats.vector_count, VALUES);
2922 assert_eq!(results.len(), 1);
2923 assert_eq!(results[0].ordinals.len(), VALUES);
2924 assert!((results[0].score - 1.0).abs() < 1e-5);
2925 }
2926}