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, IVFPQIndex, 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 ResolvedVectorCandidate = (usize, usize); fn expand_ann_candidate_documents(
324 ann_results: &[RawVectorCandidate],
325 flat: &LazyFlatVectorData,
326) -> Result<(Vec<RawVectorCandidate>, Vec<ResolvedVectorCandidate>)> {
327 let mut candidate_docs: Vec<DocId> = ann_results.iter().map(|candidate| candidate.0).collect();
328 candidate_docs.sort_unstable();
329 candidate_docs.dedup();
330
331 let mut expanded = Vec::new();
332 let mut resolved = Vec::new();
333 for doc_id in candidate_docs {
334 let (start, count) = flat.flat_indexes_for_doc_range(doc_id);
335 if count == 0 {
336 return Err(Error::Corruption(format!(
337 "ANN candidate document {doc_id} is missing from flat vector storage"
338 )));
339 }
340 let next_len = expanded
341 .len()
342 .checked_add(count)
343 .ok_or_else(|| Error::Query("ANN candidate vector expansion overflow".to_string()))?;
344 if next_len > MAX_DENSE_CANDIDATES_PER_SEGMENT {
345 return Err(Error::Query(format!(
346 "ANN candidate documents expand to more than \
347 {MAX_DENSE_CANDIDATES_PER_SEGMENT} vectors in one segment"
348 )));
349 }
350 expanded.reserve(count);
351 resolved.reserve(count);
352 let end = start
353 .checked_add(count)
354 .ok_or_else(|| Error::Corruption("flat vector range overflow".to_string()))?;
355 for flat_index in start..end {
356 let (stored_doc_id, ordinal) = flat.get_doc_id(flat_index);
357 if stored_doc_id != doc_id {
358 return Err(Error::Corruption(format!(
359 "flat vector doc map is not contiguous for document {doc_id}"
360 )));
361 }
362 let result_index = expanded.len();
363 expanded.push((doc_id, ordinal, 0.0));
364 resolved.push((result_index, flat_index));
365 }
366 }
367 Ok((expanded, resolved))
368}
369
370fn fill_probe_scores_and_prune_resolved(
379 ann_results: &[RawVectorCandidate],
380 expanded: &mut [RawVectorCandidate],
381 resolved: Vec<ResolvedVectorCandidate>,
382) -> Vec<ResolvedVectorCandidate> {
383 let mut probe_scores: FxHashMap<(DocId, u16), f32> =
384 FxHashMap::with_capacity_and_hasher(ann_results.len(), Default::default());
385 for &(doc_id, ordinal, score) in ann_results {
386 probe_scores.insert((doc_id, ordinal), score);
387 }
388 resolved
389 .into_iter()
390 .filter(|&(result_index, _)| {
391 let (doc_id, ordinal, _) = expanded[result_index];
392 match probe_scores.get(&(doc_id, ordinal)) {
393 Some(&score) => {
394 expanded[result_index].2 = score;
395 false
396 }
397 None => true,
398 }
399 })
400 .collect()
401}
402
403async fn exact_score_binary_candidate_documents(
404 ann_results: &[RawVectorCandidate],
405 flat: &LazyFlatVectorData,
406 query: &[u8],
407 dim_bits: usize,
408) -> Result<Vec<RawVectorCandidate>> {
409 let (mut expanded, resolved) = expand_ann_candidate_documents(ann_results, flat)?;
410 let resolved = fill_probe_scores_and_prune_resolved(ann_results, &mut expanded, resolved);
411 let vector_byte_size = flat.vector_byte_size();
412 let batch_len = bounded_vector_score_batch(vector_byte_size, BINARY_SCORE_BATCH);
413 let raw_capacity = batch_len
414 .checked_mul(vector_byte_size)
415 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
416 let mut raw = vec![0u8; raw_capacity];
417 let mut scores = vec![0.0f32; batch_len];
418
419 for chunk in resolved.chunks(batch_len) {
420 #[cfg(feature = "native")]
421 flat.prefetch_vectors(chunk.iter().map(|&(_, flat_index)| flat_index));
422 let raw_len = chunk
423 .len()
424 .checked_mul(vector_byte_size)
425 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
426 let raw = &mut raw[..raw_len];
427 for (buffer_index, &(_, flat_index)) in chunk.iter().enumerate() {
428 flat.read_vector_raw_into(
429 flat_index,
430 &mut raw[buffer_index * vector_byte_size..(buffer_index + 1) * vector_byte_size],
431 )
432 .await
433 .map_err(Error::Io)?;
434 }
435 crate::structures::simd::batch_hamming_scores(
436 query,
437 raw,
438 vector_byte_size,
439 dim_bits,
440 &mut scores[..chunk.len()],
441 );
442 for (buffer_index, &(result_index, _)) in chunk.iter().enumerate() {
443 expanded[result_index].2 = scores[buffer_index];
444 }
445 }
446 Ok(expanded)
447}
448
449#[cfg(feature = "sync")]
450fn exact_score_binary_candidate_documents_sync(
451 ann_results: &[RawVectorCandidate],
452 flat: &LazyFlatVectorData,
453 query: &[u8],
454 dim_bits: usize,
455) -> Result<Vec<RawVectorCandidate>> {
456 let (mut expanded, resolved) = expand_ann_candidate_documents(ann_results, flat)?;
457 let resolved = fill_probe_scores_and_prune_resolved(ann_results, &mut expanded, resolved);
458 let vector_byte_size = flat.vector_byte_size();
459 let batch_len = bounded_vector_score_batch(vector_byte_size, BINARY_SCORE_BATCH);
460 let raw_capacity = batch_len
461 .checked_mul(vector_byte_size)
462 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
463 let mut raw = vec![0u8; raw_capacity];
464 let mut scores = vec![0.0f32; batch_len];
465
466 for chunk in resolved.chunks(batch_len) {
467 let raw_len = chunk
468 .len()
469 .checked_mul(vector_byte_size)
470 .ok_or_else(|| Error::Query("binary candidate buffer size overflow".to_string()))?;
471 let raw = &mut raw[..raw_len];
472 for (buffer_index, &(_, flat_index)) in chunk.iter().enumerate() {
473 flat.read_vector_raw_into_sync(
474 flat_index,
475 &mut raw[buffer_index * vector_byte_size..(buffer_index + 1) * vector_byte_size],
476 )
477 .map_err(Error::Io)?;
478 }
479 crate::structures::simd::batch_hamming_scores(
480 query,
481 raw,
482 vector_byte_size,
483 dim_bits,
484 &mut scores[..chunk.len()],
485 );
486 for (buffer_index, &(result_index, _)) in chunk.iter().enumerate() {
487 expanded[result_index].2 = scores[buffer_index];
488 }
489 }
490 Ok(expanded)
491}
492
493fn validate_coarse_centroids(centroids: &CoarseCentroids, dim: usize) -> Result<()> {
494 let expected = (centroids.num_clusters as usize)
495 .checked_mul(dim)
496 .ok_or_else(|| Error::Corruption("coarse centroid size overflow".into()))?;
497 if centroids.num_clusters == 0
498 || centroids.dim != dim
499 || centroids.centroids.len() != expected
500 || centroids.centroids.iter().any(|value| !value.is_finite())
501 {
502 return Err(Error::Corruption(format!(
503 "invalid coarse centroids: clusters={}, dim={}, values={} (expected dim={dim}, values={expected})",
504 centroids.num_clusters,
505 centroids.dim,
506 centroids.centroids.len()
507 )));
508 }
509 Ok(())
510}
511
512fn float_query_plan(
513 centroids: &CoarseCentroids,
514 codebook: &crate::structures::PQCodebook,
515 query: &[f32],
516 nprobe: usize,
517 routing: crate::dsl::IvfRoutingMode,
518 cache: Option<&std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>>,
519) -> Result<std::sync::Arc<crate::structures::IvfPqQueryPlan>> {
520 let effective_nprobe = nprobe.clamp(1, centroids.num_clusters as usize);
521 let request_fingerprint = crate::structures::vector::ivf::routing::float_probe_fingerprint(
522 query,
523 effective_nprobe,
524 routing,
525 );
526 if let Some(cache) = cache {
527 let mut cached = cache
528 .lock()
529 .map_err(|_| Error::Internal("dense IVF probe cache is poisoned".into()))?;
530 if let Some(plan) = cached.as_ref()
531 && plan.quantizer_version == centroids.version
532 && plan.codebook_version == codebook.version
533 && plan.request_fingerprint == request_fingerprint
534 && plan.cluster_ids.len() == effective_nprobe
535 {
536 return Ok(std::sync::Arc::clone(plan));
537 }
538 let plan = std::sync::Arc::new(crate::structures::IvfPqQueryPlan::build(
539 centroids,
540 codebook,
541 query,
542 effective_nprobe,
543 routing,
544 ));
545 *cached = Some(std::sync::Arc::clone(&plan));
546 return Ok(plan);
547 }
548 Ok(std::sync::Arc::new(
549 crate::structures::IvfPqQueryPlan::build(
550 centroids,
551 codebook,
552 query,
553 effective_nprobe,
554 routing,
555 ),
556 ))
557}
558
559fn binary_probe_clusters(
560 quantizer: &crate::structures::BinaryCoarseQuantizer,
561 query: &[u8],
562 nprobe: usize,
563 routing: crate::dsl::IvfRoutingMode,
564 cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
565) -> Result<std::sync::Arc<[u32]>> {
566 let effective_nprobe = nprobe.clamp(1, quantizer.num_clusters as usize);
567 let request_fingerprint = crate::structures::vector::ivf::routing::binary_probe_fingerprint(
568 query,
569 effective_nprobe,
570 routing,
571 );
572 if let Some(cache) = cache {
573 let mut cached = cache
574 .lock()
575 .map_err(|_| Error::Internal("binary IVF probe cache is poisoned".into()))?;
576 if let Some(plan) = cached.as_ref()
577 && plan.quantizer_version == quantizer.version
578 && plan.request_fingerprint == request_fingerprint
579 && plan.cluster_ids.len() == effective_nprobe
580 {
581 return Ok(std::sync::Arc::clone(&plan.cluster_ids));
582 }
583 let plan = quantizer.probe(query, effective_nprobe, routing);
584 let clusters = std::sync::Arc::clone(&plan.cluster_ids);
585 *cached = Some(plan);
586 return Ok(clusters);
587 }
588 Ok(quantizer
589 .probe(query, effective_nprobe, routing)
590 .cluster_ids)
591}
592
593pub struct SegmentReader {
599 meta: SegmentMeta,
600 term_dict: Arc<AsyncSSTableReader<TermInfo>>,
602 postings_handle: FileHandle,
604 store: Arc<AsyncStoreReader>,
606 schema: Arc<Schema>,
607 vector_indexes: FxHashMap<u32, VectorIndex>,
609 flat_vectors: FxHashMap<u32, LazyFlatVectorData>,
611 trained_vectors: Arc<crate::segment::TrainedVectorStructures>,
613 sparse_indexes: FxHashMap<u32, SparseIndex>,
615 bmp_indexes: FxHashMap<u32, BmpIndex>,
617 positions_handle: Option<FileHandle>,
619 fast_fields: FxHashMap<u32, crate::structures::fast_field::FastFieldReader>,
621 #[cfg(feature = "native")]
626 pin_report: crate::segment::pin::PinReport,
627}
628
629impl SegmentReader {
630 pub async fn open<D: Directory>(
632 dir: &D,
633 segment_id: SegmentId,
634 schema: Arc<Schema>,
635 cache_blocks: usize,
636 ) -> Result<Self> {
637 Self::open_with_cache_blocks(dir, segment_id, schema, cache_blocks, cache_blocks).await
638 }
639
640 pub async fn open_with_cache_blocks<D: Directory>(
646 dir: &D,
647 segment_id: SegmentId,
648 schema: Arc<Schema>,
649 term_cache_blocks: usize,
650 store_cache_blocks: usize,
651 ) -> Result<Self> {
652 let files = SegmentFiles::new(segment_id.0);
653
654 let meta_slice = dir.open_read(&files.meta).await?;
656 let meta_bytes = meta_slice.read_bytes().await?;
657 let meta = SegmentMeta::deserialize(meta_bytes.as_slice())?;
658 debug_assert_eq!(meta.id, segment_id.0);
659
660 let term_dict_handle = dir.open_lazy(&files.term_dict).await?;
662 let term_dict = AsyncSSTableReader::open(term_dict_handle, term_cache_blocks).await?;
663
664 let postings_handle = dir.open_lazy(&files.postings).await?;
666
667 let store_handle = dir.open_lazy(&files.store).await?;
669 let store = AsyncStoreReader::open(store_handle, store_cache_blocks).await?;
670
671 let vectors_data = loader::load_vectors_file(dir, &files, &schema, meta.num_docs).await?;
673 let vector_indexes = vectors_data.indexes;
674 let flat_vectors = vectors_data.flat_vectors;
675
676 #[cfg(feature = "native")]
681 for (field_id, lazy_flat) in &flat_vectors {
682 if vector_indexes.contains_key(field_id) {
683 lazy_flat.advise_random_access();
684 }
685 }
686
687 let sparse_data = loader::load_sparse_file(dir, &files, meta.num_docs, &schema).await?;
689 let sparse_indexes = sparse_data.maxscore_indexes;
690 let bmp_indexes = sparse_data.bmp_indexes;
691
692 let positions_handle = loader::open_positions_file(dir, &files, &schema).await?;
694
695 let fast_fields = loader::load_fast_fields_file(dir, &files, &schema).await?;
697
698 {
700 let mut parts = vec![format!(
701 "[segment] loaded {:016x}: docs={}",
702 segment_id.0, meta.num_docs
703 )];
704 if !vector_indexes.is_empty() || !flat_vectors.is_empty() {
705 parts.push(format!(
706 "dense: {} ann + {} flat fields",
707 vector_indexes.len(),
708 flat_vectors.len()
709 ));
710 }
711 for (field_id, idx) in &sparse_indexes {
712 parts.push(format!(
713 "sparse field {}: {} dims, ~{:.1} KB",
714 field_id,
715 idx.num_dimensions(),
716 idx.num_dimensions() as f64 * 24.0 / 1024.0
717 ));
718 }
719 for (field_id, idx) in &bmp_indexes {
720 parts.push(format!(
721 "bmp field {}: {} dims, {} blocks",
722 field_id,
723 idx.dims(),
724 idx.num_blocks
725 ));
726 }
727 if !fast_fields.is_empty() {
728 parts.push(format!("fast: {} fields", fast_fields.len()));
729 }
730 log::debug!("{}", parts.join(", "));
731 }
732
733 #[allow(unused_mut)]
734 let mut reader = Self {
735 meta,
736 term_dict: Arc::new(term_dict),
737 postings_handle,
738 store: Arc::new(store),
739 schema,
740 vector_indexes,
741 flat_vectors,
742 trained_vectors: Arc::new(crate::segment::TrainedVectorStructures::default()),
743 sparse_indexes,
744 bmp_indexes,
745 positions_handle,
746 fast_fields,
747 #[cfg(feature = "native")]
748 pin_report: Default::default(),
749 };
750
751 #[cfg(feature = "native")]
753 reader.apply_pin_policy(&crate::segment::pin::pin_policy().to_owned());
754
755 Ok(reader)
756 }
757
758 #[cfg(feature = "native")]
767 pub(crate) fn apply_pin_policy(&mut self, policy: &crate::segment::pin::PinPolicy) {
768 use crate::segment::pin::PinReport;
769
770 if !policy.is_enabled() {
771 return;
772 }
773 let mut remaining = policy.budget_bytes;
774 let mut report = PinReport::default();
775
776 for bmp in self.bmp_indexes.values_mut() {
778 bmp.pin_block_starts(policy.mode, &mut remaining, &mut report);
779 }
780 for sparse in self.sparse_indexes.values_mut() {
782 sparse.pin_skip_section(policy.mode, &mut remaining, &mut report);
783 }
784 for flat in self.flat_vectors.values_mut() {
786 flat.pin_doc_ids(policy.mode, &mut remaining, &mut report);
787 }
788 for bmp in self.bmp_indexes.values_mut() {
789 bmp.pin_doc_maps(policy.mode, &mut remaining, &mut report);
790 }
791 for bmp in self.bmp_indexes.values_mut() {
793 bmp.pin_sb_grid(policy.mode, &mut remaining, &mut report);
794 }
795
796 if report.skipped_budget_bytes > 0 || report.failed_bytes > 0 {
797 log::warn!(
798 "[pin] segment {:016x}: pinned {}/{} bytes (budget skipped {}, mlock failed {}) — raise HERMES_PIN_METADATA_BUDGET_MB or RLIMIT_MEMLOCK for full coverage",
799 self.meta.id,
800 report.pinned_bytes,
801 report.intended_bytes,
802 report.skipped_budget_bytes,
803 report.failed_bytes,
804 );
805 } else if report.pinned_bytes > 0 {
806 log::info!(
807 "[pin] segment {:016x}: pinned {} bytes of hot metadata ({:?})",
808 self.meta.id,
809 report.pinned_bytes,
810 policy.mode,
811 );
812 }
813 self.pin_report = report;
814 }
815
816 pub fn meta(&self) -> &SegmentMeta {
822 &self.meta
823 }
824
825 pub fn num_docs(&self) -> u32 {
826 self.meta.num_docs
827 }
828
829 pub fn avg_field_len(&self, field: Field) -> f32 {
831 self.meta.avg_field_len(field)
832 }
833
834 pub fn schema(&self) -> &Schema {
835 &self.schema
836 }
837
838 pub fn sparse_indexes(&self) -> &FxHashMap<u32, SparseIndex> {
840 &self.sparse_indexes
841 }
842
843 pub fn sparse_index(&self, field: Field) -> Option<&SparseIndex> {
845 self.sparse_indexes.get(&field.0)
846 }
847
848 pub fn bmp_index(&self, field: Field) -> Option<&BmpIndex> {
850 self.bmp_indexes.get(&field.0)
851 }
852
853 pub fn bmp_indexes(&self) -> &FxHashMap<u32, BmpIndex> {
855 &self.bmp_indexes
856 }
857
858 pub fn vector_indexes(&self) -> &FxHashMap<u32, VectorIndex> {
860 &self.vector_indexes
861 }
862
863 pub fn flat_vectors(&self) -> &FxHashMap<u32, LazyFlatVectorData> {
865 &self.flat_vectors
866 }
867
868 pub fn fast_field(
870 &self,
871 field_id: u32,
872 ) -> Option<&crate::structures::fast_field::FastFieldReader> {
873 self.fast_fields.get(&field_id)
874 }
875
876 pub fn fast_fields(&self) -> &FxHashMap<u32, crate::structures::fast_field::FastFieldReader> {
878 &self.fast_fields
879 }
880
881 pub fn term_dict_stats(&self) -> SSTableStats {
883 self.term_dict.stats()
884 }
885
886 pub fn memory_stats(&self) -> SegmentMemoryStats {
888 let term_dict_stats = self.term_dict.stats();
889
890 let term_dict_cache_bytes = self.term_dict.cached_bytes();
894 let store_cache_bytes = self.store.cached_bytes();
895
896 let sparse_index_bytes: usize = self
898 .sparse_indexes
899 .values()
900 .map(|s| s.estimated_memory_bytes())
901 .sum::<usize>()
902 + self
903 .bmp_indexes
904 .values()
905 .map(|b| b.estimated_memory_bytes())
906 .sum::<usize>();
907
908 let dense_index_bytes: usize = self
911 .vector_indexes
912 .values()
913 .map(|v| v.estimated_memory_bytes())
914 .sum();
915
916 #[cfg(feature = "native")]
917 let (pinned_metadata_bytes, pin_intended_bytes) =
918 (self.pin_report.pinned_bytes, self.pin_report.intended_bytes);
919 #[cfg(not(feature = "native"))]
920 let (pinned_metadata_bytes, pin_intended_bytes) = (0u64, 0u64);
921
922 SegmentMemoryStats {
923 segment_id: self.meta.id,
924 num_docs: self.meta.num_docs,
925 term_dict_cache_bytes,
926 store_cache_bytes,
927 sparse_index_bytes,
928 dense_index_bytes,
929 bloom_filter_bytes: term_dict_stats.bloom_filter_size,
930 pinned_metadata_bytes,
931 pin_intended_bytes,
932 }
933 }
934
935 pub async fn get_postings(
940 &self,
941 field: Field,
942 term: &[u8],
943 ) -> Result<Option<BlockPostingList>> {
944 log::debug!(
945 "SegmentReader::get_postings field={} term_len={}",
946 field.0,
947 term.len()
948 );
949
950 let mut key = Vec::with_capacity(4 + term.len());
952 key.extend_from_slice(&field.0.to_le_bytes());
953 key.extend_from_slice(term);
954
955 let term_info = match self.term_dict.get(&key).await? {
957 Some(info) => {
958 log::debug!("SegmentReader::get_postings found term_info");
959 info
960 }
961 None => {
962 log::debug!("SegmentReader::get_postings term not found");
963 return Ok(None);
964 }
965 };
966
967 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
969 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
971 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
972 posting_list.push(doc_id, tf);
973 }
974 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
975 return Ok(Some(block_list));
976 }
977
978 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
980 Error::Corruption("TermInfo has neither inline nor external data".to_string())
981 })?;
982
983 let range = checked_file_range(
984 posting_offset,
985 posting_len,
986 self.postings_handle.len(),
987 "posting",
988 )?;
989 let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
990 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
991
992 Ok(Some(block_list))
993 }
994
995 pub async fn get_prefix_postings(
997 &self,
998 field: Field,
999 prefix: &[u8],
1000 ) -> Result<Vec<BlockPostingList>> {
1001 if prefix.is_empty() {
1002 return Err(Error::Query("prefix must not be empty".into()));
1003 }
1004 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
1006 key_prefix.extend_from_slice(&field.0.to_le_bytes());
1007 key_prefix.extend_from_slice(prefix);
1008
1009 let (entries, truncated) = self
1010 .term_dict
1011 .prefix_scan_limited(&key_prefix, MAX_PREFIX_TERMS)
1012 .await?;
1013 if truncated {
1014 return Err(Error::Query(format!(
1015 "prefix expands to more than {MAX_PREFIX_TERMS} terms"
1016 )));
1017 }
1018 let posting_count: u64 = entries
1019 .iter()
1020 .map(|(_, term_info)| term_info.doc_freq() as u64)
1021 .sum();
1022 if posting_count > MAX_PREFIX_POSTINGS {
1023 return Err(Error::Query(format!(
1024 "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
1025 )));
1026 }
1027 let mut results = Vec::with_capacity(entries.len());
1028
1029 for (_key, term_info) in entries {
1030 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
1031 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
1032 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
1033 posting_list.push(doc_id, tf);
1034 }
1035 results.push(BlockPostingList::from_posting_list(&posting_list)?);
1036 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
1037 let range = checked_file_range(
1038 posting_offset,
1039 posting_len,
1040 self.postings_handle.len(),
1041 "prefix posting",
1042 )?;
1043 let posting_bytes = self.postings_handle.read_bytes_range(range).await?;
1044 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
1045 }
1046 }
1047
1048 Ok(results)
1049 }
1050
1051 pub async fn doc(&self, local_doc_id: DocId) -> Result<Option<Document>> {
1056 self.doc_with_fields(local_doc_id, None).await
1057 }
1058
1059 pub async fn doc_with_fields(
1065 &self,
1066 local_doc_id: DocId,
1067 fields: Option<&rustc_hash::FxHashSet<u32>>,
1068 ) -> Result<Option<Document>> {
1069 let mut doc = match fields {
1070 Some(set) => {
1071 let field_ids: Vec<u32> = set.iter().copied().collect();
1072 match self
1073 .store
1074 .get_fields(local_doc_id, &self.schema, &field_ids)
1075 .await
1076 {
1077 Ok(Some(d)) => d,
1078 Ok(None) => return Ok(None),
1079 Err(e) => return Err(Error::from(e)),
1080 }
1081 }
1082 None => match self.store.get(local_doc_id, &self.schema).await {
1083 Ok(Some(d)) => d,
1084 Ok(None) => return Ok(None),
1085 Err(e) => return Err(Error::from(e)),
1086 },
1087 };
1088
1089 for (&field_id, lazy_flat) in &self.flat_vectors {
1091 if let Some(set) = fields
1093 && !set.contains(&field_id)
1094 {
1095 continue;
1096 }
1097
1098 let is_binary = lazy_flat.quantization == DenseVectorQuantization::Binary;
1099 let (start, entries) = lazy_flat.flat_indexes_for_doc(local_doc_id);
1100 for (j, &(_doc_id, _ordinal)) in entries.iter().enumerate() {
1101 let flat_idx = start + j;
1102 if is_binary {
1103 let vbs = lazy_flat.vector_byte_size();
1104 let mut raw = vec![0u8; vbs];
1105 match lazy_flat.read_vector_raw_into(flat_idx, &mut raw).await {
1106 Ok(()) => {
1107 doc.add_binary_dense_vector(Field(field_id), raw);
1108 }
1109 Err(e) => {
1110 log::warn!("Failed to hydrate binary vector field {}: {}", field_id, e);
1111 }
1112 }
1113 } else {
1114 match lazy_flat.get_vector(flat_idx).await {
1115 Ok(vec) => {
1116 doc.add_dense_vector(Field(field_id), vec);
1117 }
1118 Err(e) => {
1119 log::warn!("Failed to hydrate vector field {}: {}", field_id, e);
1120 }
1121 }
1122 }
1123 }
1124 }
1125
1126 Ok(Some(doc))
1127 }
1128
1129 pub async fn prefetch_terms(
1131 &self,
1132 field: Field,
1133 start_term: &[u8],
1134 end_term: &[u8],
1135 ) -> Result<()> {
1136 let mut start_key = Vec::with_capacity(4 + start_term.len());
1137 start_key.extend_from_slice(&field.0.to_le_bytes());
1138 start_key.extend_from_slice(start_term);
1139
1140 let mut end_key = Vec::with_capacity(4 + end_term.len());
1141 end_key.extend_from_slice(&field.0.to_le_bytes());
1142 end_key.extend_from_slice(end_term);
1143
1144 self.term_dict.prefetch_range(&start_key, &end_key).await?;
1145 Ok(())
1146 }
1147
1148 pub fn store_has_dict(&self) -> bool {
1150 self.store.has_dict()
1151 }
1152
1153 pub fn store(&self) -> &super::store::AsyncStoreReader {
1155 &self.store
1156 }
1157
1158 pub fn store_raw_blocks(&self) -> Vec<RawStoreBlock> {
1160 self.store.raw_blocks()
1161 }
1162
1163 pub fn store_data_slice(&self) -> &FileHandle {
1165 self.store.data_slice()
1166 }
1167
1168 pub async fn all_terms(&self) -> Result<Vec<(Vec<u8>, TermInfo)>> {
1170 self.term_dict.all_entries().await.map_err(Error::from)
1171 }
1172
1173 pub async fn all_terms_with_stats(&self) -> Result<Vec<(Field, String, u32)>> {
1178 let entries = self.term_dict.all_entries().await?;
1179 let mut result = Vec::with_capacity(entries.len());
1180
1181 for (key, term_info) in entries {
1182 if key.len() > 4 {
1184 let field_id = u32::from_le_bytes([key[0], key[1], key[2], key[3]]);
1185 let term_bytes = &key[4..];
1186 if let Ok(term_str) = std::str::from_utf8(term_bytes) {
1187 result.push((Field(field_id), term_str.to_string(), term_info.doc_freq()));
1188 }
1189 }
1190 }
1191
1192 Ok(result)
1193 }
1194
1195 pub fn term_dict_iter(&self) -> crate::structures::AsyncSSTableIterator<'_, TermInfo> {
1197 self.term_dict.iter()
1198 }
1199
1200 pub async fn prefetch_term_dict(&self) -> crate::Result<()> {
1204 self.term_dict
1205 .prefetch_all_data_bulk()
1206 .await
1207 .map_err(crate::Error::from)
1208 }
1209
1210 pub async fn read_postings(&self, offset: u64, len: u64) -> Result<Vec<u8>> {
1212 let range = checked_file_range(offset, len, self.postings_handle.len(), "posting")?;
1213 let bytes = self.postings_handle.read_bytes_range(range).await?;
1214 Ok(bytes.to_vec())
1215 }
1216
1217 pub async fn read_position_bytes(&self, offset: u64, len: u64) -> Result<Option<Vec<u8>>> {
1219 let handle = match &self.positions_handle {
1220 Some(h) => h,
1221 None => return Ok(None),
1222 };
1223 let range = checked_file_range(offset, len, handle.len(), "position")?;
1224 let bytes = handle.read_bytes_range(range).await?;
1225 Ok(Some(bytes.to_vec()))
1226 }
1227
1228 pub fn has_positions_file(&self) -> bool {
1230 self.positions_handle.is_some()
1231 }
1232
1233 fn validate_dense_search_request(
1237 &self,
1238 field: Field,
1239 query: &[f32],
1240 nprobe: usize,
1241 rerank_factor: f32,
1242 combiner: crate::query::MultiValueCombiner,
1243 ) -> Result<DenseSearchParams> {
1244 let entry = self
1245 .schema
1246 .get_field_entry(field)
1247 .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
1248 if entry.field_type != crate::dsl::FieldType::DenseVector {
1249 return Err(Error::InvalidFieldType {
1250 expected: "dense_vector".to_string(),
1251 got: format!("{:?}", entry.field_type),
1252 });
1253 }
1254 let config = entry.dense_vector_config.as_ref().ok_or_else(|| {
1255 Error::Schema(format!(
1256 "dense vector field '{}' has no dense vector configuration",
1257 entry.name
1258 ))
1259 })?;
1260
1261 if query.is_empty() {
1262 return Err(Error::Query(format!(
1263 "dense query vector for field '{}' must not be empty",
1264 entry.name
1265 )));
1266 }
1267 if query.len() != config.dim {
1268 return Err(Error::Query(format!(
1269 "dense query vector dimension {} does not match field '{}' dimension {}",
1270 query.len(),
1271 entry.name,
1272 config.dim
1273 )));
1274 }
1275 if let Some((index, value)) = query
1276 .iter()
1277 .enumerate()
1278 .find(|(_, value)| !value.is_finite())
1279 {
1280 return Err(Error::Query(format!(
1281 "dense query vector for field '{}' contains non-finite value {value} at index {index}",
1282 entry.name
1283 )));
1284 }
1285
1286 let nprobe = match (nprobe, config.nprobe) {
1289 (0, 0) => 32,
1290 (0, schema_nprobe) => schema_nprobe,
1291 (query_nprobe, _) => query_nprobe,
1292 };
1293 if nprobe > MAX_DENSE_NPROBE {
1294 return Err(Error::Query(format!(
1295 "dense nprobe must be at most {MAX_DENSE_NPROBE}, got {nprobe}"
1296 )));
1297 }
1298
1299 checked_dense_fetch_k(0, rerank_factor)?;
1302 combiner.validate().map_err(Error::Query)?;
1303
1304 Ok(DenseSearchParams {
1305 dim: config.dim,
1306 nprobe,
1307 unit_norm: config.unit_norm,
1308 })
1309 }
1310
1311 fn validate_binary_search_request(&self, field: Field, query: &[u8]) -> Result<usize> {
1312 let entry = self
1313 .schema
1314 .get_field_entry(field)
1315 .ok_or_else(|| Error::FieldNotFound(field.0.to_string()))?;
1316 if entry.field_type != crate::dsl::FieldType::BinaryDenseVector {
1317 return Err(Error::InvalidFieldType {
1318 expected: "binary_dense_vector".to_string(),
1319 got: format!("{:?}", entry.field_type),
1320 });
1321 }
1322 let config = entry.binary_dense_vector_config.as_ref().ok_or_else(|| {
1323 Error::Schema(format!(
1324 "binary dense vector field '{}' has no configuration",
1325 entry.name
1326 ))
1327 })?;
1328 if config.dim == 0 || !config.dim.is_multiple_of(8) {
1329 return Err(Error::Schema(format!(
1330 "binary dense vector field '{}' has invalid dimension {}",
1331 entry.name, config.dim
1332 )));
1333 }
1334 if query.len() != config.byte_len() {
1335 return Err(Error::Query(format!(
1336 "binary query byte length {} does not match field '{}' byte length {}",
1337 query.len(),
1338 entry.name,
1339 config.byte_len()
1340 )));
1341 }
1342 Ok(config.dim)
1343 }
1344
1345 fn score_quantized_batch(
1351 query: &[f32],
1352 raw: &[u8],
1353 quant: crate::dsl::DenseVectorQuantization,
1354 dim: usize,
1355 scores: &mut [f32],
1356 unit_norm: bool,
1357 ) -> Result<()> {
1358 use crate::dsl::DenseVectorQuantization;
1359 use crate::structures::simd;
1360
1361 if query.len() != dim {
1362 return Err(Error::Query(format!(
1363 "dense SIMD query dimension {} does not match vector dimension {dim}",
1364 query.len()
1365 )));
1366 }
1367 let element_size = match quant {
1368 DenseVectorQuantization::F32 => std::mem::size_of::<f32>(),
1369 DenseVectorQuantization::F16 => std::mem::size_of::<u16>(),
1370 DenseVectorQuantization::UInt8 => 1,
1371 DenseVectorQuantization::Binary => {
1372 return Err(Error::InvalidFieldType {
1373 expected: "non-binary dense vector".to_string(),
1374 got: "binary dense vector".to_string(),
1375 });
1376 }
1377 };
1378 let required_bytes = scores
1379 .len()
1380 .checked_mul(dim)
1381 .and_then(|elements| elements.checked_mul(element_size))
1382 .ok_or_else(|| Error::Corruption("dense vector batch byte length overflow".into()))?;
1383 if raw.len() < required_bytes {
1384 return Err(Error::Corruption(format!(
1385 "dense vector batch is truncated: need {required_bytes} bytes, got {}",
1386 raw.len()
1387 )));
1388 }
1389 if quant == DenseVectorQuantization::F16
1390 && required_bytes > 0
1391 && !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<u16>())
1392 {
1393 return Err(Error::Corruption(
1394 "f16 vector data is not 2-byte aligned".to_string(),
1395 ));
1396 }
1397
1398 match (quant, unit_norm) {
1399 (DenseVectorQuantization::F32, false) => {
1400 let num_floats = scores.len() * dim;
1401 if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
1402 return Err(Error::Corruption(
1403 "f32 vector data is not 4-byte aligned".to_string(),
1404 ));
1405 }
1406 let vectors: &[f32] =
1407 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
1408 simd::batch_cosine_scores(query, vectors, dim, scores);
1409 }
1410 (DenseVectorQuantization::F32, true) => {
1411 let num_floats = scores.len() * dim;
1412 if !(raw.as_ptr() as usize).is_multiple_of(std::mem::align_of::<f32>()) {
1413 return Err(Error::Corruption(
1414 "f32 vector data is not 4-byte aligned".to_string(),
1415 ));
1416 }
1417 let vectors: &[f32] =
1418 unsafe { std::slice::from_raw_parts(raw.as_ptr() as *const f32, num_floats) };
1419 simd::batch_dot_scores(query, vectors, dim, scores);
1420 }
1421 (DenseVectorQuantization::F16, false) => {
1422 simd::batch_cosine_scores_f16(query, raw, dim, scores);
1423 }
1424 (DenseVectorQuantization::F16, true) => {
1425 simd::batch_dot_scores_f16(query, raw, dim, scores);
1426 }
1427 (DenseVectorQuantization::UInt8, false) => {
1428 simd::batch_cosine_scores_u8(query, raw, dim, scores);
1429 }
1430 (DenseVectorQuantization::UInt8, true) => {
1431 simd::batch_dot_scores_u8(query, raw, dim, scores);
1432 }
1433 (DenseVectorQuantization::Binary, _) => unreachable!("validated above"),
1434 }
1435 Ok(())
1436 }
1437
1438 pub async fn search_dense_vector(
1444 &self,
1445 field: Field,
1446 query: &[f32],
1447 k: usize,
1448 nprobe: usize,
1449 rerank_factor: f32,
1450 combiner: crate::query::MultiValueCombiner,
1451 ) -> Result<Vec<VectorSearchResult>> {
1452 self.search_dense_vector_impl(field, query, k, nprobe, rerank_factor, combiner, None)
1453 .await
1454 }
1455
1456 #[allow(clippy::too_many_arguments)]
1457 pub(crate) async fn search_dense_vector_with_probe_cache(
1458 &self,
1459 field: Field,
1460 query: &[f32],
1461 k: usize,
1462 nprobe: usize,
1463 rerank_factor: f32,
1464 combiner: crate::query::MultiValueCombiner,
1465 probe_cache: &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
1466 ) -> Result<Vec<VectorSearchResult>> {
1467 self.search_dense_vector_impl(
1468 field,
1469 query,
1470 k,
1471 nprobe,
1472 rerank_factor,
1473 combiner,
1474 Some(probe_cache),
1475 )
1476 .await
1477 }
1478
1479 #[allow(clippy::too_many_arguments)]
1480 async fn search_dense_vector_impl(
1481 &self,
1482 field: Field,
1483 query: &[f32],
1484 k: usize,
1485 nprobe: usize,
1486 rerank_factor: f32,
1487 combiner: crate::query::MultiValueCombiner,
1488 probe_cache: Option<
1489 &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
1490 >,
1491 ) -> Result<Vec<VectorSearchResult>> {
1492 let params =
1493 self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
1494 let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
1495 if k == 0 {
1496 return Ok(Vec::new());
1497 }
1498
1499 let ann_index = self.vector_indexes.get(&field.0);
1500 let lazy_flat = self.flat_vectors.get(&field.0);
1501 if ann_index.is_none() && lazy_flat.is_none() {
1503 return Ok(Vec::new());
1504 }
1505
1506 if ann_index.is_some() && lazy_flat.is_none() {
1507 return Err(Error::Corruption(format!(
1508 "dense ANN field {} is missing flat vector storage",
1509 field.0
1510 )));
1511 }
1512
1513 if let Some(flat) = lazy_flat
1514 && flat.dim != params.dim
1515 {
1516 return Err(Error::Corruption(format!(
1517 "dense vector field {} has schema dimension {} but flat storage dimension {}",
1518 field.0, params.dim, flat.dim
1519 )));
1520 }
1521
1522 let t0 = std::time::Instant::now();
1524 let mut flat_results = None;
1525 let mut results: Vec<(u32, u16, f32)> = if let Some(index) = ann_index {
1526 match index {
1528 VectorIndex::IvfPq(lazy) => {
1529 let index = lazy.get().ok_or_else(|| {
1530 Error::Schema("IVF-PQ index deserialization failed".to_string())
1531 })?;
1532 let codebook =
1533 self.trained_vectors
1534 .codebooks
1535 .get(&field.0)
1536 .ok_or_else(|| {
1537 Error::Schema(format!(
1538 "IVF-PQ index requires a global codebook for field {}",
1539 field.0
1540 ))
1541 })?;
1542 let centroids =
1543 self.trained_vectors
1544 .centroids
1545 .get(&field.0)
1546 .ok_or_else(|| {
1547 Error::Schema(format!(
1548 "IVF-PQ index requires coarse centroids for field {}",
1549 field.0
1550 ))
1551 })?;
1552 validate_coarse_centroids(centroids, params.dim)?;
1553 if index.config.dim != params.dim
1554 || codebook.config.dim != params.dim
1555 || index.config.code_size != codebook.config.num_subspaces
1556 || index.centroids_version != centroids.version
1557 || index.codebook_version != codebook.version
1558 || index
1559 .clusters
1560 .iter()
1561 .any(|(cluster_id, _)| *cluster_id >= centroids.num_clusters)
1562 {
1563 return Err(Error::Corruption(format!(
1564 "IVF-PQ index/codebook/centroid metadata does not match schema dimension {}",
1565 params.dim
1566 )));
1567 }
1568 let routing = self
1569 .schema
1570 .get_field_entry(field)
1571 .and_then(|entry| entry.dense_vector_config.as_ref())
1572 .map_or(crate::dsl::IvfRoutingMode::Auto, |config| {
1573 config.ivf_routing
1574 });
1575 let query_plan = float_query_plan(
1576 centroids,
1577 codebook,
1578 query,
1579 params.nprobe,
1580 routing,
1581 probe_cache,
1582 )?;
1583 let flat = lazy_flat.expect("ANN/flat pairing validated above");
1584 index
1585 .search_distinct_documents(
1586 fetch_k.min(flat.num_docs_with_vectors()),
1587 &query_plan,
1588 )
1589 .into_iter()
1590 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
1591 .collect()
1592 }
1593 VectorIndex::BinaryIvf(_) => {
1594 Vec::new()
1596 }
1597 }
1598 } else if let Some(lazy_flat) = lazy_flat {
1599 log::debug!(
1603 "[search_dense] field {}: brute-force on {} vectors (dim={}, quant={:?})",
1604 field.0,
1605 lazy_flat.num_vectors,
1606 lazy_flat.dim,
1607 lazy_flat.quantization
1608 );
1609 let dim = lazy_flat.dim;
1610 let n = lazy_flat.num_vectors;
1611 let quant = lazy_flat.quantization;
1612 let batch_len =
1613 bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
1614 let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
1615 let mut scores = vec![0f32; batch_len];
1616
1617 for batch_start in (0..n).step_by(batch_len) {
1618 let batch_count = batch_len.min(n - batch_start);
1619 let batch_bytes = lazy_flat
1620 .read_vectors_batch(batch_start, batch_count)
1621 .await
1622 .map_err(crate::Error::Io)?;
1623 let raw = batch_bytes.as_slice();
1624
1625 Self::score_quantized_batch(
1626 query,
1627 raw,
1628 quant,
1629 dim,
1630 &mut scores[..batch_count],
1631 params.unit_norm,
1632 )?;
1633
1634 for (i, &score) in scores.iter().enumerate().take(batch_count) {
1635 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1636 collector.push(doc_id, ordinal, score);
1637 }
1638 }
1639
1640 flat_results = Some(collector.into_results());
1641 Vec::new()
1642 } else {
1643 return Ok(Vec::new());
1644 };
1645 let l1_elapsed = t0.elapsed();
1646 {
1647 let kind = match ann_index {
1648 Some(VectorIndex::IvfPq(_)) => "ivf_pq",
1649 Some(VectorIndex::BinaryIvf(_)) => "binary_ivf",
1650 None => "flat",
1651 };
1652 crate::observe::dense_l1(
1653 self.schema.index_label(),
1654 self.schema.get_field_name(field).unwrap_or("?"),
1655 kind,
1656 l1_elapsed.as_secs_f64(),
1657 flat_results.as_ref().map_or(results.len(), Vec::len),
1658 );
1659 }
1660 log::debug!(
1661 "[search_dense] field {}: L1 returned {} candidates in {:.1}ms",
1662 field.0,
1663 flat_results.as_ref().map_or(results.len(), Vec::len),
1664 l1_elapsed.as_secs_f64() * 1000.0
1665 );
1666
1667 if let Some(results) = flat_results {
1668 return Ok(results);
1669 }
1670
1671 if ann_index.is_some()
1674 && !results.is_empty()
1675 && let Some(lazy_flat) = lazy_flat
1676 {
1677 let t_rerank = std::time::Instant::now();
1678 let dim = lazy_flat.dim;
1679 let quant = lazy_flat.quantization;
1680 let vbs = lazy_flat.vector_byte_size();
1681
1682 let (expanded, mut resolved) = expand_ann_candidate_documents(&results, lazy_flat)?;
1686 results = expanded;
1687
1688 let t_resolve = t_rerank.elapsed();
1689 if !resolved.is_empty() {
1690 resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
1692
1693 let batch_len = bounded_vector_score_batch(vbs, DENSE_SCORE_BATCH);
1697 let max_batch = batch_len.min(resolved.len());
1698 let max_raw_len = max_batch
1699 .checked_mul(vbs)
1700 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
1701 let mut raw_buf = vec![0u8; max_raw_len];
1702 let mut scores = vec![0f32; max_batch];
1703 let mut read_elapsed = std::time::Duration::ZERO;
1704 let mut score_elapsed = std::time::Duration::ZERO;
1705
1706 for chunk in resolved.chunks(batch_len) {
1707 #[cfg(feature = "native")]
1711 lazy_flat.prefetch_vectors(chunk.iter().map(|&(_, flat_idx)| flat_idx));
1712 let raw_len = chunk
1713 .len()
1714 .checked_mul(vbs)
1715 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
1716 let raw = &mut raw_buf[..raw_len];
1717
1718 let t_read = std::time::Instant::now();
1719 for (buf_idx, &(_, flat_idx)) in chunk.iter().enumerate() {
1720 lazy_flat
1721 .read_vector_raw_into(
1722 flat_idx,
1723 &mut raw[buf_idx * vbs..(buf_idx + 1) * vbs],
1724 )
1725 .await
1726 .map_err(crate::Error::Io)?;
1727 }
1728 read_elapsed += t_read.elapsed();
1729
1730 let t_score = std::time::Instant::now();
1731 Self::score_quantized_batch(
1732 query,
1733 raw,
1734 quant,
1735 dim,
1736 &mut scores[..chunk.len()],
1737 params.unit_norm,
1738 )?;
1739 score_elapsed += t_score.elapsed();
1740
1741 for (buf_idx, &(ri, _)) in chunk.iter().enumerate() {
1742 results[ri].2 = scores[buf_idx];
1743 }
1744 }
1745
1746 crate::observe::dense_rerank(
1747 self.schema.index_label(),
1748 self.schema.get_field_name(field).unwrap_or("?"),
1749 t_rerank.elapsed().as_secs_f64(),
1750 t_resolve.as_secs_f64(),
1751 read_elapsed.as_secs_f64(),
1752 resolved.len(),
1753 );
1754 log::debug!(
1755 "[search_dense] field {}: rerank {} vectors (dim={}, quant={:?}, {}B/vec): resolve={:.1}ms read={:.1}ms score={:.1}ms",
1756 field.0,
1757 resolved.len(),
1758 dim,
1759 quant,
1760 vbs,
1761 t_resolve.as_secs_f64() * 1000.0,
1762 read_elapsed.as_secs_f64() * 1000.0,
1763 score_elapsed.as_secs_f64() * 1000.0,
1764 );
1765 }
1766
1767 log::debug!(
1768 "[search_dense] field {}: rerank total={:.1}ms",
1769 field.0,
1770 t_rerank.elapsed().as_secs_f64() * 1000.0
1771 );
1772 }
1773
1774 Ok(combine_grouped_ordinal_results(results, combiner, k))
1775 }
1776
1777 async fn search_binary_dense_vector_impl(
1782 &self,
1783 field: Field,
1784 query: &[u8],
1785 k: usize,
1786 combiner: crate::query::MultiValueCombiner,
1787 probe_cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
1788 ) -> Result<Vec<VectorSearchResult>> {
1789 let schema_dim = self.validate_binary_search_request(field, query)?;
1790 combiner.validate().map_err(Error::Query)?;
1791 if k == 0 {
1792 return Ok(Vec::new());
1793 }
1794 let t0 = crate::observe::Timer::start();
1795 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0) {
1796 let ivf = lazy.get().ok_or_else(|| {
1797 Error::Corruption(format!(
1798 "global binary IVF field {} could not be decoded",
1799 field.0
1800 ))
1801 })?;
1802 let config = self
1803 .schema
1804 .get_field_entry(field)
1805 .and_then(|entry| entry.binary_dense_vector_config.as_ref())
1806 .ok_or_else(|| {
1807 Error::Schema(format!(
1808 "binary IVF field {} has no schema configuration",
1809 field.0
1810 ))
1811 })?;
1812 let quantizer = self
1813 .trained_vectors
1814 .binary_quantizers
1815 .get(&field.0)
1816 .ok_or_else(|| {
1817 Error::Schema(format!(
1818 "global binary IVF field {} has no loaded quantizer",
1819 field.0
1820 ))
1821 })?;
1822 if ivf.dim_bits != schema_dim
1823 || quantizer.dim_bits != schema_dim
1824 || ivf.quantizer_version != quantizer.version
1825 || ivf.num_clusters != quantizer.num_clusters
1826 {
1827 return Err(Error::Corruption(format!(
1828 "global binary IVF field {} does not match its quantizer/schema generation",
1829 field.0
1830 )));
1831 }
1832 let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
1833 Error::Corruption(format!(
1834 "global binary IVF field {} is missing flat vector storage",
1835 field.0
1836 ))
1837 })?;
1838 let clusters = binary_probe_clusters(
1839 quantizer,
1840 query,
1841 config.nprobe,
1842 config.ivf_routing,
1843 probe_cache,
1844 )?;
1845 let single_valued = flat.num_vectors == flat.num_docs_with_vectors();
1846 let candidate_docs = k.min(flat.num_docs_with_vectors());
1847 let ann_results = if single_valued {
1848 ivf.search_in_clusters(query, candidate_docs, &clusters)
1849 } else {
1850 ivf.search_distinct_documents_in_clusters(query, candidate_docs, &clusters)
1851 };
1852 let results =
1853 exact_score_binary_candidate_documents(&ann_results, flat, query, schema_dim)
1854 .await?;
1855 crate::observe::dense_l1(
1856 self.schema.index_label(),
1857 self.schema.get_field_name(field).unwrap_or("?"),
1858 "global_binary_ivf",
1859 t0.secs(),
1860 results.len(),
1861 );
1862 return Ok(combine_grouped_ordinal_results(results, combiner, k));
1863 }
1864 let lazy_flat = match self.flat_vectors.get(&field.0) {
1865 Some(f) => f,
1866 None => return Ok(Vec::new()),
1867 };
1868
1869 let dim_bits = lazy_flat.dim;
1870 let byte_len = lazy_flat.vector_byte_size();
1871 let n = lazy_flat.num_vectors;
1872
1873 if dim_bits != schema_dim {
1874 return Err(Error::Corruption(format!(
1875 "binary vector field {} has schema dimension {} but flat storage dimension {}",
1876 field.0, schema_dim, dim_bits
1877 )));
1878 }
1879
1880 if byte_len != query.len() {
1881 return Err(Error::Schema(format!(
1882 "Binary query vector byte length {} != field byte length {}",
1883 query.len(),
1884 byte_len
1885 )));
1886 }
1887
1888 let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
1889 let mut collector = FlatDocumentCollector::new(k, combiner);
1890 let mut scores = vec![0f32; batch_len];
1891
1892 for batch_start in (0..n).step_by(batch_len) {
1893 let batch_count = batch_len.min(n - batch_start);
1894 let batch_bytes = lazy_flat
1895 .read_vectors_batch(batch_start, batch_count)
1896 .await
1897 .map_err(crate::Error::Io)?;
1898 let raw = batch_bytes.as_slice();
1899
1900 crate::structures::simd::batch_hamming_scores(
1901 query,
1902 raw,
1903 byte_len,
1904 dim_bits,
1905 &mut scores[..batch_count],
1906 );
1907
1908 for (i, &score) in scores.iter().enumerate().take(batch_count) {
1909 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
1910 collector.push(doc_id, ordinal, score);
1911 }
1912 }
1913
1914 let results = collector.into_results();
1915
1916 crate::observe::dense_l1(
1917 self.schema.index_label(),
1918 self.schema.get_field_name(field).unwrap_or("?"),
1919 "binary_flat",
1920 t0.secs(),
1921 results.len(),
1922 );
1923 Ok(results)
1924 }
1925
1926 pub async fn search_binary_dense_vector(
1927 &self,
1928 field: Field,
1929 query: &[u8],
1930 k: usize,
1931 combiner: crate::query::MultiValueCombiner,
1932 ) -> Result<Vec<VectorSearchResult>> {
1933 self.search_binary_dense_vector_impl(field, query, k, combiner, None)
1934 .await
1935 }
1936
1937 pub(crate) async fn search_binary_dense_vector_with_probe_cache(
1938 &self,
1939 field: Field,
1940 query: &[u8],
1941 k: usize,
1942 combiner: crate::query::MultiValueCombiner,
1943 probe_cache: &std::sync::Mutex<Option<crate::structures::IvfProbePlan>>,
1944 ) -> Result<Vec<VectorSearchResult>> {
1945 self.search_binary_dense_vector_impl(field, query, k, combiner, Some(probe_cache))
1946 .await
1947 }
1948
1949 pub fn coarse_centroids(&self, field_id: u32) -> Option<&Arc<CoarseCentroids>> {
1951 self.trained_vectors.centroids.get(&field_id)
1952 }
1953
1954 pub fn set_trained_vectors(
1955 &mut self,
1956 trained_vectors: Arc<crate::segment::TrainedVectorStructures>,
1957 ) {
1958 self.trained_vectors = trained_vectors;
1959 }
1960
1961 #[cfg(feature = "native")]
1967 pub(crate) fn warm_ann_indexes(&self) -> Result<()> {
1968 let started = std::time::Instant::now();
1969 for (&field_id, index) in &self.vector_indexes {
1970 let decoded = match index {
1971 VectorIndex::IvfPq(lazy) => lazy.get().is_some(),
1972 VectorIndex::BinaryIvf(lazy) => lazy.get().is_some(),
1973 };
1974 if !decoded {
1975 return Err(Error::Corruption(format!(
1976 "ANN payload for field {field_id} could not be decoded"
1977 )));
1978 }
1979 }
1980 if !self.vector_indexes.is_empty() {
1981 log::info!(
1982 "[vector_warmup] segment {:016x}: decoded {} ANN payloads in {:.1}s",
1983 self.meta.id,
1984 self.vector_indexes.len(),
1985 started.elapsed().as_secs_f64(),
1986 );
1987 }
1988 Ok(())
1989 }
1990
1991 pub fn get_ivf_pq_vector_index(&self, field: Field) -> Option<Arc<IVFPQIndex>> {
1993 match self.vector_indexes.get(&field.0) {
1994 Some(VectorIndex::IvfPq(lazy)) => lazy.get().map(Arc::clone),
1995 _ => None,
1996 }
1997 }
1998
1999 pub fn get_vector_index(&self, field: Field) -> Option<&VectorIndex> {
2001 self.vector_indexes.get(&field.0)
2002 }
2003
2004 pub async fn get_positions(
2009 &self,
2010 field: Field,
2011 term: &[u8],
2012 ) -> Result<Option<crate::structures::PositionPostingList>> {
2013 let handle = match &self.positions_handle {
2015 Some(h) => h,
2016 None => return Ok(None),
2017 };
2018
2019 let mut key = Vec::with_capacity(4 + term.len());
2021 key.extend_from_slice(&field.0.to_le_bytes());
2022 key.extend_from_slice(term);
2023
2024 let term_info = match self.term_dict.get(&key).await? {
2026 Some(info) => info,
2027 None => return Ok(None),
2028 };
2029
2030 let (offset, length) = match term_info.position_info() {
2032 Some((o, l)) => (o, l),
2033 None => return Ok(None),
2034 };
2035
2036 let range = checked_file_range(offset, length, handle.len(), "position list")?;
2040 let slice = handle.slice(range);
2041 let data = slice.read_bytes().await?;
2042
2043 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2045
2046 Ok(Some(pos_list))
2047 }
2048
2049 pub fn has_positions(&self, field: Field) -> bool {
2051 if let Some(entry) = self.schema.get_field_entry(field) {
2053 entry.positions.is_some()
2054 } else {
2055 false
2056 }
2057 }
2058}
2059
2060#[cfg(feature = "sync")]
2062impl SegmentReader {
2063 pub fn get_postings_sync(&self, field: Field, term: &[u8]) -> Result<Option<BlockPostingList>> {
2065 let mut key = Vec::with_capacity(4 + term.len());
2067 key.extend_from_slice(&field.0.to_le_bytes());
2068 key.extend_from_slice(term);
2069
2070 let term_info = match self.term_dict.get_sync(&key)? {
2072 Some(info) => info,
2073 None => return Ok(None),
2074 };
2075
2076 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2078 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2079 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2080 posting_list.push(doc_id, tf);
2081 }
2082 let block_list = BlockPostingList::from_posting_list(&posting_list)?;
2083 return Ok(Some(block_list));
2084 }
2085
2086 let (posting_offset, posting_len) = term_info.external_info().ok_or_else(|| {
2088 Error::Corruption("TermInfo has neither inline nor external data".to_string())
2089 })?;
2090
2091 let range = checked_file_range(
2092 posting_offset,
2093 posting_len,
2094 self.postings_handle.len(),
2095 "posting",
2096 )?;
2097 let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
2098 let block_list = BlockPostingList::deserialize_zero_copy(posting_bytes)?;
2099
2100 Ok(Some(block_list))
2101 }
2102
2103 pub fn get_prefix_postings_sync(
2105 &self,
2106 field: Field,
2107 prefix: &[u8],
2108 ) -> Result<Vec<BlockPostingList>> {
2109 if prefix.is_empty() {
2110 return Err(Error::Query("prefix must not be empty".into()));
2111 }
2112 let mut key_prefix = Vec::with_capacity(4 + prefix.len());
2113 key_prefix.extend_from_slice(&field.0.to_le_bytes());
2114 key_prefix.extend_from_slice(prefix);
2115
2116 let (entries, truncated) = self
2117 .term_dict
2118 .prefix_scan_limited_sync(&key_prefix, MAX_PREFIX_TERMS)?;
2119 if truncated {
2120 return Err(Error::Query(format!(
2121 "prefix expands to more than {MAX_PREFIX_TERMS} terms"
2122 )));
2123 }
2124 let posting_count: u64 = entries
2125 .iter()
2126 .map(|(_, term_info)| term_info.doc_freq() as u64)
2127 .sum();
2128 if posting_count > MAX_PREFIX_POSTINGS {
2129 return Err(Error::Query(format!(
2130 "prefix expands to {posting_count} postings (maximum {MAX_PREFIX_POSTINGS})"
2131 )));
2132 }
2133 let mut results = Vec::with_capacity(entries.len());
2134
2135 for (_key, term_info) in entries {
2136 if let Some((doc_ids, term_freqs)) = term_info.decode_inline() {
2137 let mut posting_list = crate::structures::PostingList::with_capacity(doc_ids.len());
2138 for (doc_id, tf) in doc_ids.into_iter().zip(term_freqs) {
2139 posting_list.push(doc_id, tf);
2140 }
2141 results.push(BlockPostingList::from_posting_list(&posting_list)?);
2142 } else if let Some((posting_offset, posting_len)) = term_info.external_info() {
2143 let range = checked_file_range(
2144 posting_offset,
2145 posting_len,
2146 self.postings_handle.len(),
2147 "prefix posting",
2148 )?;
2149 let posting_bytes = self.postings_handle.read_bytes_range_sync(range)?;
2150 results.push(BlockPostingList::deserialize_zero_copy(posting_bytes)?);
2151 }
2152 }
2153
2154 Ok(results)
2155 }
2156
2157 pub fn get_positions_sync(
2159 &self,
2160 field: Field,
2161 term: &[u8],
2162 ) -> Result<Option<crate::structures::PositionPostingList>> {
2163 let handle = match &self.positions_handle {
2164 Some(h) => h,
2165 None => return Ok(None),
2166 };
2167
2168 let mut key = Vec::with_capacity(4 + term.len());
2170 key.extend_from_slice(&field.0.to_le_bytes());
2171 key.extend_from_slice(term);
2172
2173 let term_info = match self.term_dict.get_sync(&key)? {
2175 Some(info) => info,
2176 None => return Ok(None),
2177 };
2178
2179 let (offset, length) = match term_info.position_info() {
2180 Some((o, l)) => (o, l),
2181 None => return Ok(None),
2182 };
2183
2184 let range = checked_file_range(offset, length, handle.len(), "position list")?;
2185 let slice = handle.slice(range);
2186 let data = slice.read_bytes_sync()?;
2187
2188 let pos_list = crate::structures::PositionPostingList::deserialize(data.as_slice())?;
2189 Ok(Some(pos_list))
2190 }
2191
2192 pub fn search_dense_vector_sync(
2195 &self,
2196 field: Field,
2197 query: &[f32],
2198 k: usize,
2199 nprobe: usize,
2200 rerank_factor: f32,
2201 combiner: crate::query::MultiValueCombiner,
2202 ) -> Result<Vec<VectorSearchResult>> {
2203 self.search_dense_vector_sync_impl(field, query, k, nprobe, rerank_factor, combiner, None)
2204 }
2205
2206 #[cfg(feature = "sync")]
2207 #[allow(clippy::too_many_arguments)]
2208 pub(crate) fn search_dense_vector_sync_with_probe_cache(
2209 &self,
2210 field: Field,
2211 query: &[f32],
2212 k: usize,
2213 nprobe: usize,
2214 rerank_factor: f32,
2215 combiner: crate::query::MultiValueCombiner,
2216 probe_cache: &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
2217 ) -> Result<Vec<VectorSearchResult>> {
2218 self.search_dense_vector_sync_impl(
2219 field,
2220 query,
2221 k,
2222 nprobe,
2223 rerank_factor,
2224 combiner,
2225 Some(probe_cache),
2226 )
2227 }
2228
2229 #[cfg(feature = "sync")]
2230 #[allow(clippy::too_many_arguments)]
2231 fn search_dense_vector_sync_impl(
2232 &self,
2233 field: Field,
2234 query: &[f32],
2235 k: usize,
2236 nprobe: usize,
2237 rerank_factor: f32,
2238 combiner: crate::query::MultiValueCombiner,
2239 probe_cache: Option<
2240 &std::sync::Mutex<Option<std::sync::Arc<crate::structures::IvfPqQueryPlan>>>,
2241 >,
2242 ) -> Result<Vec<VectorSearchResult>> {
2243 let params =
2244 self.validate_dense_search_request(field, query, nprobe, rerank_factor, combiner)?;
2245 let fetch_k = checked_dense_fetch_k(k, rerank_factor)?;
2246 if k == 0 {
2247 return Ok(Vec::new());
2248 }
2249
2250 let ann_index = self.vector_indexes.get(&field.0);
2251 let lazy_flat = self.flat_vectors.get(&field.0);
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::IvfPq(lazy) => {
2276 let index = lazy.get().ok_or_else(|| {
2277 Error::Schema("IVF-PQ index deserialization failed".to_string())
2278 })?;
2279 let codebook =
2280 self.trained_vectors
2281 .codebooks
2282 .get(&field.0)
2283 .ok_or_else(|| {
2284 Error::Schema(format!(
2285 "IVF-PQ index requires a global codebook for field {}",
2286 field.0
2287 ))
2288 })?;
2289 let centroids =
2290 self.trained_vectors
2291 .centroids
2292 .get(&field.0)
2293 .ok_or_else(|| {
2294 Error::Schema(format!(
2295 "IVF-PQ index requires coarse centroids for field {}",
2296 field.0
2297 ))
2298 })?;
2299 validate_coarse_centroids(centroids, params.dim)?;
2300 if index.config.dim != params.dim
2301 || codebook.config.dim != params.dim
2302 || index.config.code_size != codebook.config.num_subspaces
2303 || index.centroids_version != centroids.version
2304 || index.codebook_version != codebook.version
2305 || index
2306 .clusters
2307 .iter()
2308 .any(|(cluster_id, _)| *cluster_id >= centroids.num_clusters)
2309 {
2310 return Err(Error::Corruption(format!(
2311 "IVF-PQ index/codebook/centroid metadata does not match schema dimension {}",
2312 params.dim
2313 )));
2314 }
2315 let routing = self
2316 .schema
2317 .get_field_entry(field)
2318 .and_then(|entry| entry.dense_vector_config.as_ref())
2319 .map_or(crate::dsl::IvfRoutingMode::Auto, |config| {
2320 config.ivf_routing
2321 });
2322 let query_plan = float_query_plan(
2323 centroids,
2324 codebook,
2325 query,
2326 params.nprobe,
2327 routing,
2328 probe_cache,
2329 )?;
2330 let flat = lazy_flat.expect("ANN/flat pairing validated above");
2331 index
2332 .search_distinct_documents(
2333 fetch_k.min(flat.num_docs_with_vectors()),
2334 &query_plan,
2335 )
2336 .into_iter()
2337 .map(|(doc_id, ordinal, dist)| (doc_id, ordinal, 1.0 / (1.0 + dist)))
2338 .collect()
2339 }
2340 VectorIndex::BinaryIvf(_) => {
2341 Vec::new()
2343 }
2344 }
2345 } else if let Some(lazy_flat) = lazy_flat {
2346 let dim = lazy_flat.dim;
2348 let n = lazy_flat.num_vectors;
2349 let quant = lazy_flat.quantization;
2350 let batch_len =
2351 bounded_vector_score_batch(lazy_flat.vector_byte_size(), DENSE_SCORE_BATCH);
2352 let mut collector = FlatDocumentCollector::new(fetch_k.min(n), combiner);
2353 let mut scores = vec![0f32; batch_len];
2354
2355 for batch_start in (0..n).step_by(batch_len) {
2356 let batch_count = batch_len.min(n - batch_start);
2357 let batch_bytes = lazy_flat
2358 .read_vectors_batch_sync(batch_start, batch_count)
2359 .map_err(crate::Error::Io)?;
2360 let raw = batch_bytes.as_slice();
2361
2362 Self::score_quantized_batch(
2363 query,
2364 raw,
2365 quant,
2366 dim,
2367 &mut scores[..batch_count],
2368 params.unit_norm,
2369 )?;
2370
2371 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2372 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2373 collector.push(doc_id, ordinal, score);
2374 }
2375 }
2376
2377 return Ok(collector.into_results());
2378 } else {
2379 return Ok(Vec::new());
2380 };
2381
2382 if ann_index.is_some()
2384 && !results.is_empty()
2385 && let Some(lazy_flat) = lazy_flat
2386 {
2387 let dim = lazy_flat.dim;
2388 let quant = lazy_flat.quantization;
2389 let vbs = lazy_flat.vector_byte_size();
2390
2391 let (expanded, mut resolved) = expand_ann_candidate_documents(&results, lazy_flat)?;
2392 results = expanded;
2393
2394 if !resolved.is_empty() {
2395 resolved.sort_unstable_by_key(|&(_, flat_idx)| flat_idx);
2396 let batch_len = bounded_vector_score_batch(vbs, DENSE_SCORE_BATCH);
2397 let max_batch = batch_len.min(resolved.len());
2398 let max_raw_len = max_batch
2399 .checked_mul(vbs)
2400 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
2401 let mut raw_buf = vec![0u8; max_raw_len];
2402 let mut scores = vec![0f32; max_batch];
2403
2404 for chunk in resolved.chunks(batch_len) {
2405 let raw_len = chunk
2406 .len()
2407 .checked_mul(vbs)
2408 .ok_or_else(|| Error::Query("dense rerank buffer size overflow".into()))?;
2409 let raw = &mut raw_buf[..raw_len];
2410 for (buf_idx, &(_, flat_idx)) in chunk.iter().enumerate() {
2411 lazy_flat
2412 .read_vector_raw_into_sync(
2413 flat_idx,
2414 &mut raw[buf_idx * vbs..(buf_idx + 1) * vbs],
2415 )
2416 .map_err(crate::Error::Io)?;
2417 }
2418
2419 Self::score_quantized_batch(
2420 query,
2421 raw,
2422 quant,
2423 dim,
2424 &mut scores[..chunk.len()],
2425 params.unit_norm,
2426 )?;
2427
2428 for (buf_idx, &(ri, _)) in chunk.iter().enumerate() {
2429 results[ri].2 = scores[buf_idx];
2430 }
2431 }
2432 }
2433 }
2434
2435 Ok(combine_grouped_ordinal_results(results, combiner, k))
2436 }
2437
2438 #[cfg(feature = "sync")]
2443 fn search_binary_dense_vector_sync_impl(
2444 &self,
2445 field: Field,
2446 query: &[u8],
2447 k: usize,
2448 combiner: crate::query::MultiValueCombiner,
2449 probe_cache: Option<&std::sync::Mutex<Option<crate::structures::IvfProbePlan>>>,
2450 ) -> Result<Vec<VectorSearchResult>> {
2451 let schema_dim = self.validate_binary_search_request(field, query)?;
2452 combiner.validate().map_err(Error::Query)?;
2453 if k == 0 {
2454 return Ok(Vec::new());
2455 }
2456 let t0 = crate::observe::Timer::start();
2457 if let Some(VectorIndex::BinaryIvf(lazy)) = self.vector_indexes.get(&field.0) {
2458 let ivf = lazy.get().ok_or_else(|| {
2459 Error::Corruption(format!(
2460 "global binary IVF field {} could not be decoded",
2461 field.0
2462 ))
2463 })?;
2464 let config = self
2465 .schema
2466 .get_field_entry(field)
2467 .and_then(|entry| entry.binary_dense_vector_config.as_ref())
2468 .ok_or_else(|| {
2469 Error::Schema(format!(
2470 "binary IVF field {} has no schema configuration",
2471 field.0
2472 ))
2473 })?;
2474 let quantizer = self
2475 .trained_vectors
2476 .binary_quantizers
2477 .get(&field.0)
2478 .ok_or_else(|| {
2479 Error::Schema(format!(
2480 "global binary IVF field {} has no loaded quantizer",
2481 field.0
2482 ))
2483 })?;
2484 if ivf.dim_bits != schema_dim
2485 || quantizer.dim_bits != schema_dim
2486 || ivf.quantizer_version != quantizer.version
2487 || ivf.num_clusters != quantizer.num_clusters
2488 {
2489 return Err(Error::Corruption(format!(
2490 "global binary IVF field {} does not match its quantizer/schema generation",
2491 field.0
2492 )));
2493 }
2494 let flat = self.flat_vectors.get(&field.0).ok_or_else(|| {
2495 Error::Corruption(format!(
2496 "global binary IVF field {} is missing flat vector storage",
2497 field.0
2498 ))
2499 })?;
2500 let clusters = binary_probe_clusters(
2501 quantizer,
2502 query,
2503 config.nprobe,
2504 config.ivf_routing,
2505 probe_cache,
2506 )?;
2507 let candidate_docs = k.min(flat.num_docs_with_vectors());
2508 let ann_results = if flat.num_vectors == flat.num_docs_with_vectors() {
2509 ivf.search_in_clusters(query, candidate_docs, &clusters)
2510 } else {
2511 ivf.search_distinct_documents_in_clusters(query, candidate_docs, &clusters)
2512 };
2513 let results =
2514 exact_score_binary_candidate_documents_sync(&ann_results, flat, query, schema_dim)?;
2515 crate::observe::dense_l1(
2516 self.schema.index_label(),
2517 self.schema.get_field_name(field).unwrap_or("?"),
2518 "global_binary_ivf",
2519 t0.secs(),
2520 results.len(),
2521 );
2522 return Ok(combine_grouped_ordinal_results(results, combiner, k));
2523 }
2524 let lazy_flat = match self.flat_vectors.get(&field.0) {
2525 Some(f) => f,
2526 None => return Ok(Vec::new()),
2527 };
2528
2529 let dim_bits = lazy_flat.dim;
2530 let byte_len = lazy_flat.vector_byte_size();
2531 let n = lazy_flat.num_vectors;
2532
2533 if dim_bits != schema_dim {
2534 return Err(Error::Corruption(format!(
2535 "binary vector field {} has schema dimension {} but flat storage dimension {}",
2536 field.0, schema_dim, dim_bits
2537 )));
2538 }
2539
2540 if byte_len != query.len() {
2541 return Err(Error::Schema(format!(
2542 "Binary query vector byte length {} != field byte length {}",
2543 query.len(),
2544 byte_len
2545 )));
2546 }
2547
2548 let batch_len = bounded_vector_score_batch(byte_len, BINARY_SCORE_BATCH);
2549 let mut collector = FlatDocumentCollector::new(k, combiner);
2550 let mut scores = vec![0f32; batch_len];
2551
2552 for batch_start in (0..n).step_by(batch_len) {
2553 let batch_count = batch_len.min(n - batch_start);
2554 let batch_bytes = lazy_flat
2555 .read_vectors_batch_sync(batch_start, batch_count)
2556 .map_err(crate::Error::Io)?;
2557 let raw = batch_bytes.as_slice();
2558
2559 crate::structures::simd::batch_hamming_scores(
2560 query,
2561 raw,
2562 byte_len,
2563 dim_bits,
2564 &mut scores[..batch_count],
2565 );
2566
2567 for (i, &score) in scores.iter().enumerate().take(batch_count) {
2568 let (doc_id, ordinal) = lazy_flat.get_doc_id(batch_start + i);
2569 collector.push(doc_id, ordinal, score);
2570 }
2571 }
2572
2573 let results = collector.into_results();
2574
2575 crate::observe::dense_l1(
2576 self.schema.index_label(),
2577 self.schema.get_field_name(field).unwrap_or("?"),
2578 "binary_flat",
2579 t0.secs(),
2580 results.len(),
2581 );
2582 Ok(results)
2583 }
2584
2585 #[cfg(feature = "sync")]
2586 pub fn search_binary_dense_vector_sync(
2587 &self,
2588 field: Field,
2589 query: &[u8],
2590 k: usize,
2591 combiner: crate::query::MultiValueCombiner,
2592 ) -> Result<Vec<VectorSearchResult>> {
2593 self.search_binary_dense_vector_sync_impl(field, query, k, combiner, None)
2594 }
2595
2596 #[cfg(feature = "sync")]
2597 pub(crate) fn search_binary_dense_vector_sync_with_probe_cache(
2598 &self,
2599 field: Field,
2600 query: &[u8],
2601 k: usize,
2602 combiner: crate::query::MultiValueCombiner,
2603 probe_cache: &std::sync::Mutex<Option<crate::structures::IvfProbePlan>>,
2604 ) -> Result<Vec<VectorSearchResult>> {
2605 self.search_binary_dense_vector_sync_impl(field, query, k, combiner, Some(probe_cache))
2606 }
2607}
2608
2609#[cfg(test)]
2610mod dense_search_safety_tests {
2611 use super::*;
2612
2613 #[test]
2614 fn dense_fetch_count_rejects_non_finite_and_unbounded_factors() {
2615 for factor in [
2616 f32::NAN,
2617 f32::INFINITY,
2618 f32::NEG_INFINITY,
2619 0.0,
2620 0.5,
2621 MAX_DENSE_RERANK_FACTOR + 1.0,
2622 ] {
2623 assert!(
2624 checked_dense_fetch_k(10, factor).is_err(),
2625 "factor={factor}"
2626 );
2627 }
2628 }
2629
2630 #[test]
2631 fn flat_document_collector_does_not_let_one_multivalue_doc_crowd_out_others() {
2632 let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
2633 collector.push(1, 0, 1.0);
2634 collector.push(1, 1, 0.9);
2635 collector.push(2, 0, 0.8);
2636
2637 let results = collector.into_results();
2638 assert_eq!(
2639 results
2640 .iter()
2641 .map(|result| result.doc_id)
2642 .collect::<Vec<_>>(),
2643 vec![1, 2]
2644 );
2645 assert_eq!(results[0].ordinals.len(), 2);
2646 }
2647
2648 #[test]
2649 fn flat_document_collector_evicts_by_score_then_doc_id() {
2650 let mut collector = FlatDocumentCollector::new(2, crate::query::MultiValueCombiner::Max);
2651 collector.push(1, 0, 0.5);
2652 collector.push(3, 0, 0.8);
2653 collector.push(2, 0, 0.9);
2654 let results = collector.into_results();
2655 assert_eq!(
2656 results
2657 .iter()
2658 .map(|result| result.doc_id)
2659 .collect::<Vec<_>>(),
2660 vec![2, 3]
2661 );
2662
2663 let mut tied = FlatDocumentCollector::new(1, crate::query::MultiValueCombiner::Max);
2664 tied.push(2, 0, 1.0);
2665 tied.push(1, 0, 1.0);
2666 let results = tied.into_results();
2667 assert_eq!(results[0].doc_id, 1);
2668 }
2669
2670 #[test]
2671 fn dense_fetch_count_rounds_up_and_detects_overflow() {
2672 assert_eq!(checked_dense_fetch_k(3, 1.5).unwrap(), 5);
2673 assert_eq!(checked_dense_fetch_k(5_000, 3.0).unwrap(), 15_000);
2674 assert!(checked_dense_fetch_k(50_000, 3.0).is_err());
2675 assert!(checked_dense_fetch_k(usize::MAX, 2.0).is_err());
2676 }
2677
2678 #[test]
2679 fn file_ranges_reject_overflow_and_truncation() {
2680 assert_eq!(checked_file_range(4, 3, 7, "test").unwrap(), 4..7);
2681 assert!(checked_file_range(u64::MAX, 1, u64::MAX, "test").is_err());
2682 assert!(checked_file_range(5, 3, 7, "test").is_err());
2683 }
2684
2685 #[test]
2689 fn probe_scored_candidates_skip_flat_rereads() {
2690 let ann_results: Vec<RawVectorCandidate> = vec![(7, 0, 0.75), (9, 2, 0.5)];
2691 let mut expanded: Vec<RawVectorCandidate> =
2693 vec![(7, 0, 0.0), (7, 1, 0.0), (9, 0, 0.0), (9, 2, 0.0)];
2694 let resolved: Vec<ResolvedVectorCandidate> = vec![(0, 10), (1, 11), (2, 20), (3, 22)];
2695
2696 let remaining = fill_probe_scores_and_prune_resolved(&ann_results, &mut expanded, resolved);
2697
2698 assert_eq!(expanded[0].2, 0.75);
2700 assert_eq!(expanded[3].2, 0.5);
2701 assert_eq!(remaining, vec![(1, 11), (2, 20)]);
2703 assert_eq!(expanded[1].2, 0.0);
2704 assert_eq!(expanded[2].2, 0.0);
2705 }
2706}