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 = 200_000;
20const MAX_ANN_ORDINAL_OVERFETCH: usize = 32;
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, IVFPQIndex, IVFRaBitQIndex, PQCodebook,
73 RaBitQIndex, SSTableStats, TermInfo,
74};
75use crate::{DocId, Error, Result};
76
77use super::store::{AsyncStoreReader, RawStoreBlock};
78use super::types::{SegmentFiles, SegmentId, SegmentMeta};
79
80pub(crate) fn combine_ordinal_results(
86 raw: impl IntoIterator<Item = (u32, u16, f32)>,
87 combiner: crate::query::MultiValueCombiner,
88 limit: usize,
89) -> Vec<VectorSearchResult> {
90 let collected: Vec<(u32, u16, f32)> = raw.into_iter().collect();
91
92 let num_raw = collected.len();
93 if log::log_enabled!(log::Level::Debug) {
94 let mut ids: Vec<u32> = collected.iter().map(|(d, _, _)| *d).collect();
95 ids.sort_unstable();
96 ids.dedup();
97 log::debug!(
98 "combine_ordinal_results: {} raw entries, {} unique docs, combiner={:?}, limit={}",
99 num_raw,
100 ids.len(),
101 combiner,
102 limit
103 );
104 }
105
106 let all_single = collected.iter().all(|&(_, ord, _)| ord == 0);
108 if all_single {
109 let mut results: Vec<VectorSearchResult> = collected
110 .into_iter()
111 .map(|(doc_id, _, score)| VectorSearchResult::new(doc_id, score, vec![(0, score)]))
112 .collect();
113 results.sort_unstable_by(|a, b| {
114 b.score
115 .total_cmp(&a.score)
116 .then_with(|| a.doc_id.cmp(&b.doc_id))
117 });
118 results.truncate(limit);
119 return results;
120 }
121
122 let mut doc_ordinals: rustc_hash::FxHashMap<DocId, Vec<(u32, f32)>> =
124 rustc_hash::FxHashMap::default();
125 for (doc_id, ordinal, score) in collected {
126 doc_ordinals
127 .entry(doc_id as DocId)
128 .or_default()
129 .push((ordinal as u32, score));
130 }
131 let mut results: Vec<VectorSearchResult> = doc_ordinals
132 .into_iter()
133 .map(|(doc_id, ordinals)| {
134 let combined_score = combiner.combine(&ordinals);
135 VectorSearchResult::new(doc_id, combined_score, ordinals)
136 })
137 .collect();
138 results.sort_unstable_by(|a, b| {
139 b.score
140 .total_cmp(&a.score)
141 .then_with(|| a.doc_id.cmp(&b.doc_id))
142 });
143 results.truncate(limit);
144 results
145}
146
147struct HeapVectorResult(VectorSearchResult);
152
153impl PartialEq for HeapVectorResult {
154 fn eq(&self, other: &Self) -> bool {
155 self.0.score.to_bits() == other.0.score.to_bits() && self.0.doc_id == other.0.doc_id
156 }
157}
158
159impl Eq for HeapVectorResult {}
160
161impl Ord for HeapVectorResult {
162 fn cmp(&self, other: &Self) -> Ordering {
163 other
166 .0
167 .score
168 .total_cmp(&self.0.score)
169 .then_with(|| self.0.doc_id.cmp(&other.0.doc_id))
170 }
171}
172
173impl PartialOrd for HeapVectorResult {
174 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
175 Some(self.cmp(other))
176 }
177}
178
179struct FlatDocumentCollector {
183 heap: BinaryHeap<HeapVectorResult>,
184 limit: usize,
185 combiner: crate::query::MultiValueCombiner,
186 current_doc: Option<DocId>,
187 current_ordinals: Vec<(u32, f32)>,
188}
189
190impl FlatDocumentCollector {
191 fn new(limit: usize, combiner: crate::query::MultiValueCombiner) -> Self {
192 Self {
193 heap: BinaryHeap::with_capacity(limit.min(8 * 1024)),
194 limit,
195 combiner,
196 current_doc: None,
197 current_ordinals: Vec::new(),
198 }
199 }
200
201 fn push(&mut self, doc_id: DocId, ordinal: u16, score: f32) {
202 if self.current_doc.is_some_and(|current| current != doc_id) {
203 self.finish_current();
204 }
205 self.current_doc = Some(doc_id);
206 self.current_ordinals.push((ordinal as u32, score));
207 }
208
209 fn finish_current(&mut self) {
210 let Some(doc_id) = self.current_doc.take() else {
211 return;
212 };
213 let score = self.combiner.combine(&self.current_ordinals);
214 let should_retain = self.heap.len() < self.limit
215 || self.heap.peek().is_some_and(|worst| {
216 HeapVectorResult(VectorSearchResult::new(doc_id, score, Vec::new()))
217 .cmp(worst)
218 .is_lt()
219 });
220
221 if !should_retain {
222 self.current_ordinals.clear();
226 return;
227 }
228
229 let ordinals = std::mem::take(&mut self.current_ordinals);
230 let entry = HeapVectorResult(VectorSearchResult::new(doc_id, score, ordinals));
231 if self.heap.len() < self.limit {
232 self.heap.push(entry);
233 } else if let Some(mut worst) = self.heap.peek_mut() {
234 let mut evicted = std::mem::replace(&mut worst.0, entry.0);
237 evicted.ordinals.clear();
238 self.current_ordinals = evicted.ordinals;
239 }
240 }
241
242 fn into_results(mut self) -> Vec<VectorSearchResult> {
243 self.finish_current();
244 let mut results: Vec<_> = self.heap.into_iter().map(|entry| entry.0).collect();
245 results.sort_unstable_by(|a, b| {
246 b.score
247 .total_cmp(&a.score)
248 .then_with(|| a.doc_id.cmp(&b.doc_id))
249 });
250 results
251 }
252}
253
254fn combine_grouped_ordinal_results(
257 raw: impl IntoIterator<Item = RawVectorCandidate>,
258 combiner: crate::query::MultiValueCombiner,
259 limit: usize,
260) -> Vec<VectorSearchResult> {
261 let mut collector = FlatDocumentCollector::new(limit, combiner);
262 for (doc_id, ordinal, score) in raw {
263 collector.push(doc_id, ordinal, score);
264 }
265 collector.into_results()
266}
267
268#[derive(Clone, Copy)]
269struct DenseSearchParams {
270 dim: usize,
271 nprobe: usize,
272 unit_norm: bool,
273}
274
275fn checked_dense_fetch_k(k: usize, rerank_factor: f32) -> Result<usize> {
277 if !rerank_factor.is_finite() || !(1.0..=MAX_DENSE_RERANK_FACTOR).contains(&rerank_factor) {
278 return Err(Error::Query(format!(
279 "dense rerank_factor must be finite and in [1, {MAX_DENSE_RERANK_FACTOR}], got {rerank_factor}"
280 )));
281 }
282
283 let fetch = (k as f64) * (rerank_factor as f64);
284 if !fetch.is_finite()
285 || fetch > usize::MAX as f64
286 || fetch > MAX_DENSE_CANDIDATES_PER_SEGMENT as f64
287 {
288 return Err(Error::Query(format!(
289 "dense candidate count exceeds the per-segment maximum of \
290 {MAX_DENSE_CANDIDATES_PER_SEGMENT}: k={k}, rerank_factor={rerank_factor}"
291 )));
292 }
293 Ok(fetch.ceil() as usize)
294}
295
296fn ann_ordinal_fetch_k(fetch_k: usize, num_vectors: usize, num_docs: usize) -> usize {
297 if num_vectors == 0 || num_docs == 0 {
298 return 0;
299 }
300 let average_values_per_doc = num_vectors
301 .div_ceil(num_docs)
302 .clamp(1, MAX_ANN_ORDINAL_OVERFETCH);
303 fetch_k
304 .saturating_mul(average_values_per_doc)
305 .min(MAX_DENSE_CANDIDATES_PER_SEGMENT)
306 .min(num_vectors)
307}
308
309fn progressive_ann_search<F>(
314 target_docs: usize,
315 initial_fetch: usize,
316 max_vectors: usize,
317 mut search: F,
318) -> Result<Vec<RawVectorCandidate>>
319where
320 F: FnMut(usize) -> Vec<RawVectorCandidate>,
321{
322 let max_fetch = max_vectors.min(MAX_DENSE_CANDIDATES_PER_SEGMENT);
323 if target_docs == 0 || max_fetch == 0 {
324 return Ok(Vec::new());
325 }
326
327 let target_docs = target_docs.min(max_fetch);
328 let mut fetch = initial_fetch.max(target_docs).min(max_fetch);
329 loop {
330 let results = search(fetch);
331 let mut docs = rustc_hash::FxHashSet::with_capacity_and_hasher(
332 target_docs.min(results.len()),
333 Default::default(),
334 );
335 for &(doc_id, _, _) in &results {
336 docs.insert(doc_id);
337 if docs.len() >= target_docs {
338 return Ok(results);
339 }
340 }
341 if results.len() < fetch || fetch == max_vectors {
342 return Ok(results);
343 }
344 if fetch == max_fetch {
345 return Err(Error::Query(format!(
346 "ANN search reached the per-segment candidate limit of \
347 {MAX_DENSE_CANDIDATES_PER_SEGMENT} with only {} of {target_docs} requested \
348 documents; reduce k or the number of vector values per document",
349 docs.len()
350 )));
351 }
352
353 let next_fetch = fetch.saturating_mul(2).min(max_fetch);
354 if next_fetch == fetch {
355 return Ok(results);
356 }
357 fetch = next_fetch;
358 }
359}
360
361fn single_scan_ann_search<F>(
373 target_docs: usize,
374 initial_fetch: usize,
375 max_vectors: usize,
376 search: F,
377) -> Result<Vec<RawVectorCandidate>>
378where
379 F: FnOnce(usize) -> Vec<RawVectorCandidate>,
380{
381 let max_fetch = max_vectors.min(MAX_DENSE_CANDIDATES_PER_SEGMENT);
382 if target_docs == 0 || max_fetch == 0 {
383 return Ok(Vec::new());
384 }
385 let target_docs = target_docs.min(max_fetch);
386
387 let mut results = search(max_fetch);
388 results.truncate(max_fetch);
389
390 let mut docs = rustc_hash::FxHashSet::with_capacity_and_hasher(target_docs, Default::default());
393 let mut satisfied_len = None;
394 for (i, &(doc_id, _, _)) in results.iter().enumerate() {
395 docs.insert(doc_id);
396 if docs.len() >= target_docs {
397 satisfied_len = Some(i + 1);
398 break;
399 }
400 }
401
402 let mut fetch = initial_fetch.max(target_docs).min(max_fetch);
403 loop {
404 let round_len = fetch.min(results.len());
405 if satisfied_len.is_some_and(|len| len <= round_len) {
406 results.truncate(round_len);
407 return Ok(results);
408 }
409 if results.len() < fetch || fetch == max_vectors {
410 return Ok(results);
411 }
412 if fetch == max_fetch {
413 return Err(Error::Query(format!(
414 "ANN search reached the per-segment candidate limit of \
415 {MAX_DENSE_CANDIDATES_PER_SEGMENT} with only {} of {target_docs} requested \
416 documents; reduce k or the number of vector values per document",
417 docs.len()
418 )));
419 }
420 let next_fetch = fetch.saturating_mul(2).min(max_fetch);
421 if next_fetch == fetch {
422 return Ok(results);
423 }
424 fetch = next_fetch;
425 }
426}
427
428#[inline]
429fn bounded_vector_score_batch(vector_byte_size: usize, preferred: usize) -> usize {
430 preferred.min((MAX_VECTOR_SCORE_BATCH_BYTES / vector_byte_size.max(1)).max(1))
431}
432
433fn checked_file_range(
434 offset: u64,
435 length: u64,
436 file_length: u64,
437 description: &str,
438) -> Result<std::ops::Range<u64>> {
439 let end = offset
440 .checked_add(length)
441 .ok_or_else(|| Error::Corruption(format!("{description} byte range overflows u64")))?;
442 if end > file_length {
443 return Err(Error::Corruption(format!(
444 "{description} byte range {offset}..{end} exceeds file length {file_length}"
445 )));
446 }
447 Ok(offset..end)
448}
449
450type RawVectorCandidate = (u32, u16, f32);
451type ResolvedVectorCandidate = (usize, usize); fn expand_ann_candidate_documents(
457 ann_results: &[RawVectorCandidate],
458 flat: &LazyFlatVectorData,
459) -> Result<(Vec<RawVectorCandidate>, Vec<ResolvedVectorCandidate>)> {
460 let mut candidate_docs: Vec<DocId> = ann_results.iter().map(|candidate| candidate.0).collect();
461 candidate_docs.sort_unstable();
462 candidate_docs.dedup();
463
464 let mut expanded = Vec::new();
465 let mut resolved = Vec::new();
466 for doc_id in candidate_docs {
467 let (start, count) = flat.flat_indexes_for_doc_range(doc_id);
468 if count == 0 {
469 return Err(Error::Corruption(format!(
470 "ANN candidate document {doc_id} is missing from flat vector storage"
471 )));
472 }
473 let next_len = expanded
474 .len()
475 .checked_add(count)
476 .ok_or_else(|| Error::Query("ANN candidate vector expansion overflow".to_string()))?;
477 if next_len > MAX_DENSE_CANDIDATES_PER_SEGMENT {
478 return Err(Error::Query(format!(
479 "ANN candidate documents expand to more than \
480 {MAX_DENSE_CANDIDATES_PER_SEGMENT} vectors in one segment"
481 )));
482 }
483 expanded.reserve(count);
484 resolved.reserve(count);
485 let end = start
486 .checked_add(count)
487 .ok_or_else(|| Error::Corruption("flat vector range overflow".to_string()))?;
488 for flat_index in start..end {
489 let (stored_doc_id, ordinal) = flat.get_doc_id(flat_index);
490 if stored_doc_id != doc_id {
491 return Err(Error::Corruption(format!(
492 "flat vector doc map is not contiguous for document {doc_id}"
493 )));
494 }
495 let result_index = expanded.len();
496 expanded.push((doc_id, ordinal, 0.0));
497 resolved.push((result_index, flat_index));
498 }
499 }
500 Ok((expanded, resolved))
501}
502
503fn fill_probe_scores_and_prune_resolved(
512 ann_results: &[RawVectorCandidate],
513 expanded: &mut [RawVectorCandidate],
514 resolved: Vec<ResolvedVectorCandidate>,
515) -> Vec<ResolvedVectorCandidate> {
516 let mut probe_scores: FxHashMap<(DocId, u16), f32> =
517 FxHashMap::with_capacity_and_hasher(ann_results.len(), Default::default());
518 for &(doc_id, ordinal, score) in ann_results {
519 probe_scores.insert((doc_id, ordinal), score);
520 }
521 resolved
522 .into_iter()
523 .filter(|&(result_index, _)| {
524 let (doc_id, ordinal, _) = expanded[result_index];
525 match probe_scores.get(&(doc_id, ordinal)) {
526 Some(&score) => {
527 expanded[result_index].2 = score;
528 false
529 }
530 None => true,
531 }
532 })
533 .collect()
534}
535
536async fn exact_score_binary_candidate_documents(
537 ann_results: &[RawVectorCandidate],
538 flat: &LazyFlatVectorData,
539 query: &[u8],
540 dim_bits: usize,
541) -> Result<Vec<RawVectorCandidate>> {
542 let (mut expanded, resolved) = expand_ann_candidate_documents(ann_results, flat)?;
543 let resolved = fill_probe_scores_and_prune_resolved(ann_results, &mut expanded, resolved);
544 let vector_byte_size = flat.vector_byte_size();
545 let batch_len = bounded_vector_score_batch(vector_byte_size, BINARY_SCORE_BATCH);
546 let raw_capacity = batch_len
547 .checked_mul(vector_byte_size)
548 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
549 let mut raw = vec![0u8; raw_capacity];
550 let mut scores = vec![0.0f32; batch_len];
551
552 for chunk in resolved.chunks(batch_len) {
553 #[cfg(feature = "native")]
554 flat.prefetch_vectors(chunk.iter().map(|&(_, flat_index)| flat_index));
555 let raw_len = chunk
556 .len()
557 .checked_mul(vector_byte_size)
558 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
559 let raw = &mut raw[..raw_len];
560 for (buffer_index, &(_, flat_index)) in chunk.iter().enumerate() {
561 flat.read_vector_raw_into(
562 flat_index,
563 &mut raw[buffer_index * vector_byte_size..(buffer_index + 1) * vector_byte_size],
564 )
565 .await
566 .map_err(Error::Io)?;
567 }
568 crate::structures::simd::batch_hamming_scores(
569 query,
570 raw,
571 vector_byte_size,
572 dim_bits,
573 &mut scores[..chunk.len()],
574 );
575 for (buffer_index, &(result_index, _)) in chunk.iter().enumerate() {
576 expanded[result_index].2 = scores[buffer_index];
577 }
578 }
579 Ok(expanded)
580}
581
582#[cfg(feature = "sync")]
583fn exact_score_binary_candidate_documents_sync(
584 ann_results: &[RawVectorCandidate],
585 flat: &LazyFlatVectorData,
586 query: &[u8],
587 dim_bits: usize,
588) -> Result<Vec<RawVectorCandidate>> {
589 let (mut expanded, resolved) = expand_ann_candidate_documents(ann_results, flat)?;
590 let resolved = fill_probe_scores_and_prune_resolved(ann_results, &mut expanded, resolved);
591 let vector_byte_size = flat.vector_byte_size();
592 let batch_len = bounded_vector_score_batch(vector_byte_size, BINARY_SCORE_BATCH);
593 let raw_capacity = batch_len
594 .checked_mul(vector_byte_size)
595 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
596 let mut raw = vec![0u8; raw_capacity];
597 let mut scores = vec![0.0f32; batch_len];
598
599 for chunk in resolved.chunks(batch_len) {
600 let raw_len = chunk
601 .len()
602 .checked_mul(vector_byte_size)
603 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
604 let raw = &mut raw[..raw_len];
605 for (buffer_index, &(_, flat_index)) in chunk.iter().enumerate() {
606 flat.read_vector_raw_into_sync(
607 flat_index,
608 &mut raw[buffer_index * vector_byte_size..(buffer_index + 1) * vector_byte_size],
609 )
610 .map_err(Error::Io)?;
611 }
612 crate::structures::simd::batch_hamming_scores(
613 query,
614 raw,
615 vector_byte_size,
616 dim_bits,
617 &mut scores[..chunk.len()],
618 );
619 for (buffer_index, &(result_index, _)) in chunk.iter().enumerate() {
620 expanded[result_index].2 = scores[buffer_index];
621 }
622 }
623 Ok(expanded)
624}
625
626fn validate_coarse_centroids(centroids: &CoarseCentroids, dim: usize) -> Result<()> {
627 let expected = (centroids.num_clusters as usize)
628 .checked_mul(dim)
629 .ok_or_else(|| Error::Corruption("coarse centroid size overflow".into()))?;
630 if centroids.num_clusters == 0
631 || centroids.dim != dim
632 || centroids.centroids.len() != expected
633 || centroids.centroids.iter().any(|value| !value.is_finite())
634 {
635 return Err(Error::Corruption(format!(
636 "invalid coarse centroids: clusters={}, dim={}, values={} (expected dim={dim}, values={expected})",
637 centroids.num_clusters,
638 centroids.dim,
639 centroids.centroids.len()
640 )));
641 }
642 Ok(())
643}
644
645pub struct SegmentReader {
651 meta: SegmentMeta,
652 term_dict: Arc<AsyncSSTableReader<TermInfo>>,
654 postings_handle: FileHandle,
656 store: Arc<AsyncStoreReader>,
658 schema: Arc<Schema>,
659 vector_indexes: FxHashMap<u32, VectorIndex>,
661 flat_vectors: FxHashMap<u32, LazyFlatVectorData>,
663 coarse_centroids: FxHashMap<u32, Arc<CoarseCentroids>>,
665 sparse_indexes: FxHashMap<u32, SparseIndex>,
667 bmp_indexes: FxHashMap<u32, BmpIndex>,
669 positions_handle: Option<FileHandle>,
671 fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldReader>,
673 #[cfg(feature = "native")]
678 pin_report: crate::segment::pin::PinReport,
679}
680
681impl SegmentReader {
682 pub async fn open<D: Directory>(
684 dir: &D,
685 segment_id: SegmentId,
686 schema: Arc<Schema>,
687 cache_blocks: usize,
688 ) -> Result<Self> {
689 Self::open_with_cache_blocks(dir, segment_id, schema, cache_blocks, cache_blocks).await
690 }
691
692 pub async fn open_with_cache_blocks<D: Directory>(
698 dir: &D,
699 segment_id: SegmentId,
700 schema: Arc<Schema>,
701 term_cache_blocks: usize,
702 store_cache_blocks: usize,
703 ) -> Result<Self> {
704 let files = SegmentFiles::new(segment_id.0);
705
706 let meta_slice = dir.open_read(&files.meta).await?;
708 let meta_bytes = meta_slice.read_bytes().await?;
709 let meta = SegmentMeta::deserialize(meta_bytes.as_slice())?;
710 debug_assert_eq!(meta.id, segment_id.0);
711
712 let term_dict_handle = dir.open_lazy(&files.term_dict).await?;
714 let term_dict = AsyncSSTableReader::open(term_dict_handle, term_cache_blocks).await?;
715
716 let postings_handle = dir.open_lazy(&files.postings).await?;
718
719 let store_handle = dir.open_lazy(&files.store).await?;
721 let store = AsyncStoreReader::open(store_handle, store_cache_blocks).await?;
722
723 let vectors_data = loader::load_vectors_file(dir, &files, &schema, meta.num_docs).await?;
725 let vector_indexes = vectors_data.indexes;
726 let flat_vectors = vectors_data.flat_vectors;
727
728 #[cfg(feature = "native")]
733 for (field_id, lazy_flat) in &flat_vectors {
734 if vector_indexes.contains_key(field_id) {
735 lazy_flat.advise_random_access();
736 }
737 }
738
739 let sparse_data = loader::load_sparse_file(dir, &files, meta.num_docs, &schema).await?;
741 let sparse_indexes = sparse_data.maxscore_indexes;
742 let bmp_indexes = sparse_data.bmp_indexes;
743
744 let positions_handle = loader::open_positions_file(dir, &files, &schema).await?;
746
747 let fast_fields = loader::load_fast_fields_file(dir, &files, &schema).await?;
749
750 {
752 let mut parts = vec![format!(
753 "[segment] loaded {:016x}: docs={}",
754 segment_id.0, meta.num_docs
755 )];
756 if !vector_indexes.is_empty() || !flat_vectors.is_empty() {
757 parts.push(format!(
758 "dense: {} ann + {} flat fields",
759 vector_indexes.len(),
760 flat_vectors.len()
761 ));
762 }
763 for (field_id, idx) in &sparse_indexes {
764 parts.push(format!(
765 "sparse field {}: {} dims, ~{:.1} KB",
766 field_id,
767 idx.num_dimensions(),
768 idx.num_dimensions() as f64 * 24.0 / 1024.0
769 ));
770 }
771 for (field_id, idx) in &bmp_indexes {
772 parts.push(format!(
773 "bmp field {}: {} dims, {} blocks",
774 field_id,
775 idx.dims(),
776 idx.num_blocks
777 ));
778 }
779 if !fast_fields.is_empty() {
780 parts.push(format!("fast: {} fields", fast_fields.len()));
781 }
782 log::debug!("{}", parts.join(", "));
783 }
784
785 #[allow(unused_mut)]
786 let mut reader = Self {
787 meta,
788 term_dict: Arc::new(term_dict),
789 postings_handle,
790 store: Arc::new(store),
791 schema,
792 vector_indexes,
793 flat_vectors,
794 coarse_centroids: FxHashMap::default(),
795 sparse_indexes,
796 bmp_indexes,
797 positions_handle,
798 fast_fields,
799 #[cfg(feature = "native")]
800 pin_report: Default::default(),
801 };
802
803 #[cfg(feature = "native")]
805 reader.apply_pin_policy(&crate::segment::pin::pin_policy().to_owned());
806
807 Ok(reader)
808 }
809
810 #[cfg(feature = "native")]
819 pub(crate) fn apply_pin_policy(&mut self, policy: &crate::segment::pin::PinPolicy) {
820 use crate::segment::pin::PinReport;
821
822 if !policy.is_enabled() {
823 return;
824 }
825 let mut remaining = policy.budget_bytes;
826 let mut report = PinReport::default();
827
828 for bmp in self.bmp_indexes.values_mut() {
830 bmp.pin_block_starts(policy.mode, &mut remaining, &mut report);
831 }
832 for sparse in self.sparse_indexes.values_mut() {
834 sparse.pin_skip_section(policy.mode, &mut remaining, &mut report);
835 }
836 for flat in self.flat_vectors.values_mut() {
838 flat.pin_doc_ids(policy.mode, &mut remaining, &mut report);
839 }
840 for bmp in self.bmp_indexes.values_mut() {
841 bmp.pin_doc_maps(policy.mode, &mut remaining, &mut report);
842 }
843 for bmp in self.bmp_indexes.values_mut() {
845 bmp.pin_sb_grid(policy.mode, &mut remaining, &mut report);
846 }
847
848 if report.skipped_budget_bytes > 0 || report.failed_bytes > 0 {
849 log::warn!(
850 "[pin] segment {:016x}: pinned {}/{} bytes (budget skipped {}, mlock failed {}) — raise HERMES_PIN_METADATA_BUDGET_MB or RLIMIT_MEMLOCK for full coverage",
851 self.meta.id,
852 report.pinned_bytes,
853 report.intended_bytes,
854 report.skipped_budget_bytes,
855 report.failed_bytes,
856 );
857 } else if report.pinned_bytes > 0 {
858 log::info!(
859 "[pin] segment {:016x}: pinned {} bytes of hot metadata ({:?})",
860 self.meta.id,
861 report.pinned_bytes,
862 policy.mode,
863 );
864 }
865 self.pin_report = report;
866 }
867
868 pub fn meta(&self) -> &SegmentMeta {
874 &self.meta
875 }
876
877 pub fn num_docs(&self) -> u32 {
878 self.meta.num_docs
879 }
880
881 pub fn avg_field_len(&self, field: Field) -> f32 {
883 self.meta.avg_field_len(field)
884 }
885
886 pub fn schema(&self) -> &Schema {
887 &self.schema
888 }
889
890 pub fn sparse_indexes(&self) -> &FxHashMap<u32, SparseIndex> {
892 &self.sparse_indexes
893 }
894
895 pub fn sparse_index(&self, field: Field) -> Option<&SparseIndex> {
897 self.sparse_indexes.get(&field.0)
898 }
899
900 pub fn bmp_index(&self, field: Field) -> Option<&BmpIndex> {
902 self.bmp_indexes.get(&field.0)
903 }
904
905 pub fn bmp_indexes(&self) -> &FxHashMap<u32, BmpIndex> {
907 &self.bmp_indexes
908 }
909
910 pub fn vector_indexes(&self) -> &FxHashMap<u32, VectorIndex> {
912 &self.vector_indexes
913 }
914
915 pub fn flat_vectors(&self) -> &FxHashMap<u32, LazyFlatVectorData> {
917 &self.flat_vectors
918 }
919
920 pub fn fast_field(
922 &self,
923 field_id: u32,
924 ) -> Option<&crate::structures::fast_field::FastFieldReader> {
925 self.fast_fields.get(&field_id)
926 }
927
928 pub fn fast_fields(&self) -> &FxHashMap<u32, crate::structures::fast_field::FastFieldReader> {
930 &self.fast_fields
931 }
932
933 pub fn term_dict_stats(&self) -> SSTableStats {
935 self.term_dict.stats()
936 }
937
938 pub fn memory_stats(&self) -> SegmentMemoryStats {
940 let term_dict_stats = self.term_dict.stats();
941
942 let term_dict_cache_bytes = self.term_dict.cached_bytes();
946 let store_cache_bytes = self.store.cached_bytes();
947
948 let sparse_index_bytes: usize = self
950 .sparse_indexes
951 .values()
952 .map(|s| s.estimated_memory_bytes())
953 .sum::<usize>()
954 + self
955 .bmp_indexes
956 .values()
957 .map(|b| b.estimated_memory_bytes())
958 .sum::<usize>();
959
960 let dense_index_bytes: usize = self
963 .vector_indexes
964 .values()
965 .map(|v| v.estimated_memory_bytes())
966 .sum();
967
968 #[cfg(feature = "native")]
969 let (pinned_metadata_bytes, pin_intended_bytes) =
970 (self.pin_report.pinned_bytes, self.pin_report.intended_bytes);
971 #[cfg(not(feature = "native"))]
972 let (pinned_metadata_bytes, pin_intended_bytes) = (0u64, 0u64);
973
974 SegmentMemoryStats {
975 segment_id: self.meta.id,
976 num_docs: self.meta.num_docs,
977 term_dict_cache_bytes,
978 store_cache_bytes,
979 sparse_index_bytes,
980 dense_index_bytes,
981 bloom_filter_bytes: term_dict_stats.bloom_filter_size,
982 pinned_metadata_bytes,
983 pin_intended_bytes,
984 }
985 }
986
987 pub async fn get_postings(
992 &self,
993 field: Field,
994 term: &[u8],
995 ) -> Result<Option<BlockPostingList>> {
996 log::debug!(
997 "SegmentReader::get_postings field={} term_len={}",
998 field.0,
999 term.len()
1000 );
1001
1002 let mut key = Vec::with_capacity(4 + term.len());
1004 key.extend_from_slice(&field.0.to_le_bytes());
1005 key.extend_from_slice(term);
1006
1007 let term_info = match self.term_dict.get(&key).await? {
1009 Some(info) => {
1010 log::debug!("SegmentReader::get_postings found term_info");
1011 info
1012 }
1013 None => {
1014 log::debug!("SegmentReader::get_postings term not found");
1015 return Ok(None);
1016 }
1017 };
1018
1019 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1021 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1023 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1024 posting_list.push(doc_id, tf);
1025 }
1026 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
1027 return Ok(Some(block_list));
1028 }
1029
1030 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
1032 Error::Corruption("TermInfo has neither inline nor external data".to_string())
1033 })?;
1034
1035 let range = checked_file_range(
1036 posting_offset,
1037 posting_len,
1038 self.postings_handle.len(),
1039 "posting",
1040 )?;
1041 let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
1042 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
1043
1044 Ok(Some(block_list))
1045 }
1046
1047 pub async fn get_prefix_postings(
1049 &self,
1050 field: Field,
1051 prefix: &[u8],
1052 ) -> Result<Vec<BlockPostingList>> {
1053 if prefix.is_empty() {
1054 return Err(Error::Query("prefix must not be empty".into()));
1055 }
1056 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
1058 key_prefix.extend_from_slice(&field.0.to_le_bytes());
1059 key_prefix.extend_from_slice(prefix);
1060
1061 let (entries, truncated) = self
1062 .term_dict
1063 .prefix_scan_limited(&key_prefix, MAX_PREFIX_TERMS)
1064 .await?;
1065 if truncated {
1066 return Err(Error::Query(format!(
1067 "prefix expands to more than {MAX_PREFIX_TERMS} terms"
1068 )));
1069 }
1070 let posting_count: u64 = entries
1071 .iter()
1072 .map(|(_, term_info)| term_info.doc_freq() as u64)
1073 .sum();
1074 if posting_count > MAX_PREFIX_POSTINGS {
1075 return Err(Error::Query(format!(
1076 "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
1077 )));
1078 }
1079 let mut results = Vec::with_capacity(entries.len());
1080
1081 for (_key, term_info) in entries {
1082 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1083 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1084 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1085 posting_list.push(doc_id, tf);
1086 }
1087 results.push(BlockPostingList::from_posting_list(&posting_list)?);
1088 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
1089 let range = checked_file_range(
1090 posting_offset,
1091 posting_len,
1092 self.postings_handle.len(),
1093 "prefix posting",
1094 )?;
1095 let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
1096 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
1097 }
1098 }
1099
1100 Ok(results)
1101 }
1102
1103 pub async fn doc(&self, local_doc_id: DocId) -> Result<Option<Document>> {
1108 self.doc_with_fields(local_doc_id, None).await
1109 }
1110
1111 pub async fn doc_with_fields(
1117 &self,
1118 local_doc_id: DocId,
1119 fields: Option<&rustc_hash::FxHashSet<u32>>,
1120 ) -> Result<Option<Document>> {
1121 let mut doc = match fields {
1122 Some(set) => {
1123 let field_ids: Vec<u32> = set.iter().copied().collect();
1124 match self
1125 .store
1126 .get_fields(local_doc_id, &self.schema, &field_ids)
1127 .await
1128 {
1129 Ok(Some(d)) => d,
1130 Ok(None) => return Ok(None),
1131 Err(e) => return Err(Error::from(e)),
1132 }
1133 }
1134 None => match self.store.get(local_doc_id, &self.schema).await {
1135 Ok(Some(d)) => d,
1136 Ok(None) => return Ok(None),
1137 Err(e) => return Err(Error::from(e)),
1138 },
1139 };
1140
1141 for (&field_id, lazy_flat) in &self.flat_vectors {
1143 if let Some(set) = fields
1145 && !set.contains(&field_id)
1146 {
1147 continue;
1148 }
1149
1150 let is_binary = lazy_flat.quantization == DenseVectorQuantization::Binary;
1151 let (start, entries) = lazy_flat.flat_indexes_for_doc(local_doc_id);
1152 for (j, &(_doc_id, _ordinal)) in entries.iter().enumerate() {
1153 let flat_idx = start + j;
1154 if is_binary {
1155 let vbs = lazy_flat.vector_byte_size();
1156 let mut raw = vec![0u8; vbs];
1157 match lazy_flat.read_vector_raw_into(flat_idx, &mut raw).await {
1158 Ok(()) => {
1159 doc.add_binary_dense_vector(Field(field_id), raw);
1160 }
1161 Err(e) => {
1162 log::warn!("Failed to hydrate binary vector field {}: {}", field_id, e);
1163 }
1164 }
1165 } else {
1166 match lazy_flat.get_vector(flat_idx).await {
1167 Ok(vec) => {
1168 doc.add_dense_vector(Field(field_id), vec);
1169 }
1170 Err(e) => {
1171 log::warn!("Failed to hydrate vector field {}: {}", field_id, e);
1172 }
1173 }
1174 }
1175 }
1176 }
1177
1178 Ok(Some(doc))
1179 }
1180
1181 pub async fn prefetch_terms(
1183 &self,
1184 field: Field,
1185 start_term: &[u8],
1186 end_term: &[u8],
1187 ) -> Result<()> {
1188 let mut start_key = Vec::with_capacity(4 + start_term.len());
1189 start_key.extend_from_slice(&field.0.to_le_bytes());
1190 start_key.extend_from_slice(start_term);
1191
1192 let mut end_key = Vec::with_capacity(4 + end_term.len());
1193 end_key.extend_from_slice(&field.0.to_le_bytes());
1194 end_key.extend_from_slice(end_term);
1195
1196 self.term_dict.prefetch_range(&start_key, &end_key).await?;
1197 Ok(())
1198 }
1199
1200 pub fn store_has_dict(&self) -> bool {
1202 self.store.has_dict()
1203 }
1204
1205 pub fn store(&self) -> &super::store::AsyncStoreReader {
1207 &self.store
1208 }
1209
1210 pub fn store_raw_blocks(&self) -> Vec<RawStoreBlock> {
1212 self.store.raw_blocks()
1213 }
1214
1215 pub fn store_data_slice(&self) -> &FileHandle {
1217 self.store.data_slice()
1218 }
1219
1220 pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
1222 self.term_dict.all_entries().await.map_err(Error::from)
1223 }
1224
1225 pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
1230 let entries = self.term_dict.all_entries().await?;
1231 let mut result = Vec::with_capacity(entries.len());
1232
1233 for (key, term_info) in entries {
1234 if key.len() > 4 {
1236 let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
1237 let term_bytes = &key[4..];
1238 if let Ok(term_str) = std::str::from_utf8(term_bytes) {
1239 result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
1240 }
1241 }
1242 }
1243
1244 Ok(result)
1245 }
1246
1247 pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
1249 self.term_dict.iter()
1250 }
1251
1252 pub async fn prefetch_term_dict(&self) -> crate::Result<()> {
1256 self.term_dict
1257 .prefetch_all_data_bulk()
1258 .await
1259 .map_err(crate::Error::from)
1260 }
1261
1262 pub async fn read_postings(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
1264 let range = checked_file_range(offset, len, self.postings_handle.len(), "posting")?;
1265 let bytes = self.postings_handle.read_bytes_range(range).await?;
1266 Ok(bytes.to_vec())
1267 }
1268
1269 pub async fn read_position_bytes(&self, offset: u64, len: u64) -> Result<Option<Vec<u8>>> {
1271 let handle = match &self.positions_handle {
1272 Some(h) => h,
1273 None => return Ok(None),
1274 };
1275 let range = checked_file_range(offset, len, handle.len(), "position")?;
1276 let bytes = handle.read_bytes_range(range).await?;
1277 Ok(Some(bytes.to_vec()))
1278 }
1279
1280 pub fn has_positions_file(&self) -> bool {
1282 self.positions_handle.is_some()
1283 }
1284
1285 fn validate_dense_search_request(
1289 &self,
1290 field: Field,
1291 query: &[f32],
1292 nprobe: usize,
1293 rerank_factor: f32,
1294 combiner: crate::query::MultiValueCombiner,
1295 ) -> Result<DenseSearchParams> {
1296 let entry = self
1297 .schema
1298 .get_field_entry(field)
1299 .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
1300 if entry.field_type != crate::dsl::FieldType::DenseVector {
1301 return Err(Error::InvalidFieldType {
1302 expected: "dense_vector".to_string(),
1303 got: format!("{:?}", entry.field_type),
1304 });
1305 }
1306 let config = entry.dense_vector_config.as_ref().ok_or_else(|| {
1307 Error::Schema(format!(
1308 "dense vector field '{}' has no dense vector configuration",
1309 entry.name
1310 ))
1311 })?;
1312
1313 if query.is_empty() {
1314 return Err(Error::Query(format!(
1315 "dense query vector for field '{}' must not be empty",
1316 entry.name
1317 )));
1318 }
1319 if query.len() != config.dim {
1320 return Err(Error::Query(format!(
1321 "dense query vector dimension {} does not match field '{}' dimension {}",
1322 query.len(),
1323 entry.name,
1324 config.dim
1325 )));
1326 }
1327 if let Some((index, value)) = query
1328 .iter()
1329 .enumerate()
1330 .find(|(_, value)| !value.is_finite())
1331 {
1332 return Err(Error::Query(format!(
1333 "dense query vector for field '{}' contains non-finite value {value} at index {index}",
1334 entry.name
1335 )));
1336 }
1337
1338 let nprobe = match (nprobe, config.nprobe) {
1341 (0, 0) => 32,
1342 (0, schema_nprobe) => schema_nprobe,
1343 (query_nprobe, _) => query_nprobe,
1344 };
1345 if nprobe > MAX_DENSE_NPROBE {
1346 return Err(Error::Query(format!(
1347 "dense nprobe must be at most {MAX_DENSE_NPROBE}, got {nprobe}"
1348 )));
1349 }
1350
1351 checked_dense_fetch_k(0, rerank_factor)?;
1354 combiner.validate().map_err(Error::Query)?;
1355
1356 Ok(DenseSearchParams {
1357 dim: config.dim,
1358 nprobe,
1359 unit_norm: config.unit_norm,
1360 })
1361 }
1362
1363 fn validate_binary_search_request(&self, field: Field, query: &[u8]) -> Result<usize> {
1364 let entry = self
1365 .schema
1366 .get_field_entry(field)
1367 .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
1368 if entry.field_type != crate::dsl::FieldType::BinaryDenseVector {
1369 return Err(Error::InvalidFieldType {
1370 expected: "binary_dense_vector".to_string(),
1371 got: format!("{:?}", entry.field_type),
1372 });
1373 }
1374 let config = entry.binary_dense_vector_config.as_ref().ok_or_else(|| {
1375 Error::Schema(format!(
1376 "binary dense vector field '{}' has no configuration",
1377 entry.name
1378 ))
1379 })?;
1380 if config.dim == 0 || !config.dim.is_multiple_of(8) {
1381 return Err(Error::Schema(format!(
1382 "binary dense vector field '{}' has invalid dimension {}",
1383 entry.name, config.dim
1384 )));
1385 }
1386 if query.len() != config.byte_len() {
1387 return Err(Error::Query(format!(
1388 "binary query byte length {} does not match field '{}' byte length {}",
1389 query.len(),
1390 entry.name,
1391 config.byte_len()
1392 )));
1393 }
1394 Ok(config.dim)
1395 }
1396
1397 fn score_quantized_batch(
1403 query: &[f32],
1404 raw: &[u8],
1405 quant: crate::dsl::DenseVectorQuantization,
1406 dim: usize,
1407 scores: &mut [f32],
1408 unit_norm: bool,
1409 ) -> Result<()> {
1410 use crate::dsl::DenseVectorQuantization;
1411 use crate::structures::simd;
1412
1413 if query.len() != dim {
1414 return Err(Error::Query(format!(
1415 "dense SIMD query dimension {} does not match vector dimension {dim}",
1416 query.len()
1417 )));
1418 }
1419 let element_size = match quant {
1420 DenseVectorQuantization::F32 => std::mem::size_of::<f32>(),
1421 DenseVectorQuantization::F16 => std::mem::size_of::<u16>(),
1422 DenseVectorQuantization::UInt8 => 1,
1423 DenseVectorQuantization::Binary => {
1424 return Err(Error::InvalidFieldType {
1425 expected: "non-binary dense vector".to_string(),
1426 got: "binary dense vector".to_string(),
1427 });
1428 }
1429 };
1430 let required_bytes = scores
1431 .len()
1432 .checked_mul(dim)
1433 .and_then(|elements| elements.checked_mul(element_size))
1434 .ok_or_else(|| Error::Corruption("dense vector batch byte length overflow".into()))?;
1435 if raw.len() < required_bytes {
1436 return Err(Error::Corruption(format!(
1437 "dense vector batch is truncated: need {required_bytes} bytes, got {}",
1438 raw.len()
1439 )));
1440 }
1441 if quant == DenseVectorQuantization::F16
1442 && required_bytes > 0
1443 && !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<u16>())
1444 {
1445 return Err(Error::Corruption(
1446 "f16 vector data is not 2-byte aligned".to_string(),
1447 ));
1448 }
1449
1450 match (quant, unit_norm) {
1451 (DenseVectorQuantization::F32, false) => {
1452 let num_floats = scores.len() * dim;
1453 if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
1454 return Err(Error::Corruption(
1455 "f32 vector data is not 4-byte aligned".to_string(),
1456 ));
1457 }
1458 let vectors: &[f32] =
1459 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
1460 simd::batch_cosine_scores(query, vectors, dim, scores);
1461 }
1462 (DenseVectorQuantization::F32, true) => {
1463 let num_floats = scores.len() * dim;
1464 if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
1465 return Err(Error::Corruption(
1466 "f32 vector data is not 4-byte aligned".to_string(),
1467 ));
1468 }
1469 let vectors: &[f32] =
1470 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
1471 simd::batch_dot_scores(query, vectors, dim, scores);
1472 }
1473 (DenseVectorQuantization::F16, false) => {
1474 simd::batch_cosine_scores_f16(query, raw, dim, scores);
1475 }
1476 (DenseVectorQuantization::F16, true) => {
1477 simd::batch_dot_scores_f16(query, raw, dim, scores);
1478 }
1479 (DenseVectorQuantization::UInt8, false) => {
1480 simd::batch_cosine_scores_u8(query, raw, dim, scores);
1481 }
1482 (DenseVectorQuantization::UInt8, true) => {
1483 simd::batch_dot_scores_u8(query, raw, dim, scores);
1484 }
1485 (DenseVectorQuantization::Binary, _) => unreachable!("validated above"),
1486 }
1487 Ok(())
1488 }
1489
1490 pub async fn search_dense_vector(
1496 &self,
1497 field: Field,
1498 query: &[f32],
1499 k: usize,
1500 nprobe: usize,
1501 rerank_factor: f32,
1502 combiner: crate::query::MultiValueCombiner,
1503 ) -> Result<Vec<VectorSearchResult>> {
1504 let params =
1505 self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
1506 let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
1507 if k == 0 {
1508 return Ok(Vec::new());
1509 }
1510
1511 let ann_index = self.vector_indexes.get(&field.0);
1512 let lazy_flat = self.flat_vectors.get(&field.0);
1513 let ann_fetch_k = lazy_flat.map_or(fetch_k, |flat| {
1514 ann_ordinal_fetch_k(fetch_k, flat.num_vectors, flat.num_docs_with_vectors())
1515 });
1516
1517 if ann_index.is_none() && lazy_flat.is_none() {
1519 return Ok(Vec::new());
1520 }
1521
1522 if ann_index.is_some() && lazy_flat.is_none() {
1523 return Err(Error::Corruption(format!(
1524 "dense ANN field {} is missing flat vector storage",
1525 field.0
1526 )));
1527 }
1528
1529 if let Some(flat) = lazy_flat
1530 && flat.dim != params.dim
1531 {
1532 return Err(Error::Corruption(format!(
1533 "dense vector field {} has schema dimension {} but flat storage dimension {}",
1534 field.0, params.dim, flat.dim
1535 )));
1536 }
1537
1538 let t0 = std::time::Instant::now();
1540 let mut flat_results = None;
1541 let mut results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
1542 match index {
1544 VectorIndex::RaBitQ(lazy) => {
1545 let rabitq = lazy.get().ok_or_else(|| {
1546 Error::Schema("RaBitQ index deserialization failed".to_string())
1547 })?;
1548 if rabitq.codebook.config.dim != params.dim {
1549 return Err(Error::Corruption(format!(
1550 "RaBitQ index dimension {} does not match schema dimension {}",
1551 rabitq.codebook.config.dim, params.dim
1552 )));
1553 }
1554 let flat = lazy_flat.expect("ANN/flat pairing validated above");
1555 progressive_ann_search(
1556 fetch_k.min(flat.num_docs_with_vectors()),
1557 ann_fetch_k,
1558 rabitq.len().min(flat.num_vectors),
1559 |candidate_k| {
1560 rabitq
1561 .search(query, candidate_k)
1562 .into_iter()
1563 .map(|(doc_id, ordinal, dist)| {
1564 (doc_id, ordinal, 1.0 / (1.0 + dist))
1565 })
1566 .collect()
1567 },
1568 )?
1569 }
1570 VectorIndex::IVF(lazy) => {
1571 let (index, codebook) = lazy.get().ok_or_else(|| {
1572 Error::Schema("IVF index deserialization failed".to_string())
1573 })?;
1574 let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
1575 Error::Schema(format!(
1576 "IVF index requires coarse centroids for field {}",
1577 field.0
1578 ))
1579 })?;
1580 validate_coarse_centroids(centroids, params.dim)?;
1581 if index.config.dim != params.dim
1582 || codebook.config.dim != params.dim
1583 || index.centroids_version != centroids.version
1584 || index.codebook_version != codebook.version
1585 || index
1586 .clusters
1587 .iter()
1588 .any(|(cluster_id, _)| cluster_id >= centroids.num_clusters)
1589 {
1590 return Err(Error::Corruption(format!(
1591 "IVF index/codebook/centroid metadata does not match schema dimension {}",
1592 params.dim
1593 )));
1594 }
1595 let effective_nprobe = params.nprobe.min(centroids.num_clusters as usize);
1596 let flat = lazy_flat.expect("ANN/flat pairing validated above");
1597 progressive_ann_search(
1598 fetch_k.min(flat.num_docs_with_vectors()),
1599 ann_fetch_k,
1600 index.len().min(flat.num_vectors),
1601 |candidate_k| {
1602 index
1603 .search(
1604 centroids,
1605 codebook,
1606 query,
1607 candidate_k,
1608 Some(effective_nprobe),
1609 )
1610 .into_iter()
1611 .map(|(doc_id, ordinal, dist)| {
1612 (doc_id, ordinal, 1.0 / (1.0 + dist))
1613 })
1614 .collect()
1615 },
1616 )?
1617 }
1618 VectorIndex::ScaNN(lazy) => {
1619 let (index, codebook) = lazy.get().ok_or_else(|| {
1620 Error::Schema("ScaNN index deserialization failed".to_string())
1621 })?;
1622 let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
1623 Error::Schema(format!(
1624 "ScaNN index requires coarse centroids for field {}",
1625 field.0
1626 ))
1627 })?;
1628 validate_coarse_centroids(centroids, params.dim)?;
1629 if index.config.dim != params.dim
1630 || codebook.config.dim != params.dim
1631 || index.centroids_version != centroids.version
1632 || index.codebook_version != codebook.version
1633 || index
1634 .clusters
1635 .iter()
1636 .any(|(cluster_id, _)| cluster_id >= centroids.num_clusters)
1637 {
1638 return Err(Error::Corruption(format!(
1639 "ScaNN index/codebook/centroid metadata does not match schema dimension {}",
1640 params.dim
1641 )));
1642 }
1643 let effective_nprobe = params.nprobe.min(centroids.num_clusters as usize);
1644 let flat = lazy_flat.expect("ANN/flat pairing validated above");
1645 progressive_ann_search(
1646 fetch_k.min(flat.num_docs_with_vectors()),
1647 ann_fetch_k,
1648 index.len().min(flat.num_vectors),
1649 |candidate_k| {
1650 index
1651 .search(
1652 centroids,
1653 codebook,
1654 query,
1655 candidate_k,
1656 Some(effective_nprobe),
1657 )
1658 .into_iter()
1659 .map(|(doc_id, ordinal, dist)| {
1660 (doc_id, ordinal, 1.0 / (1.0 + dist))
1661 })
1662 .collect()
1663 },
1664 )?
1665 }
1666 VectorIndex::BinaryIvf(_) => {
1667 Vec::new()
1669 }
1670 }
1671 } else if let Some(lazy_flat) = lazy_flat {
1672 log::debug!(
1676 "[search_dense] field {}: brute-force on {} vectors (dim={}, quant={:?})",
1677 field.0,
1678 lazy_flat.num_vectors,
1679 lazy_flat.dim,
1680 lazy_flat.quantization
1681 );
1682 let dim = lazy_flat.dim;
1683 let n = lazy_flat.num_vectors;
1684 let quant = lazy_flat.quantization;
1685 let batch_len =
1686 bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
1687 let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
1688 let mut scores = vec![0f32; batch_len];
1689
1690 for batch_start in (0..n).step_by(batch_len) {
1691 let batch_count = batch_len.min(n - batch_start);
1692 let batch_bytes = lazy_flat
1693 .read_vectors_batch(batch_start, batch_count)
1694 .await
1695 .map_err(crate::Error::Io)?;
1696 let raw = batch_bytes.as_slice();
1697
1698 Self::score_quantized_batch(
1699 query,
1700 raw,
1701 quant,
1702 dim,
1703 &mut scores[..batch_count],
1704 params.unit_norm,
1705 )?;
1706
1707 for (i, &score) in scores.iter().enumerate().take(batch_count) {
1708 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1709 collector.push(doc_id, ordinal, score);
1710 }
1711 }
1712
1713 flat_results = Some(collector.into_results());
1714 Vec::new()
1715 } else {
1716 return Ok(Vec::new());
1717 };
1718 let l1_elapsed = t0.elapsed();
1719 {
1720 let kind = match ann_index {
1721 Some(VectorIndex::RaBitQ(_)) => "rabitq",
1722 Some(VectorIndex::IVF(_)) => "ivf_rabitq",
1723 Some(VectorIndex::ScaNN(_)) => "scann",
1724 Some(VectorIndex::BinaryIvf(_)) => "binary_ivf",
1725 None => "flat",
1726 };
1727 crate::observe::dense_l1(
1728 self.schema.index_label(),
1729 self.schema.get_field_name(field).unwrap_or("?"),
1730 kind,
1731 l1_elapsed.as_secs_f64(),
1732 flat_results.as_ref().map_or(results.len(), Vec::len),
1733 );
1734 }
1735 log::debug!(
1736 "[search_dense] field {}: L1 returned {} candidates in {:.1}ms",
1737 field.0,
1738 flat_results.as_ref().map_or(results.len(), Vec::len),
1739 l1_elapsed.as_secs_f64() * 1000.0
1740 );
1741
1742 if let Some(results) = flat_results {
1743 return Ok(results);
1744 }
1745
1746 if ann_index.is_some()
1749 && !results.is_empty()
1750 && let Some(lazy_flat) = lazy_flat
1751 {
1752 let t_rerank = std::time::Instant::now();
1753 let dim = lazy_flat.dim;
1754 let quant = lazy_flat.quantization;
1755 let vbs = lazy_flat.vector_byte_size();
1756
1757 let (expanded, mut resolved) = expand_ann_candidate_documents(&results, lazy_flat)?;
1761 results = expanded;
1762
1763 let t_resolve = t_rerank.elapsed();
1764 if !resolved.is_empty() {
1765 resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
1767
1768 let batch_len = bounded_vector_score_batch(vbs, DENSE_SCORE_BATCH);
1772 let max_batch = batch_len.min(resolved.len());
1773 let max_raw_len = max_batch
1774 .checked_mul(vbs)
1775 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
1776 let mut raw_buf = vec![0u8; max_raw_len];
1777 let mut scores = vec![0f32; max_batch];
1778 let mut read_elapsed = std::time::Duration::ZERO;
1779 let mut score_elapsed = std::time::Duration::ZERO;
1780
1781 for chunk in resolved.chunks(batch_len) {
1782 #[cfg(feature = "native")]
1786 lazy_flat.prefetch_vectors(chunk.iter().map(|&(_, flat_idx)| flat_idx));
1787 let raw_len = chunk
1788 .len()
1789 .checked_mul(vbs)
1790 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
1791 let raw = &mut raw_buf[..raw_len];
1792
1793 let t_read = std::time::Instant::now();
1794 for (buf_idx, &(_, flat_idx)) in chunk.iter().enumerate() {
1795 lazy_flat
1796 .read_vector_raw_into(
1797 flat_idx,
1798 &mut raw[buf_idx * vbs..(buf_idx + 1) * vbs],
1799 )
1800 .await
1801 .map_err(crate::Error::Io)?;
1802 }
1803 read_elapsed += t_read.elapsed();
1804
1805 let t_score = std::time::Instant::now();
1806 Self::score_quantized_batch(
1807 query,
1808 raw,
1809 quant,
1810 dim,
1811 &mut scores[..chunk.len()],
1812 params.unit_norm,
1813 )?;
1814 score_elapsed += t_score.elapsed();
1815
1816 for (buf_idx, &(ri, _)) in chunk.iter().enumerate() {
1817 results[ri].2 = scores[buf_idx];
1818 }
1819 }
1820
1821 crate::observe::dense_rerank(
1822 self.schema.index_label(),
1823 self.schema.get_field_name(field).unwrap_or("?"),
1824 t_rerank.elapsed().as_secs_f64(),
1825 t_resolve.as_secs_f64(),
1826 read_elapsed.as_secs_f64(),
1827 resolved.len(),
1828 );
1829 log::debug!(
1830 "[search_dense] field {}: rerank {} vectors (dim={}, quant={:?}, {}B/vec): resolve={:.1}ms read={:.1}ms score={:.1}ms",
1831 field.0,
1832 resolved.len(),
1833 dim,
1834 quant,
1835 vbs,
1836 t_resolve.as_secs_f64() * 1000.0,
1837 read_elapsed.as_secs_f64() * 1000.0,
1838 score_elapsed.as_secs_f64() * 1000.0,
1839 );
1840 }
1841
1842 log::debug!(
1843 "[search_dense] field {}: rerank total={:.1}ms",
1844 field.0,
1845 t_rerank.elapsed().as_secs_f64() * 1000.0
1846 );
1847 }
1848
1849 Ok(combine_grouped_ordinal_results(results, combiner, k))
1850 }
1851
1852 fn binary_ivf_nprobe(&self, field: Field) -> Option<usize> {
1854 self.schema
1855 .get_field_entry(field)
1856 .and_then(|e| e.binary_dense_vector_config.as_ref())
1857 .map(|c| c.nprobe)
1858 .filter(|&n| n > 0)
1859 }
1860
1861 pub async fn search_binary_dense_vector(
1865 &self,
1866 field: Field,
1867 query: &[u8],
1868 k: usize,
1869 combiner: crate::query::MultiValueCombiner,
1870 ) -> Result<Vec<VectorSearchResult>> {
1871 let schema_dim = self.validate_binary_search_request(field, query)?;
1872 combiner.validate().map_err(Error::Query)?;
1873 if k == 0 {
1874 return Ok(Vec::new());
1875 }
1876 let t0 = crate::observe::Timer::start();
1877 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0)
1881 && let Some(ivf) = lazy.get()
1882 {
1883 if ivf.config.dim_bits != schema_dim {
1884 return Err(Error::Corruption(format!(
1885 "binary IVF field {} has schema dimension {} but index dimension {}",
1886 field.0, schema_dim, ivf.config.dim_bits
1887 )));
1888 }
1889 let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
1890 Error::Corruption(format!(
1891 "binary IVF field {} is missing flat vector storage",
1892 field.0
1893 ))
1894 })?;
1895 if flat.dim != schema_dim
1896 || flat.quantization != crate::dsl::DenseVectorQuantization::Binary
1897 {
1898 return Err(Error::Corruption(format!(
1899 "binary IVF field {} has inconsistent flat vector metadata",
1900 field.0
1901 )));
1902 }
1903 let nprobe = self.binary_ivf_nprobe(field);
1904 let initial_fetch =
1905 ann_ordinal_fetch_k(k, flat.num_vectors, flat.num_docs_with_vectors());
1906 let ann_results = single_scan_ann_search(
1910 k.min(flat.num_docs_with_vectors()),
1911 initial_fetch,
1912 ivf.len().min(flat.num_vectors),
1913 |candidate_k| ivf.search(query, candidate_k, nprobe),
1914 )?;
1915 let results =
1916 exact_score_binary_candidate_documents(&ann_results, flat, query, schema_dim)
1917 .await?;
1918 crate::observe::dense_l1(
1919 self.schema.index_label(),
1920 self.schema.get_field_name(field).unwrap_or("?"),
1921 "binary_ivf",
1922 t0.secs(),
1923 results.len(),
1924 );
1925 return Ok(combine_grouped_ordinal_results(results, combiner, k));
1926 }
1927
1928 let lazy_flat = match self.flat_vectors.get(&field.0) {
1929 Some(f) => f,
1930 None => return Ok(Vec::new()),
1931 };
1932
1933 let dim_bits = lazy_flat.dim;
1934 let byte_len = lazy_flat.vector_byte_size();
1935 let n = lazy_flat.num_vectors;
1936
1937 if dim_bits != schema_dim {
1938 return Err(Error::Corruption(format!(
1939 "binary vector field {} has schema dimension {} but flat storage dimension {}",
1940 field.0, schema_dim, dim_bits
1941 )));
1942 }
1943
1944 if byte_len != query.len() {
1945 return Err(Error::Schema(format!(
1946 "Binary query vector byte length {} != field byte length {}",
1947 query.len(),
1948 byte_len
1949 )));
1950 }
1951
1952 let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
1953 let mut collector = FlatDocumentCollector::new(k, combiner);
1954 let mut scores = vec![0f32; batch_len];
1955
1956 for batch_start in (0..n).step_by(batch_len) {
1957 let batch_count = batch_len.min(n - batch_start);
1958 let batch_bytes = lazy_flat
1959 .read_vectors_batch(batch_start, batch_count)
1960 .await
1961 .map_err(crate::Error::Io)?;
1962 let raw = batch_bytes.as_slice();
1963
1964 crate::structures::simd::batch_hamming_scores(
1965 query,
1966 raw,
1967 byte_len,
1968 dim_bits,
1969 &mut scores[..batch_count],
1970 );
1971
1972 for (i, &score) in scores.iter().enumerate().take(batch_count) {
1973 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1974 collector.push(doc_id, ordinal, score);
1975 }
1976 }
1977
1978 let results = collector.into_results();
1979
1980 crate::observe::dense_l1(
1981 self.schema.index_label(),
1982 self.schema.get_field_name(field).unwrap_or("?"),
1983 "binary_flat",
1984 t0.secs(),
1985 results.len(),
1986 );
1987 Ok(results)
1988 }
1989
1990 pub fn has_dense_vector_index(&self, field: Field) -> bool {
1992 self.vector_indexes.contains_key(&field.0) || self.flat_vectors.contains_key(&field.0)
1993 }
1994
1995 pub fn get_dense_vector_index(&self, field: Field) -> Option<Arc<RaBitQIndex>> {
1997 match self.vector_indexes.get(&field.0) {
1998 Some(VectorIndex::RaBitQ(lazy)) => lazy.get().cloned(),
1999 _ => None,
2000 }
2001 }
2002
2003 pub fn get_ivf_vector_index(
2005 &self,
2006 field: Field,
2007 ) -> Option<(Arc<IVFRaBitQIndex>, Arc<crate::structures::RaBitQCodebook>)> {
2008 match self.vector_indexes.get(&field.0) {
2009 Some(VectorIndex::IVF(lazy)) => lazy.get().map(|(i, c)| (i.clone(), c.clone())),
2010 _ => None,
2011 }
2012 }
2013
2014 pub fn coarse_centroids(&self, field_id: u32) -> Option<&Arc<CoarseCentroids>> {
2016 self.coarse_centroids.get(&field_id)
2017 }
2018
2019 pub fn set_coarse_centroids(&mut self, centroids: FxHashMap<u32, Arc<CoarseCentroids>>) {
2021 self.coarse_centroids = centroids;
2022 }
2023
2024 pub fn get_scann_vector_index(
2026 &self,
2027 field: Field,
2028 ) -> Option<(Arc<IVFPQIndex>, Arc<PQCodebook>)> {
2029 match self.vector_indexes.get(&field.0) {
2030 Some(VectorIndex::ScaNN(lazy)) => lazy.get().map(|(i, c)| (i.clone(), c.clone())),
2031 _ => None,
2032 }
2033 }
2034
2035 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
2037 self.vector_indexes.get(&field.0)
2038 }
2039
2040 pub async fn get_positions(
2045 &self,
2046 field: Field,
2047 term: &[u8],
2048 ) -> Result<Option<crate::structures::PositionPostingList>> {
2049 let handle = match &self.positions_handle {
2051 Some(h) => h,
2052 None => return Ok(None),
2053 };
2054
2055 let mut key = Vec::with_capacity(4 + term.len());
2057 key.extend_from_slice(&field.0.to_le_bytes());
2058 key.extend_from_slice(term);
2059
2060 let term_info = match self.term_dict.get(&key).await? {
2062 Some(info) => info,
2063 None => return Ok(None),
2064 };
2065
2066 let (offset, length) = match term_info.position_info() {
2068 Some((o, l)) => (o, l),
2069 None => return Ok(None),
2070 };
2071
2072 let range = checked_file_range(offset, length, handle.len(), "position list")?;
2076 let slice = handle.slice(range);
2077 let data = slice.read_bytes().await?;
2078
2079 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2081
2082 Ok(Some(pos_list))
2083 }
2084
2085 pub fn has_positions(&self, field: Field) -> bool {
2087 if let Some(entry) = self.schema.get_field_entry(field) {
2089 entry.positions.is_some()
2090 } else {
2091 false
2092 }
2093 }
2094}
2095
2096#[cfg(feature = "sync")]
2098impl SegmentReader {
2099 pub fn get_postings_sync(&self, field: Field, term: &[u8]) -> Result<Option<BlockPostingList>> {
2101 let mut key = Vec::with_capacity(4 + term.len());
2103 key.extend_from_slice(&field.0.to_le_bytes());
2104 key.extend_from_slice(term);
2105
2106 let term_info = match self.term_dict.get_sync(&key)? {
2108 Some(info) => info,
2109 None => return Ok(None),
2110 };
2111
2112 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2114 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2115 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2116 posting_list.push(doc_id, tf);
2117 }
2118 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
2119 return Ok(Some(block_list));
2120 }
2121
2122 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
2124 Error::Corruption("TermInfo has neither inline nor external data".to_string())
2125 })?;
2126
2127 let range = checked_file_range(
2128 posting_offset,
2129 posting_len,
2130 self.postings_handle.len(),
2131 "posting",
2132 )?;
2133 let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
2134 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
2135
2136 Ok(Some(block_list))
2137 }
2138
2139 pub fn get_prefix_postings_sync(
2141 &self,
2142 field: Field,
2143 prefix: &[u8],
2144 ) -> Result<Vec<BlockPostingList>> {
2145 if prefix.is_empty() {
2146 return Err(Error::Query("prefix must not be empty".into()));
2147 }
2148 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
2149 key_prefix.extend_from_slice(&field.0.to_le_bytes());
2150 key_prefix.extend_from_slice(prefix);
2151
2152 let (entries, truncated) = self
2153 .term_dict
2154 .prefix_scan_limited_sync(&key_prefix, MAX_PREFIX_TERMS)?;
2155 if truncated {
2156 return Err(Error::Query(format!(
2157 "prefix expands to more than {MAX_PREFIX_TERMS} terms"
2158 )));
2159 }
2160 let posting_count: u64 = entries
2161 .iter()
2162 .map(|(_, term_info)| term_info.doc_freq() as u64)
2163 .sum();
2164 if posting_count > MAX_PREFIX_POSTINGS {
2165 return Err(Error::Query(format!(
2166 "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
2167 )));
2168 }
2169 let mut results = Vec::with_capacity(entries.len());
2170
2171 for (_key, term_info) in entries {
2172 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2173 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2174 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2175 posting_list.push(doc_id, tf);
2176 }
2177 results.push(BlockPostingList::from_posting_list(&posting_list)?);
2178 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
2179 let range = checked_file_range(
2180 posting_offset,
2181 posting_len,
2182 self.postings_handle.len(),
2183 "prefix posting",
2184 )?;
2185 let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
2186 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
2187 }
2188 }
2189
2190 Ok(results)
2191 }
2192
2193 pub fn get_positions_sync(
2195 &self,
2196 field: Field,
2197 term: &[u8],
2198 ) -> Result<Option<crate::structures::PositionPostingList>> {
2199 let handle = match &self.positions_handle {
2200 Some(h) => h,
2201 None => return Ok(None),
2202 };
2203
2204 let mut key = Vec::with_capacity(4 + term.len());
2206 key.extend_from_slice(&field.0.to_le_bytes());
2207 key.extend_from_slice(term);
2208
2209 let term_info = match self.term_dict.get_sync(&key)? {
2211 Some(info) => info,
2212 None => return Ok(None),
2213 };
2214
2215 let (offset, length) = match term_info.position_info() {
2216 Some((o, l)) => (o, l),
2217 None => return Ok(None),
2218 };
2219
2220 let range = checked_file_range(offset, length, handle.len(), "position list")?;
2221 let slice = handle.slice(range);
2222 let data = slice.read_bytes_sync()?;
2223
2224 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2225 Ok(Some(pos_list))
2226 }
2227
2228 pub fn search_dense_vector_sync(
2231 &self,
2232 field: Field,
2233 query: &[f32],
2234 k: usize,
2235 nprobe: usize,
2236 rerank_factor: f32,
2237 combiner: crate::query::MultiValueCombiner,
2238 ) -> Result<Vec<VectorSearchResult>> {
2239 let params =
2240 self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
2241 let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
2242 if k == 0 {
2243 return Ok(Vec::new());
2244 }
2245
2246 let ann_index = self.vector_indexes.get(&field.0);
2247 let lazy_flat = self.flat_vectors.get(&field.0);
2248 let ann_fetch_k = lazy_flat.map_or(fetch_k, |flat| {
2249 ann_ordinal_fetch_k(fetch_k, flat.num_vectors, flat.num_docs_with_vectors())
2250 });
2251
2252 if ann_index.is_none() && lazy_flat.is_none() {
2253 return Ok(Vec::new());
2254 }
2255
2256 if ann_index.is_some() && lazy_flat.is_none() {
2257 return Err(Error::Corruption(format!(
2258 "dense ANN field {} is missing flat vector storage",
2259 field.0
2260 )));
2261 }
2262
2263 if let Some(flat) = lazy_flat
2264 && flat.dim != params.dim
2265 {
2266 return Err(Error::Corruption(format!(
2267 "dense vector field {} has schema dimension {} but flat storage dimension {}",
2268 field.0, params.dim, flat.dim
2269 )));
2270 }
2271
2272 let mut results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
2273 match index {
2275 VectorIndex::RaBitQ(lazy) => {
2276 let rabitq = lazy.get().ok_or_else(|| {
2277 Error::Schema("RaBitQ index deserialization failed".to_string())
2278 })?;
2279 if rabitq.codebook.config.dim != params.dim {
2280 return Err(Error::Corruption(format!(
2281 "RaBitQ index dimension {} does not match schema dimension {}",
2282 rabitq.codebook.config.dim, params.dim
2283 )));
2284 }
2285 let flat = lazy_flat.expect("ANN/flat pairing validated above");
2286 progressive_ann_search(
2287 fetch_k.min(flat.num_docs_with_vectors()),
2288 ann_fetch_k,
2289 rabitq.len().min(flat.num_vectors),
2290 |candidate_k| {
2291 rabitq
2292 .search(query, candidate_k)
2293 .into_iter()
2294 .map(|(doc_id, ordinal, dist)| {
2295 (doc_id, ordinal, 1.0 / (1.0 + dist))
2296 })
2297 .collect()
2298 },
2299 )?
2300 }
2301 VectorIndex::IVF(lazy) => {
2302 let (index, codebook) = lazy.get().ok_or_else(|| {
2303 Error::Schema("IVF index deserialization failed".to_string())
2304 })?;
2305 let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
2306 Error::Schema(format!(
2307 "IVF index requires coarse centroids for field {}",
2308 field.0
2309 ))
2310 })?;
2311 validate_coarse_centroids(centroids, params.dim)?;
2312 if index.config.dim != params.dim
2313 || codebook.config.dim != params.dim
2314 || index.centroids_version != centroids.version
2315 || index.codebook_version != codebook.version
2316 || index
2317 .clusters
2318 .iter()
2319 .any(|(cluster_id, _)| cluster_id >= centroids.num_clusters)
2320 {
2321 return Err(Error::Corruption(format!(
2322 "IVF index/codebook/centroid metadata does not match schema dimension {}",
2323 params.dim
2324 )));
2325 }
2326 let effective_nprobe = params.nprobe.min(centroids.num_clusters as usize);
2327 let flat = lazy_flat.expect("ANN/flat pairing validated above");
2328 progressive_ann_search(
2329 fetch_k.min(flat.num_docs_with_vectors()),
2330 ann_fetch_k,
2331 index.len().min(flat.num_vectors),
2332 |candidate_k| {
2333 index
2334 .search(
2335 centroids,
2336 codebook,
2337 query,
2338 candidate_k,
2339 Some(effective_nprobe),
2340 )
2341 .into_iter()
2342 .map(|(doc_id, ordinal, dist)| {
2343 (doc_id, ordinal, 1.0 / (1.0 + dist))
2344 })
2345 .collect()
2346 },
2347 )?
2348 }
2349 VectorIndex::ScaNN(lazy) => {
2350 let (index, codebook) = lazy.get().ok_or_else(|| {
2351 Error::Schema("ScaNN index deserialization failed".to_string())
2352 })?;
2353 let centroids = self.coarse_centroids.get(&field.0).ok_or_else(|| {
2354 Error::Schema(format!(
2355 "ScaNN index requires coarse centroids for field {}",
2356 field.0
2357 ))
2358 })?;
2359 validate_coarse_centroids(centroids, params.dim)?;
2360 if index.config.dim != params.dim
2361 || codebook.config.dim != params.dim
2362 || index.centroids_version != centroids.version
2363 || index.codebook_version != codebook.version
2364 || index
2365 .clusters
2366 .iter()
2367 .any(|(cluster_id, _)| cluster_id >= centroids.num_clusters)
2368 {
2369 return Err(Error::Corruption(format!(
2370 "ScaNN index/codebook/centroid metadata does not match schema dimension {}",
2371 params.dim
2372 )));
2373 }
2374 let effective_nprobe = params.nprobe.min(centroids.num_clusters as usize);
2375 let flat = lazy_flat.expect("ANN/flat pairing validated above");
2376 progressive_ann_search(
2377 fetch_k.min(flat.num_docs_with_vectors()),
2378 ann_fetch_k,
2379 index.len().min(flat.num_vectors),
2380 |candidate_k| {
2381 index
2382 .search(
2383 centroids,
2384 codebook,
2385 query,
2386 candidate_k,
2387 Some(effective_nprobe),
2388 )
2389 .into_iter()
2390 .map(|(doc_id, ordinal, dist)| {
2391 (doc_id, ordinal, 1.0 / (1.0 + dist))
2392 })
2393 .collect()
2394 },
2395 )?
2396 }
2397 VectorIndex::BinaryIvf(_) => {
2398 Vec::new()
2400 }
2401 }
2402 } else if let Some(lazy_flat) = lazy_flat {
2403 let dim = lazy_flat.dim;
2405 let n = lazy_flat.num_vectors;
2406 let quant = lazy_flat.quantization;
2407 let batch_len =
2408 bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
2409 let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
2410 let mut scores = vec![0f32; batch_len];
2411
2412 for batch_start in (0..n).step_by(batch_len) {
2413 let batch_count = batch_len.min(n - batch_start);
2414 let batch_bytes = lazy_flat
2415 .read_vectors_batch_sync(batch_start, batch_count)
2416 .map_err(crate::Error::Io)?;
2417 let raw = batch_bytes.as_slice();
2418
2419 Self::score_quantized_batch(
2420 query,
2421 raw,
2422 quant,
2423 dim,
2424 &mut scores[..batch_count],
2425 params.unit_norm,
2426 )?;
2427
2428 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2429 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2430 collector.push(doc_id, ordinal, score);
2431 }
2432 }
2433
2434 return Ok(collector.into_results());
2435 } else {
2436 return Ok(Vec::new());
2437 };
2438
2439 if ann_index.is_some()
2441 && !results.is_empty()
2442 && let Some(lazy_flat) = lazy_flat
2443 {
2444 let dim = lazy_flat.dim;
2445 let quant = lazy_flat.quantization;
2446 let vbs = lazy_flat.vector_byte_size();
2447
2448 let (expanded, mut resolved) = expand_ann_candidate_documents(&results, lazy_flat)?;
2449 results = expanded;
2450
2451 if !resolved.is_empty() {
2452 resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
2453 let batch_len = bounded_vector_score_batch(vbs, DENSE_SCORE_BATCH);
2454 let max_batch = batch_len.min(resolved.len());
2455 let max_raw_len = max_batch
2456 .checked_mul(vbs)
2457 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
2458 let mut raw_buf = vec![0u8; max_raw_len];
2459 let mut scores = vec![0f32; max_batch];
2460
2461 for chunk in resolved.chunks(batch_len) {
2462 let raw_len = chunk
2463 .len()
2464 .checked_mul(vbs)
2465 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
2466 let raw = &mut raw_buf[..raw_len];
2467 for (buf_idx, &(_, flat_idx)) in chunk.iter().enumerate() {
2468 lazy_flat
2469 .read_vector_raw_into_sync(
2470 flat_idx,
2471 &mut raw[buf_idx * vbs..(buf_idx + 1) * vbs],
2472 )
2473 .map_err(crate::Error::Io)?;
2474 }
2475
2476 Self::score_quantized_batch(
2477 query,
2478 raw,
2479 quant,
2480 dim,
2481 &mut scores[..chunk.len()],
2482 params.unit_norm,
2483 )?;
2484
2485 for (buf_idx, &(ri, _)) in chunk.iter().enumerate() {
2486 results[ri].2 = scores[buf_idx];
2487 }
2488 }
2489 }
2490 }
2491
2492 Ok(combine_grouped_ordinal_results(results, combiner, k))
2493 }
2494
2495 #[cfg(feature = "sync")]
2500 pub fn search_binary_dense_vector_sync(
2501 &self,
2502 field: Field,
2503 query: &[u8],
2504 k: usize,
2505 combiner: crate::query::MultiValueCombiner,
2506 ) -> Result<Vec<VectorSearchResult>> {
2507 let schema_dim = self.validate_binary_search_request(field, query)?;
2508 combiner.validate().map_err(Error::Query)?;
2509 if k == 0 {
2510 return Ok(Vec::new());
2511 }
2512 let t0 = crate::observe::Timer::start();
2513 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0)
2515 && let Some(ivf) = lazy.get()
2516 {
2517 if ivf.config.dim_bits != schema_dim {
2518 return Err(Error::Corruption(format!(
2519 "binary IVF field {} has schema dimension {} but index dimension {}",
2520 field.0, schema_dim, ivf.config.dim_bits
2521 )));
2522 }
2523 let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
2524 Error::Corruption(format!(
2525 "binary IVF field {} is missing flat vector storage",
2526 field.0
2527 ))
2528 })?;
2529 if flat.dim != schema_dim
2530 || flat.quantization != crate::dsl::DenseVectorQuantization::Binary
2531 {
2532 return Err(Error::Corruption(format!(
2533 "binary IVF field {} has inconsistent flat vector metadata",
2534 field.0
2535 )));
2536 }
2537 let nprobe = self.binary_ivf_nprobe(field);
2538 let initial_fetch =
2539 ann_ordinal_fetch_k(k, flat.num_vectors, flat.num_docs_with_vectors());
2540 let ann_results = single_scan_ann_search(
2544 k.min(flat.num_docs_with_vectors()),
2545 initial_fetch,
2546 ivf.len().min(flat.num_vectors),
2547 |candidate_k| ivf.search(query, candidate_k, nprobe),
2548 )?;
2549 let results =
2550 exact_score_binary_candidate_documents_sync(&ann_results, flat, query, schema_dim)?;
2551 crate::observe::dense_l1(
2552 self.schema.index_label(),
2553 self.schema.get_field_name(field).unwrap_or("?"),
2554 "binary_ivf",
2555 t0.secs(),
2556 results.len(),
2557 );
2558 return Ok(combine_grouped_ordinal_results(results, combiner, k));
2559 }
2560
2561 let lazy_flat = match self.flat_vectors.get(&field.0) {
2562 Some(f) => f,
2563 None => return Ok(Vec::new()),
2564 };
2565
2566 let dim_bits = lazy_flat.dim;
2567 let byte_len = lazy_flat.vector_byte_size();
2568 let n = lazy_flat.num_vectors;
2569
2570 if dim_bits != schema_dim {
2571 return Err(Error::Corruption(format!(
2572 "binary vector field {} has schema dimension {} but flat storage dimension {}",
2573 field.0, schema_dim, dim_bits
2574 )));
2575 }
2576
2577 if byte_len != query.len() {
2578 return Err(Error::Schema(format!(
2579 "Binary query vector byte length {} != field byte length {}",
2580 query.len(),
2581 byte_len
2582 )));
2583 }
2584
2585 let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
2586 let mut collector = FlatDocumentCollector::new(k, combiner);
2587 let mut scores = vec![0f32; batch_len];
2588
2589 for batch_start in (0..n).step_by(batch_len) {
2590 let batch_count = batch_len.min(n - batch_start);
2591 let batch_bytes = lazy_flat
2592 .read_vectors_batch_sync(batch_start, batch_count)
2593 .map_err(crate::Error::Io)?;
2594 let raw = batch_bytes.as_slice();
2595
2596 crate::structures::simd::batch_hamming_scores(
2597 query,
2598 raw,
2599 byte_len,
2600 dim_bits,
2601 &mut scores[..batch_count],
2602 );
2603
2604 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2605 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2606 collector.push(doc_id, ordinal, score);
2607 }
2608 }
2609
2610 let results = collector.into_results();
2611
2612 crate::observe::dense_l1(
2613 self.schema.index_label(),
2614 self.schema.get_field_name(field).unwrap_or("?"),
2615 "binary_flat",
2616 t0.secs(),
2617 results.len(),
2618 );
2619 Ok(results)
2620 }
2621}
2622
2623#[cfg(test)]
2624mod dense_search_safety_tests {
2625 use super::*;
2626
2627 #[test]
2628 fn dense_fetch_count_rejects_non_finite_and_unbounded_factors() {
2629 for factor in [
2630 f32::NAN,
2631 f32::INFINITY,
2632 f32::NEG_INFINITY,
2633 0.0,
2634 0.5,
2635 MAX_DENSE_RERANK_FACTOR + 1.0,
2636 ] {
2637 assert!(
2638 checked_dense_fetch_k(10, factor).is_err(),
2639 "factor={factor}"
2640 );
2641 }
2642 }
2643
2644 #[test]
2645 fn flat_document_collector_does_not_let_one_multivalue_doc_crowd_out_others() {
2646 let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
2647 collector.push(1, 0, 1.0);
2648 collector.push(1, 1, 0.9);
2649 collector.push(2, 0, 0.8);
2650
2651 let results = collector.into_results();
2652 assert_eq!(
2653 results
2654 .iter()
2655 .map(|result| result.doc_id)
2656 .collect::<Vec<_>>(),
2657 vec![1, 2]
2658 );
2659 assert_eq!(results[0].ordinals.len(), 2);
2660 }
2661
2662 #[test]
2663 fn flat_document_collector_evicts_by_score_then_doc_id() {
2664 let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
2665 collector.push(1, 0, 0.5);
2666 collector.push(3, 0, 0.8);
2667 collector.push(2, 0, 0.9);
2668 let results = collector.into_results();
2669 assert_eq!(
2670 results
2671 .iter()
2672 .map(|result| result.doc_id)
2673 .collect::<Vec<_>>(),
2674 vec![2, 3]
2675 );
2676
2677 let mut tied = FlatDocumentCollector::new(1, crate::query::MultiValueCombiner::Max);
2678 tied.push(2, 0, 1.0);
2679 tied.push(1, 0, 1.0);
2680 let results = tied.into_results();
2681 assert_eq!(results[0].doc_id, 1);
2682 }
2683
2684 #[test]
2685 fn dense_fetch_count_rounds_up_and_detects_overflow() {
2686 assert_eq!(checked_dense_fetch_k(3, 1.5).unwrap(), 5);
2687 assert_eq!(checked_dense_fetch_k(50_000, 3.0).unwrap(), 150_000);
2688 assert!(checked_dense_fetch_k(50_000, 32.0).is_err());
2689 assert!(checked_dense_fetch_k(usize::MAX, 2.0).is_err());
2690 }
2691
2692 #[test]
2693 fn ann_fetch_depth_accounts_for_multivalue_density_and_stays_bounded() {
2694 assert_eq!(ann_ordinal_fetch_k(100, 1_000, 100), 1_000);
2695 assert_eq!(
2696 ann_ordinal_fetch_k(100_000, usize::MAX, 1),
2697 MAX_DENSE_CANDIDATES_PER_SEGMENT
2698 );
2699 assert_eq!(ann_ordinal_fetch_k(100, 0, 10), 0);
2700 }
2701
2702 #[test]
2703 fn ann_search_deepens_until_skewed_results_contain_enough_documents() {
2704 let ranked = [
2705 (1, 0, 1.0),
2706 (1, 1, 0.99),
2707 (1, 2, 0.98),
2708 (1, 3, 0.97),
2709 (1, 4, 0.96),
2710 (1, 5, 0.95),
2711 (2, 0, 0.9),
2712 (3, 0, 0.8),
2713 ];
2714 let mut fetches = Vec::new();
2715 let results = progressive_ann_search(2, 2, ranked.len(), |fetch| {
2716 fetches.push(fetch);
2717 ranked.iter().copied().take(fetch).collect()
2718 })
2719 .unwrap();
2720
2721 assert_eq!(fetches, vec![2, 4, 8]);
2722 assert!(results.iter().any(|&(doc_id, _, _)| doc_id == 2));
2723 }
2724
2725 #[test]
2726 fn ann_search_stops_when_probed_population_is_exhausted() {
2727 let ranked = [(1, 0, 1.0), (1, 1, 0.9)];
2728 let mut calls = 0;
2729 let results = progressive_ann_search(3, 4, 100, |fetch| {
2730 calls += 1;
2731 ranked.iter().copied().take(fetch).collect()
2732 })
2733 .unwrap();
2734
2735 assert_eq!(calls, 1);
2736 assert_eq!(results.len(), 2);
2737 }
2738
2739 #[test]
2740 fn file_ranges_reject_overflow_and_truncation() {
2741 assert_eq!(checked_file_range(4, 3, 7, "test").unwrap(), 4..7);
2742 assert!(checked_file_range(u64::MAX, 1, u64::MAX, "test").is_err());
2743 assert!(checked_file_range(5, 3, 7, "test").is_err());
2744 }
2745
2746 #[test]
2751 fn single_scan_ann_search_matches_progressive_with_one_search_call() {
2752 let ranked: Vec<RawVectorCandidate> = vec![
2754 (1, 0, 1.0),
2755 (1, 1, 0.99),
2756 (1, 2, 0.98),
2757 (1, 3, 0.97),
2758 (1, 4, 0.96),
2759 (1, 5, 0.95),
2760 (2, 0, 0.9),
2761 (3, 0, 0.8),
2762 (2, 1, 0.7),
2763 (4, 0, 0.6),
2764 (5, 0, 0.5),
2765 ];
2766
2767 for target_docs in 0..=6 {
2768 for initial_fetch in [0, 1, 2, 3, 8, 16, 64] {
2769 for max_vectors in [0, 1, 2, 8, ranked.len(), 100] {
2770 let progressive =
2771 progressive_ann_search(target_docs, initial_fetch, max_vectors, |fetch| {
2772 ranked.iter().copied().take(fetch).collect()
2773 });
2774 let mut calls = 0;
2775 let single =
2776 single_scan_ann_search(target_docs, initial_fetch, max_vectors, |fetch| {
2777 calls += 1;
2778 ranked.iter().copied().take(fetch).collect()
2779 });
2780 let case =
2781 format!("target={target_docs} initial={initial_fetch} max={max_vectors}");
2782 match (progressive, single) {
2783 (Ok(a), Ok(b)) => assert_eq!(a, b, "{case}"),
2784 (Err(a), Err(b)) => assert_eq!(a.to_string(), b.to_string(), "{case}"),
2785 (a, b) => panic!("verdict mismatch for {case}: {a:?} vs {b:?}"),
2786 }
2787 assert!(calls <= 1, "{case}: search must run at most once");
2788 }
2789 }
2790 }
2791 }
2792
2793 #[test]
2797 fn probe_scored_candidates_skip_flat_rereads() {
2798 let ann_results: Vec<RawVectorCandidate> = vec![(7, 0, 0.75), (9, 2, 0.5)];
2799 let mut expanded: Vec<RawVectorCandidate> =
2801 vec![(7, 0, 0.0), (7, 1, 0.0), (9, 0, 0.0), (9, 2, 0.0)];
2802 let resolved: Vec<ResolvedVectorCandidate> = vec![(0, 10), (1, 11), (2, 20), (3, 22)];
2803
2804 let remaining = fill_probe_scores_and_prune_resolved(&ann_results, &mut expanded, resolved);
2805
2806 assert_eq!(expanded[0].2, 0.75);
2808 assert_eq!(expanded[3].2, 0.5);
2809 assert_eq!(remaining, vec![(1, 11), (2, 20)]);
2811 assert_eq!(expanded[1].2, 0.0);
2812 assert_eq!(expanded[2].2, 0.0);
2813 }
2814}